From 3449ad2441444ed7cccbba162346920200293dd1 Mon Sep 17 00:00:00 2001 From: TheGreatAlgo <37487508+TheGreatAlgo@users.noreply.github.com> Date: Mon, 27 Jul 2026 09:16:37 -0400 Subject: [PATCH 1/5] fix: multigateway --- py_hamt/__init__.py | 8 +- py_hamt/store_httpx.py | 447 ++++++++++++++++---- tests/test_k15_multi_gateway_failover.py | 505 +++++++++++++++++++++++ 3 files changed, 869 insertions(+), 91 deletions(-) create mode 100644 tests/test_k15_multi_gateway_failover.py diff --git a/py_hamt/__init__.py b/py_hamt/__init__.py index 8bb5d59..6ba81bf 100644 --- a/py_hamt/__init__.py +++ b/py_hamt/__init__.py @@ -6,13 +6,19 @@ ShardedZarrV1DeprecationWarning, ShardReadMode, ) -from .store_httpx import ContentAddressedStore, InMemoryCAS, KuboCAS +from .store_httpx import ( + ContentAddressedStore, + GatewayContentMismatch, + InMemoryCAS, + KuboCAS, +) from .zarr_hamt_store import ZarrHAMTStore __all__ = [ "blake3_hashfn", "HAMT", "ContentAddressedStore", + "GatewayContentMismatch", "InMemoryCAS", "KuboCAS", "ZarrHAMTStore", diff --git a/py_hamt/store_httpx.py b/py_hamt/store_httpx.py index ae0e950..90add2b 100644 --- a/py_hamt/store_httpx.py +++ b/py_hamt/store_httpx.py @@ -24,6 +24,67 @@ # a request sleep unbounded (e.g. ``Retry-After: inf`` or a far-future date). _MAX_RETRY_AFTER_SECONDS = 300.0 +# Consecutive failures before a gateway is considered unhealthy and moved to the +# back of the rotation, and how long it stays there before being retried. A +# tripped gateway is never permanently removed: after the cooldown it is probed +# again by ordinary traffic, so a gateway that recovers rejoins on its own. +_GATEWAY_FAILURE_THRESHOLD = 3 +_GATEWAY_COOLDOWN_SECONDS = 30.0 + + +def _normalize_gateway_base_url(gateway_base_url: str) -> str: + """Normalize a gateway base URL to a ``.../ipfs/`` prefix. + + Accepts a bare host (``https://example.com``), an explicit path + (``https://example.com/ipfs``), and either with a trailing slash. + """ + gateway_base_url = gateway_base_url.rstrip("/") + if not gateway_base_url.endswith("/ipfs"): + gateway_base_url = f"{gateway_base_url}/ipfs" + return f"{gateway_base_url}/" + + +class _GatewayHealth: + """Consecutive-failure circuit breaker for a single gateway. + + Gateways are never removed from the rotation permanently. Once + ``_GATEWAY_FAILURE_THRESHOLD`` consecutive failures trip the breaker, the + gateway is deprioritized (tried only after every healthy gateway) until + ``_GATEWAY_COOLDOWN_SECONDS`` elapse, after which ordinary traffic probes it + again. Any success resets the counter. + + The event-loop clock is read lazily so instances remain constructible + outside async context; a gateway that has never been used is healthy. + """ + + __slots__ = ("consecutive_failures", "tripped_at") + + def __init__(self) -> None: + self.consecutive_failures: int = 0 + self.tripped_at: float | None = None + + def record_success(self) -> None: + self.consecutive_failures = 0 + self.tripped_at = None + + def record_failure(self, now: float) -> None: + self.consecutive_failures += 1 + if self.consecutive_failures >= _GATEWAY_FAILURE_THRESHOLD: + self.tripped_at = now + + def is_healthy(self, now: float) -> bool: + if self.tripped_at is None: + return True + if now - self.tripped_at >= _GATEWAY_COOLDOWN_SECONDS: + # Cooldown elapsed. Clear the trip so a single probe failure does + # not immediately re-trip on a stale counter, but keep the gateway + # on probation by leaving the failure count one short of the + # threshold: one more failure re-trips it right away. + self.tripped_at = None + self.consecutive_failures = _GATEWAY_FAILURE_THRESHOLD - 1 + return True + return False + def _retry_delay( initial_delay: float, @@ -55,6 +116,67 @@ def _retry_delay( return backoff_delay + jitter +class _LoadStats: + """Mutable trace counters shared across the gateways one load attempts.""" + + __slots__ = ("response_bytes", "retries", "status") + + def __init__(self) -> None: + self.response_bytes: int = 0 + self.retries: int = 0 + self.status: str = "ok" + + +class GatewayContentMismatch(Exception): + """A gateway returned bytes that do not hash to the requested CID. + + Raised only when content verification is enabled. Treated as a per-gateway + failure, so a multi-gateway ``KuboCAS`` fails over to the next gateway + rather than returning corrupt data to the caller. + """ + + +def _cid_is_verifiable(cid: CID, offset: Optional[int], suffix: Optional[int]) -> bool: + """Whether a gateway response for ``cid`` can be checked against its digest. + + Verification requires hashing the *complete* block, which holds only for: + + * **Full-body reads.** A Range request yields a slice, which does not hash + to the CID. + * **Non-``dag-pb`` CIDs.** A gateway serving a ``dag-pb`` CID returns the + reassembled UnixFS file, not the encoded block the CID commits to, so the + digest legitimately differs. ``raw`` and ``dag-cbor`` blocks -- what the + HAMT itself stores -- are returned verbatim and do hash correctly. + """ + if offset is not None or suffix is not None: + return False + if cid.codec.code == KuboCAS.DAG_PB_MARKER: + return False + # An identity multihash inlines its content in the CID; nothing was fetched. + return cid.hashfun.name != "identity" + + +def _verify_cid_content(cid: CID, data: bytes) -> None: + """Raise ``GatewayContentMismatch`` if ``data`` does not hash to ``cid``. + + A hash function the local ``multiformats`` build cannot compute is not a + mismatch; we cannot prove the content wrong, so the read is allowed through. + """ + try: + computed = multihash.digest(data, cid.hashfun.name) + except Exception: # pragma: no cover - depends on multiformats build + logger.debug( + "Cannot verify CID %s: unsupported hash function %s", + cid, + cid.hashfun.name, + ) + return + if computed != cid.digest: + raise GatewayContentMismatch( + f"gateway returned {len(data)} bytes that do not hash to {cid}" + ) + + def _slice_requested_range( data: bytes, offset: Optional[int], @@ -333,6 +455,19 @@ class KuboCAS(ContentAddressedStore): - **rpc_base_url / gateway_base_url** (str | None): override daemon endpoints (defaults match the local daemon ports). Gateway URLs may end with `/ipfs` and may include a trailing slash. + - **gateway_base_urls** (list[str] | None): read from several gateways with + automatic failover. Mutually exclusive with `gateway_base_url`. Each read + tries one gateway at a time, healthy gateways first, until one succeeds; + requests are not raced in parallel. A gateway that fails three times in a + row is moved to the back of the rotation for 30 seconds and then probed + again. `concurrency` applies per gateway. If every gateway fails, an + `ExceptionGroup` of the underlying errors is raised. + - **verify_content** (bool): check that returned bytes hash to the + requested CID, raising `GatewayContentMismatch` (and failing over to the + next gateway) when they do not. Worth enabling when reading from public + gateways you do not control. Only full-body reads of non-`dag-pb` CIDs + can be verified; Range reads and `dag-pb` reads are passed through + unchecked because neither returns the exact bytes the CID commits to. - **chunker** (str): chunking algorithm specification for Kubo's `add` RPC. Accepted formats are `"size-"`, `"rabin"`, or `"rabin---"`. @@ -355,6 +490,8 @@ def __init__( gateway_base_url: str | None = None, concurrency: int = 32, *, + gateway_base_urls: list[str] | None = None, + verify_content: bool = False, client_factory: Optional[Callable[[], httpx.AsyncClient]] = None, headers: dict[str, str] | None = None, auth: Tuple[str, str] | None = None, @@ -430,6 +567,13 @@ def __init__( self._semaphore_per_loop: Dict[ asyncio.AbstractEventLoop, asyncio.Semaphore ] = {} + # Gateway reads get a semaphore per (loop, gateway) so ``concurrency`` + # means "in-flight requests per gateway". Sharing one budget across + # gateways would divide effective parallelism by the gateway count and + # let a slow gateway starve the healthy ones of slots. + self._gateway_semaphore_per_loop: Dict[ + Tuple[asyncio.AbstractEventLoop, str], asyncio.Semaphore + ] = {} # Now, perform validation that might raise an exception chunker_pattern = r"(?:size-[1-9]\d*|rabin(?:-[1-9]\d*-[1-9]\d*-[1-9]\d*)?)" @@ -442,20 +586,39 @@ def __init__( if rpc_base_url is None: rpc_base_url = KuboCAS.KUBO_DEFAULT_LOCAL_RPC_BASE_URL # pragma - if gateway_base_url is None: - gateway_base_url = KuboCAS.KUBO_DEFAULT_LOCAL_GATEWAY_BASE_URL - gateway_base_url = gateway_base_url.rstrip("/") - if not gateway_base_url.endswith("/ipfs"): - gateway_base_url = f"{gateway_base_url}/ipfs" - gateway_base_url = f"{gateway_base_url}/" + if gateway_base_urls is not None: + if gateway_base_url is not None: + raise ValueError( + "gateway_base_url and gateway_base_urls are mutually " + "exclusive; pass every gateway in gateway_base_urls" + ) + if not gateway_base_urls: + raise ValueError("gateway_base_urls must not be empty") + normalized = [_normalize_gateway_base_url(url) for url in gateway_base_urls] + # Preserve caller order while dropping duplicates: a repeated + # gateway would otherwise get several rotation slots and several + # independent concurrency budgets pointed at one host. + self.gateway_base_urls: list[str] = list(dict.fromkeys(normalized)) + else: + if gateway_base_url is None: + gateway_base_url = KuboCAS.KUBO_DEFAULT_LOCAL_GATEWAY_BASE_URL + self.gateway_base_urls = [_normalize_gateway_base_url(gateway_base_url)] pin_string: str = "true" if pin_on_add else "false" self.rpc_url: str = f"{rpc_base_url}/api/v0/add?hash={self.hasher}&chunker={self.chunker}&pin={pin_string}" """@private""" - self.gateway_base_url: str = gateway_base_url + self.gateway_base_url: str = self.gateway_base_urls[0] """@private""" + self.verify_content: bool = verify_content + """@private""" + # Health is per gateway but shared across event loops: a gateway that is + # rate-limiting or down is doing so regardless of which loop observed it. + self._gateway_health: Dict[str, _GatewayHealth] = { + url: _GatewayHealth() for url in self.gateway_base_urls + } + if client is not None: # Bind the user-supplied client lazily on first async use. self._owns_client = False @@ -536,6 +699,7 @@ def _loop_semaphore(self) -> asyncio.Semaphore: self._client_per_loop = {} self._internally_created_clients = set() self._semaphore_per_loop = {} + self._gateway_semaphore_per_loop = {} loop: asyncio.AbstractEventLoop = asyncio.get_running_loop() try: @@ -545,6 +709,49 @@ def _loop_semaphore(self) -> asyncio.Semaphore: self._semaphore_per_loop[loop] = semaphore return semaphore + def _gateway_semaphore(self, gateway_base_url: str) -> asyncio.Semaphore: + """Get or create the concurrency semaphore for one gateway on this loop. + + With a single gateway this is equivalent to ``_loop_semaphore``; with + several it keeps each gateway's ``concurrency`` budget independent. + """ + loop: asyncio.AbstractEventLoop = asyncio.get_running_loop() + key = (loop, gateway_base_url) + try: + return self._gateway_semaphore_per_loop[key] + except KeyError: + semaphore = asyncio.Semaphore(self._concurrency) + self._gateway_semaphore_per_loop[key] = semaphore + return semaphore + + def _ordered_gateways(self) -> list[str]: + """Gateways to try, best first. + + Ordered by health, then by whether a concurrency slot is free right now. + The second key avoids head-of-line blocking: attempts wait on their + gateway's semaphore, so without it a read queued behind a saturated + gateway would stall even when an idle gateway could serve it + immediately -- precisely the case multiple gateways exist to handle. + + Deprioritizing rather than dropping unhealthy gateways means a run where + every gateway has tripped still attempts them all instead of failing + with nothing tried. + """ + if len(self.gateway_base_urls) == 1: + return self.gateway_base_urls + + now = asyncio.get_running_loop().time() + # is_healthy() clears an expired trip, so evaluate it exactly once per + # gateway rather than inside a sort key, which may call it repeatedly. + ranks: Dict[str, tuple[int, int]] = {} + for url in self.gateway_base_urls: + unhealthy = 0 if self._gateway_health[url].is_healthy(now) else 1 + busy = 1 if self._gateway_semaphore(url).locked() else 0 + ranks[url] = (unhealthy, busy) + + # Stable sort, so configured order breaks ties within a rank. + return sorted(self.gateway_base_urls, key=lambda url: ranks[url]) + def _loop_client(self) -> httpx.AsyncClient: """Get or create a client for the current event loop. @@ -563,6 +770,7 @@ def _loop_client(self) -> httpx.AsyncClient: self._closed = False self._client_per_loop = {} self._semaphore_per_loop = {} + self._gateway_semaphore_per_loop = {} loop: asyncio.AbstractEventLoop = asyncio.get_running_loop() try: @@ -678,6 +886,7 @@ async def aclose(self) -> None: self._client_per_loop.clear() self._internally_created_clients.clear() self._semaphore_per_loop.clear() + self._gateway_semaphore_per_loop.clear() self._closed = True # At this point, _client_per_loop should be empty or only contain @@ -710,6 +919,7 @@ def __del__(self) -> None: # so just clear the references self._client_per_loop.clear() self._semaphore_per_loop.clear() + self._gateway_semaphore_per_loop.clear() self._closed = True return @@ -732,6 +942,7 @@ def __del__(self) -> None: if hasattr(self, "_client_per_loop"): self._client_per_loop.clear() self._semaphore_per_loop.clear() + self._gateway_semaphore_per_loop.clear() self._closed = True # --------------------------------------------------------------------- # @@ -785,6 +996,102 @@ async def save(self, data: bytes, codec: ContentAddressedStore.CodecInput) -> CI ) raise RuntimeError("Exited the retry loop unexpectedly.") # pragma: no cover + async def _load_from_gateway( + self, + gateway_base_url: str, + cid: CID, + headers: Dict[str, str], + offset: Optional[int], + length: Optional[int], + suffix: Optional[int], + stats: "_LoadStats", + ) -> bytes: + """Fetch ``cid`` from one gateway, retrying that gateway's transients. + + Raises on failure so the caller can fail over. ``stats`` accumulates the + byte count and retry total across every gateway attempted, so the trace + emitted by ``load`` reflects the whole operation rather than the last leg. + """ + url = f"{gateway_base_url}{cid}" + client = self._loop_client() + semaphore = self._gateway_semaphore(gateway_base_url) + retry_count = 0 + + while retry_count <= self.max_retries: + try: + async with semaphore: # Throttle each gateway attempt + response = await client.get(url, headers=headers or None) + # An unsatisfiable range is answered with 416 by a compliant + # gateway; return b"" to match Python-slice semantics (and + # InMemoryCAS) instead of raising. + if ( + response.status_code == httpx.codes.REQUESTED_RANGE_NOT_SATISFIABLE + and _range_not_satisfiable_is_empty(response, offset, suffix) + ): + return b"" + response.raise_for_status() + content = response.content + stats.response_bytes = len(content) + if headers: + if response.status_code == httpx.codes.OK: + logger.debug( + "Gateway ignored Range request for CID %s; " + "slicing the complete response locally", + cid, + ) + return _slice_requested_range(content, offset, length, suffix) + if response.status_code == httpx.codes.PARTIAL_CONTENT: + # Trust the partial body only after proving its + # Content-Range matches the requested byte window. + _validate_partial_content( + response, offset, length, suffix, stats.response_bytes + ) + return content + # Any other 2xx to a Range request is unexpected: we + # cannot know which bytes it carries, so fail rather than + # return a possibly-wrong window. + raise httpx.HTTPStatusError( + f"unexpected {response.status_code} response to a " + "Range request", + request=response.request, + response=response, + ) + if self.verify_content and _cid_is_verifiable(cid, offset, suffix): + # A mismatch means this gateway served wrong bytes. Raising + # here routes it through the caller's failover path like any + # other per-gateway failure. + _verify_cid_content(cid, content) + return content + + except httpx.RequestError: + if retry_count >= self.max_retries: + stats.status = "request_error" + raise + retry_count += 1 + stats.retries += 1 + await asyncio.sleep( + _retry_delay(self.initial_delay, self.backoff_factor, retry_count) + ) + + except httpx.HTTPStatusError as error: + if ( + error.response.status_code not in _RETRYABLE_STATUS_CODES + or retry_count >= self.max_retries + ): + stats.status = "http_error" + raise + retry_count += 1 + stats.retries += 1 + await asyncio.sleep( + _retry_delay( + self.initial_delay, + self.backoff_factor, + retry_count, + error.response, + ) + ) + raise RuntimeError("Exited the retry loop unexpectedly.") # pragma: no cover + async def load( self, id: IPLDKind, @@ -800,12 +1107,20 @@ async def load( other HTTP errors fail immediately. Zero-length and zero-suffix reads return immediately without a gateway request. Concurrency slots are held per HTTP attempt and released during retry backoff. + + When several gateways are configured, each is tried in turn -- healthy + ones first -- until one succeeds. Requests are *not* raced in parallel: + fanning every read out to every gateway would multiply egress and burn + each gateway's rate-limit budget N times over, which is the opposite of + what helps when rate limiting is the problem being solved. A gateway + that fails ``_GATEWAY_FAILURE_THRESHOLD`` times consecutively is moved + to the back of the rotation for a cooldown. If every gateway fails, the + collected errors are raised together as an ``ExceptionGroup``. """ if (offset is not None and length == 0) or (offset is None and suffix == 0): return b"" cid = cast(CID, id) - url: str = f"{self.gateway_base_url + str(cid)}" headers: Dict[str, str] = {} # Construct the Range header if required @@ -823,97 +1138,49 @@ async def load( headers["Range"] = f"bytes=-{suffix}" trace_started_at = instrumentation.begin_cas_load(cid, bool(headers)) - response_bytes = 0 - final_status = "ok" - final_retry_count = 0 + stats = _LoadStats() + gateways = self._ordered_gateways() + failures: list[Exception] = [] try: - client = self._loop_client() - semaphore = self._loop_semaphore() - retry_count = 0 - - while retry_count <= self.max_retries: + for gateway_base_url in gateways: + health = self._gateway_health[gateway_base_url] try: - async with semaphore: # Throttle each gateway attempt - response = await client.get(url, headers=headers or None) - # An unsatisfiable range is answered with 416 by a compliant - # gateway; return b"" to match Python-slice semantics (and - # InMemoryCAS) instead of raising. - if ( - response.status_code - == httpx.codes.REQUESTED_RANGE_NOT_SATISFIABLE - and _range_not_satisfiable_is_empty(response, offset, suffix) - ): - final_retry_count = retry_count - return b"" - response.raise_for_status() - content = response.content - response_bytes = len(content) - final_retry_count = retry_count - if headers: - if response.status_code == httpx.codes.OK: - logger.debug( - "Gateway ignored Range request for CID %s; " - "slicing the complete response locally", - cid, - ) - return _slice_requested_range( - content, offset, length, suffix - ) - if response.status_code == httpx.codes.PARTIAL_CONTENT: - # Trust the partial body only after proving its - # Content-Range matches the requested byte window. - _validate_partial_content( - response, offset, length, suffix, response_bytes - ) - return content - # Any other 2xx to a Range request is unexpected: we - # cannot know which bytes it carries, so fail rather than - # return a possibly-wrong window. - raise httpx.HTTPStatusError( - f"unexpected {response.status_code} response to a " - "Range request", - request=response.request, - response=response, + content = await self._load_from_gateway( + gateway_base_url, cid, headers, offset, length, suffix, stats + ) + except (httpx.HTTPError, GatewayContentMismatch) as error: + health.record_failure(asyncio.get_running_loop().time()) + failures.append(error) + if len(gateways) > 1: + logger.debug( + "Gateway %s failed for CID %s (%s); trying the next one", + gateway_base_url, + cid, + error, ) + continue + else: + health.record_success() + # A gateway leg may have set a failure status before a later + # gateway succeeded; the operation as a whole is a success. + stats.status = "ok" return content - except httpx.RequestError: - if retry_count >= self.max_retries: - final_status = "request_error" - final_retry_count = retry_count - raise - retry_count += 1 - await asyncio.sleep( - _retry_delay( - self.initial_delay, self.backoff_factor, retry_count - ) - ) - - except httpx.HTTPStatusError as error: - if ( - error.response.status_code not in _RETRYABLE_STATUS_CODES - or retry_count >= self.max_retries - ): - final_status = "http_error" - final_retry_count = retry_count - raise - retry_count += 1 - await asyncio.sleep( - _retry_delay( - self.initial_delay, - self.backoff_factor, - retry_count, - error.response, - ) - ) + # Every gateway failed. With one configured, re-raise its error + # unchanged so existing single-gateway callers keep seeing the exact + # httpx exception type they handle today. + if len(failures) == 1: + raise failures[0] + raise ExceptionGroup( + f"all {len(gateways)} gateways failed for CID {cid}", failures + ) finally: instrumentation.end_cas_load( trace_started_at, - byte_count=response_bytes, - retries=final_retry_count, - status=final_status, + byte_count=stats.response_bytes, + retries=stats.retries, + status=stats.status, ) - raise RuntimeError("Exited the retry loop unexpectedly.") # pragma: no cover # --------------------------------------------------------------------- # # pin_cid() - method to pin a CID # diff --git a/tests/test_k15_multi_gateway_failover.py b/tests/test_k15_multi_gateway_failover.py new file mode 100644 index 0000000..5e1c019 --- /dev/null +++ b/tests/test_k15_multi_gateway_failover.py @@ -0,0 +1,505 @@ +"""Multi-gateway failover, per-gateway concurrency, and content verification.""" + +import asyncio +import threading +from collections.abc import Iterator +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Callable + +import httpx +import pytest +from multiformats import CID, multihash + +from py_hamt import GatewayContentMismatch, KuboCAS +from py_hamt import store_httpx as store_module + +BODY = b"multi gateway body" +# raw + sha2-256 CID over BODY, so verify_content can check it for real. +GOOD_CID = CID("base32", 1, "raw", multihash.digest(BODY, "sha2-256")) + + +@dataclass +class FakeGateway: + """A stub IPFS gateway recording every request it serves.""" + + url: str + hits: list[str] = field(default_factory=list) + # Set by tests to control responses; returns (status, body). + responder: Callable[[str], tuple[int, bytes]] = lambda _cid: (200, BODY) + max_concurrent: int = 0 + _inflight: int = 0 + _lock: threading.Lock = field(default_factory=threading.Lock) + + +def _serve(gateway_holder: list[FakeGateway]) -> Iterator[FakeGateway]: + """Run a threaded HTTP server backed by ``gateway_holder[0]``.""" + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, fmt: str, *args: object) -> None: + pass + + def do_GET(self) -> None: + gw = gateway_holder[0] + cid = self.path.rsplit("/", 1)[-1].split("?", 1)[0] + with gw._lock: + gw.hits.append(cid) + gw._inflight += 1 + gw.max_concurrent = max(gw.max_concurrent, gw._inflight) + try: + status, body = gw.responder(cid) + finally: + with gw._lock: + gw._inflight -= 1 + self.send_response(status) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if body: + self.wfile.write(body) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + gateway_holder[0].url = f"http://127.0.0.1:{server.server_address[1]}" + try: + yield gateway_holder[0] + finally: + server.shutdown() + server.server_close() + thread.join() + + +@pytest.fixture +def gateway_a() -> Iterator[FakeGateway]: + yield from _serve([FakeGateway(url="")]) + + +@pytest.fixture +def gateway_b() -> Iterator[FakeGateway]: + yield from _serve([FakeGateway(url="")]) + + +@pytest.fixture +def gateway_c() -> Iterator[FakeGateway]: + yield from _serve([FakeGateway(url="")]) + + +def make_cas(*gateways: FakeGateway, **kwargs: object) -> KuboCAS: + return KuboCAS( + gateway_base_urls=[gw.url for gw in gateways], + rpc_base_url=gateways[0].url, + initial_delay=0.01, + **kwargs, # type: ignore[arg-type] + ) + + +# --------------------------------------------------------------------------- # +# construction and normalization # +# --------------------------------------------------------------------------- # + + +def test_gateway_base_urls_normalize_and_expose_first() -> None: + cas = KuboCAS( + gateway_base_urls=[ + "https://a.example.com", + "https://b.example.com/ipfs", + "https://c.example.com/ipfs/", + ], + rpc_base_url="https://a.example.com", + ) + assert cas.gateway_base_urls == [ + "https://a.example.com/ipfs/", + "https://b.example.com/ipfs/", + "https://c.example.com/ipfs/", + ] + # gateway_base_url stays meaningful for single-gateway callers and docs. + assert cas.gateway_base_url == "https://a.example.com/ipfs/" + + +def test_gateway_base_urls_deduplicates_after_normalization() -> None: + cas = KuboCAS( + gateway_base_urls=[ + "https://a.example.com", + "https://a.example.com/ipfs/", + "https://b.example.com", + ], + rpc_base_url="https://a.example.com", + ) + assert cas.gateway_base_urls == [ + "https://a.example.com/ipfs/", + "https://b.example.com/ipfs/", + ] + + +def test_gateway_base_url_and_urls_are_mutually_exclusive() -> None: + with pytest.raises(ValueError, match="mutually exclusive"): + KuboCAS( + gateway_base_url="https://a.example.com", + gateway_base_urls=["https://b.example.com"], + ) + + +def test_empty_gateway_base_urls_rejected() -> None: + with pytest.raises(ValueError, match="must not be empty"): + KuboCAS(gateway_base_urls=[]) + + +def test_single_gateway_default_is_unchanged() -> None: + cas = KuboCAS(rpc_base_url="https://a.example.com") + assert cas.gateway_base_urls == [ + f"{KuboCAS.KUBO_DEFAULT_LOCAL_GATEWAY_BASE_URL}/ipfs/" + ] + + +# --------------------------------------------------------------------------- # +# failover # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_failover_to_second_gateway_on_server_error( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + gateway_a.responder = lambda _cid: (500, b"") + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert gateway_a.hits, "the first gateway should have been tried" + assert gateway_b.hits, "the second gateway should have served the read" + + +@pytest.mark.asyncio +async def test_first_healthy_gateway_wins_and_others_are_untouched( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """Reads are sequential, not raced: a success must not fan out.""" + async with make_cas(gateway_a, gateway_b) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert len(gateway_a.hits) == 1 + assert gateway_b.hits == [] + + +@pytest.mark.asyncio +async def test_404_fails_over_rather_than_failing_the_read( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """A gateway that lacks the block is a reason to ask another one. + + 404 is non-retryable *on one gateway*, but across gateways it is exactly + the case failover exists for. + """ + gateway_a.responder = lambda _cid: (404, b"") + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert len(gateway_a.hits) == 1 + assert len(gateway_b.hits) == 1 + + +@pytest.mark.asyncio +async def test_all_gateways_failing_raises_exception_group( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + gateway_a.responder = lambda _cid: (500, b"") + gateway_b.responder = lambda _cid: (503, b"") + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + with pytest.raises(ExceptionGroup) as excinfo: + await cas.load(GOOD_CID) + + assert "all 2 gateways failed" in str(excinfo.value) + assert len(excinfo.value.exceptions) == 2 + # The underlying errors survive, rather than being flattened to a message. + assert all( + isinstance(exc, httpx.HTTPStatusError) for exc in excinfo.value.exceptions + ) + + +@pytest.mark.asyncio +async def test_single_gateway_still_raises_bare_httpx_error( + gateway_a: FakeGateway, +) -> None: + """One gateway must keep its pre-existing exception contract.""" + gateway_a.responder = lambda _cid: (500, b"") + + async with make_cas(gateway_a, max_retries=0) as cas: + with pytest.raises(httpx.HTTPStatusError): + await cas.load(GOOD_CID) + + +@pytest.mark.asyncio +async def test_retries_are_exhausted_per_gateway_before_failover( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + gateway_a.responder = lambda _cid: (503, b"") + + async with make_cas(gateway_a, gateway_b, max_retries=2) as cas: + assert await cas.load(GOOD_CID) == BODY + + # Initial attempt plus two retries against the failing gateway. + assert len(gateway_a.hits) == 3 + assert len(gateway_b.hits) == 1 + + +# --------------------------------------------------------------------------- # +# circuit breaker # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_unhealthy_gateway_is_deprioritized_then_probed_again( + gateway_a: FakeGateway, gateway_b: FakeGateway, monkeypatch: pytest.MonkeyPatch +) -> None: + gateway_a.responder = lambda _cid: (500, b"") + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): + assert await cas.load(GOOD_CID) == BODY + + tripped_hits = len(gateway_a.hits) + assert tripped_hits == store_module._GATEWAY_FAILURE_THRESHOLD + + # Now tripped: subsequent reads should skip straight to gateway B. + for _ in range(3): + assert await cas.load(GOOD_CID) == BODY + assert len(gateway_a.hits) == tripped_hits, ( + "tripped gateway still receiving traffic" + ) + + # After the cooldown it is probed again by ordinary traffic. + monkeypatch.setattr(store_module, "_GATEWAY_COOLDOWN_SECONDS", 0.0) + assert await cas.load(GOOD_CID) == BODY + assert len(gateway_a.hits) == tripped_hits + 1 + + +@pytest.mark.asyncio +async def test_recovered_gateway_resumes_priority( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """A success resets the failure counter, so a flap does not trip later.""" + responses = [(500, b""), (500, b"")] + + def flaky(_cid: str) -> tuple[int, bytes]: + return responses.pop(0) if responses else (200, BODY) + + gateway_a.responder = flaky + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + # Two failures: one short of the threshold, so A stays in front. + assert await cas.load(GOOD_CID) == BODY + assert await cas.load(GOOD_CID) == BODY + # A recovers and serves this one itself. + assert await cas.load(GOOD_CID) == BODY + health = cas._gateway_health[cas.gateway_base_urls[0]] + assert health.consecutive_failures == 0 + # A further failure does not immediately trip a stale counter. + assert await cas.load(GOOD_CID) == BODY + assert len(gateway_b.hits) == 2 + + +@pytest.mark.asyncio +async def test_all_gateways_unhealthy_still_attempts_every_one( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """Deprioritized never means dropped, or a total outage would try nothing.""" + gateway_a.responder = lambda _cid: (500, b"") + gateway_b.responder = lambda _cid: (500, b"") + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): + with pytest.raises(ExceptionGroup): + await cas.load(GOOD_CID) + + hits_before = (len(gateway_a.hits), len(gateway_b.hits)) + with pytest.raises(ExceptionGroup): + await cas.load(GOOD_CID) + + assert len(gateway_a.hits) == hits_before[0] + 1 + assert len(gateway_b.hits) == hits_before[1] + 1 + + +def test_health_cooldown_leaves_gateway_on_probation() -> None: + health = store_module._GatewayHealth() + for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): + health.record_failure(now=0.0) + + assert not health.is_healthy(now=1.0) + # Cooldown elapsed: eligible again, but one failure away from re-tripping. + assert health.is_healthy(now=store_module._GATEWAY_COOLDOWN_SECONDS + 1.0) + health.record_failure(now=store_module._GATEWAY_COOLDOWN_SECONDS + 1.0) + assert not health.is_healthy(now=store_module._GATEWAY_COOLDOWN_SECONDS + 1.0) + + +# --------------------------------------------------------------------------- # +# per-gateway concurrency # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_concurrency_budget_is_per_gateway( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """Each gateway gets its own ``concurrency`` slots, not a shared pool. + + Asserted structurally rather than by racing wall-clock: each gateway must + get a distinct semaphore carrying the full ``concurrency`` budget, so + saturating one cannot starve reads routed to another. Timing-based variants + of this check are dominated by connection-pool and handler-thread + scheduling rather than by the semaphores under test. + """ + async with make_cas(gateway_a, gateway_b, concurrency=2, max_retries=0) as cas: + url_a, url_b = cas.gateway_base_urls + + # Drive a real read through each gateway so both semaphores are the + # ones load() actually used, not ones conjured by the assertion. + gateway_a.responder = lambda _cid: (500, b"") + assert await cas.load(GOOD_CID) == BODY + assert gateway_a.hits and gateway_b.hits + + sem_a = cas._gateway_semaphore(url_a) + sem_b = cas._gateway_semaphore(url_b) + assert sem_a is not sem_b, "gateways must not share a concurrency budget" + + # Each carries the full budget, and exhausting A leaves B untouched. + await sem_a.acquire() + await sem_a.acquire() + assert sem_a.locked() + assert not sem_b.locked() + sem_a.release() + sem_a.release() + + +@pytest.mark.asyncio +async def test_gateway_semaphore_bounds_in_flight_requests( + gateway_a: FakeGateway, +) -> None: + import time + + def slow(_cid: str) -> tuple[int, bytes]: + time.sleep(0.05) + return 200, BODY + + gateway_a.responder = slow + + async with make_cas(gateway_a, concurrency=2) as cas: + await asyncio.gather(*(cas.load(GOOD_CID) for _ in range(6))) + + assert gateway_a.max_concurrent <= 2 + + +@pytest.mark.asyncio +async def test_saturated_gateway_does_not_block_an_idle_one( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """A busy gateway must not stall reads an idle gateway could serve. + + Attempts wait on their gateway's semaphore, so ordering strictly by + configured position would queue this read behind saturated gateway A even + though B is free -- the head-of-line stall that undermines the whole point + of configuring several gateways. + """ + async with make_cas(gateway_a, gateway_b, concurrency=1, max_retries=0) as cas: + sem_a = cas._gateway_semaphore(cas.gateway_base_urls[0]) + await sem_a.acquire() # gateway A has no free slot + try: + result = await asyncio.wait_for(cas.load(GOOD_CID), timeout=5.0) + finally: + sem_a.release() + + assert result == BODY + assert gateway_a.hits == [], "saturated gateway should have been skipped" + assert len(gateway_b.hits) == 1 + + +@pytest.mark.asyncio +async def test_gateway_semaphores_cleared_on_aclose(gateway_a: FakeGateway) -> None: + cas = make_cas(gateway_a) + await cas.load(GOOD_CID) + assert cas._gateway_semaphore_per_loop + await cas.aclose() + assert cas._gateway_semaphore_per_loop == {} + + +# --------------------------------------------------------------------------- # +# content verification # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_verification_rejects_wrong_bytes_and_fails_over( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """A gateway serving valid-looking but wrong content must not win.""" + gateway_a.responder = lambda _cid: (200, b"corrupted payload") + + async with make_cas( + gateway_a, gateway_b, verify_content=True, max_retries=0 + ) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert len(gateway_a.hits) == 1 + assert len(gateway_b.hits) == 1 + + +@pytest.mark.asyncio +async def test_verification_failure_on_sole_gateway_raises( + gateway_a: FakeGateway, +) -> None: + gateway_a.responder = lambda _cid: (200, b"corrupted payload") + + async with make_cas(gateway_a, verify_content=True, max_retries=0) as cas: + with pytest.raises(GatewayContentMismatch, match="do not hash to"): + await cas.load(GOOD_CID) + + +@pytest.mark.asyncio +async def test_verification_accepts_matching_bytes(gateway_a: FakeGateway) -> None: + async with make_cas(gateway_a, verify_content=True) as cas: + assert await cas.load(GOOD_CID) == BODY + + +@pytest.mark.asyncio +async def test_verification_off_by_default_passes_bad_bytes_through( + gateway_a: FakeGateway, +) -> None: + """Opt-in: the default path must not change behaviour or cost a hash.""" + gateway_a.responder = lambda _cid: (200, b"corrupted payload") + + async with make_cas(gateway_a) as cas: + assert await cas.load(GOOD_CID) == b"corrupted payload" + + +@pytest.mark.asyncio +async def test_range_reads_are_not_verified(gateway_a: FakeGateway) -> None: + """A Range response is a slice and cannot hash to the CID.""" + gateway_a.responder = lambda _cid: (200, BODY) + + async with make_cas(gateway_a, verify_content=True) as cas: + # The stub ignores Range and returns 200, so load slices locally. + assert await cas.load(GOOD_CID, offset=0, length=5) == BODY[:5] + + +def test_dag_pb_cids_are_not_verifiable() -> None: + """Gateways return reassembled UnixFS files for dag-pb, not the block.""" + dag_pb_cid = CID("base32", 1, "dag-pb", multihash.digest(BODY, "sha2-256")) + assert not store_module._cid_is_verifiable(dag_pb_cid, None, None) + assert store_module._cid_is_verifiable(GOOD_CID, None, None) + assert not store_module._cid_is_verifiable(GOOD_CID, 0, None) + assert not store_module._cid_is_verifiable(GOOD_CID, None, 4) + + +def test_unsupported_hash_function_is_not_a_mismatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """We cannot prove content wrong with a hash we cannot compute.""" + + def unsupported(*_args: object, **_kwargs: object) -> bytes: + raise KeyError("unsupported multihash") + + monkeypatch.setattr(store_module.multihash, "digest", unsupported) + store_module._verify_cid_content(GOOD_CID, b"anything at all") From 0b8255da31048be72fe6b8eb378b2d3c5eb96771 Mon Sep 17 00:00:00 2001 From: TheGreatAlgo <37487508+TheGreatAlgo@users.noreply.github.com> Date: Mon, 27 Jul 2026 09:51:34 -0400 Subject: [PATCH 2/5] fix: gateway failover --- py_hamt/store_httpx.py | 117 ++++++++-- tests/test_k15_multi_gateway_failover.py | 280 ++++++++++++++++++++++- 2 files changed, 379 insertions(+), 18 deletions(-) diff --git a/py_hamt/store_httpx.py b/py_hamt/store_httpx.py index 90add2b..3a5a832 100644 --- a/py_hamt/store_httpx.py +++ b/py_hamt/store_httpx.py @@ -3,11 +3,13 @@ import random import re import threading +import time import warnings from abc import ABC, abstractmethod from datetime import datetime, timezone from email.utils import parsedate_to_datetime from typing import Any, Callable, Dict, Literal, Optional, Tuple, cast +from urllib.parse import urlsplit import httpx from dag_cbor.ipld import IPLDKind @@ -31,6 +33,18 @@ _GATEWAY_FAILURE_THRESHOLD = 3 _GATEWAY_COOLDOWN_SECONDS = 30.0 +# Request headers that carry credentials. When a read falls over to a gateway +# on a different origin than the one the credentials were configured for, these +# are stripped: a token minted for a private gateway must not be handed to a +# public fallback simply because the private one returned 503. +_SENSITIVE_HEADERS = frozenset({"authorization", "proxy-authorization", "cookie"}) + + +def _origin_of(url: str) -> tuple[str, str, int | None]: + """Scheme/host/port triple used to decide if two URLs share an origin.""" + parsed = urlsplit(url) + return (parsed.scheme.lower(), (parsed.hostname or "").lower(), parsed.port) + def _normalize_gateway_base_url(gateway_base_url: str) -> str: """Normalize a gateway base URL to a ``.../ipfs/`` prefix. @@ -53,8 +67,11 @@ class _GatewayHealth: ``_GATEWAY_COOLDOWN_SECONDS`` elapse, after which ordinary traffic probes it again. Any success resets the counter. - The event-loop clock is read lazily so instances remain constructible - outside async context; a gateway that has never been used is healthy. + Times are ``time.monotonic()`` readings. That clock is process-wide, so a + trip recorded while one event loop is running stays comparable from another + -- unlike ``loop.time()``, whose epoch is only meaningful within a single + loop and would make the cooldown expire instantly or never once this shared + state is touched from a second loop. """ __slots__ = ("consecutive_failures", "tripped_at") @@ -150,28 +167,51 @@ def _cid_is_verifiable(cid: CID, offset: Optional[int], suffix: Optional[int]) - """ if offset is not None or suffix is not None: return False - if cid.codec.code == KuboCAS.DAG_PB_MARKER: - return False - # An identity multihash inlines its content in the CID; nothing was fetched. - return cid.hashfun.name != "identity" + # An identity multihash inlines the block in the CID, but load() still + # fetches and returns whatever the gateway sends, so the response is + # verifiable -- by direct comparison rather than rehashing. + return cid.codec.code != KuboCAS.DAG_PB_MARKER def _verify_cid_content(cid: CID, data: bytes) -> None: """Raise ``GatewayContentMismatch`` if ``data`` does not hash to ``cid``. - A hash function the local ``multiformats`` build cannot compute is not a - mismatch; we cannot prove the content wrong, so the read is allowed through. + An ``identity`` multihash inlines the block in the CID itself, so the bytes + are compared directly rather than rehashed. + + The digest is computed at the CID's own digest length. That is required for + correctness in both directions: variable-output functions (blake3, the + default hasher here) *reject* a call that omits the size, and a legitimately + truncated digest would never match one computed at full length. + + Verification is skipped only when the local ``multiformats`` build cannot + compute the function at all -- we cannot prove the content wrong, so the + read is allowed through with a warning. That path must stay narrow: silently + skipping is indistinguishable from passing, which would make + ``verify_content`` worthless exactly when it matters. """ + raw_digest = bytes(cid.raw_digest) + if cid.hashfun.name == "identity": + # Nothing was hashed: the CID carries the content verbatim. + if data != raw_digest: + raise GatewayContentMismatch( + f"gateway returned {len(data)} bytes that do not match the " + f"content inlined in identity CID {cid}" + ) + return + try: - computed = multihash.digest(data, cid.hashfun.name) + computed = multihash.digest(data, cid.hashfun.name, size=len(raw_digest)) except Exception: # pragma: no cover - depends on multiformats build - logger.debug( - "Cannot verify CID %s: unsupported hash function %s", + logger.warning( + "Cannot verify CID %s: local multiformats cannot compute %s at " + "%d bytes; returning unverified gateway content", cid, cid.hashfun.name, + len(raw_digest), ) return - if computed != cid.digest: + if bytes(multihash.unwrap(computed)) != raw_digest: raise GatewayContentMismatch( f"gateway returned {len(data)} bytes that do not hash to {cid}" ) @@ -468,6 +508,9 @@ class KuboCAS(ContentAddressedStore): gateways you do not control. Only full-body reads of non-`dag-pb` CIDs can be verified; Range reads and `dag-pb` reads are passed through unchecked because neither returns the exact bytes the CID commits to. + Credentials are never sent to a gateway outside the origin of + `gateway_base_url`/`rpc_base_url`, so a private primary can safely be + paired with public fallbacks. - **chunker** (str): chunking algorithm specification for Kubo's `add` RPC. Accepted formats are `"size-"`, `"rabin"`, or `"rabin---"`. @@ -611,6 +654,14 @@ def __init__( self.gateway_base_url: str = self.gateway_base_urls[0] """@private""" + # Origins the caller's credentials were configured for: the primary + # gateway and the RPC endpoint. Reads to any other gateway drop + # credentialed headers and client auth (see _load_from_gateway). + self._credentialed_origins: set[tuple[str, str, int | None]] = { + _origin_of(self.gateway_base_url), + _origin_of(rpc_base_url), + } + self.verify_content: bool = verify_content """@private""" # Health is per gateway but shared across event loops: a gateway that is @@ -740,7 +791,7 @@ def _ordered_gateways(self) -> list[str]: if len(self.gateway_base_urls) == 1: return self.gateway_base_urls - now = asyncio.get_running_loop().time() + now = time.monotonic() # is_healthy() clears an expired trip, so evaluate it exactly once per # gateway rather than inside a sort key, which may call it repeatedly. ranks: Dict[str, tuple[int, int]] = {} @@ -1011,16 +1062,46 @@ async def _load_from_gateway( Raises on failure so the caller can fail over. ``stats`` accumulates the byte count and retry total across every gateway attempted, so the trace emitted by ``load`` reflects the whole operation rather than the last leg. + + Credentials configured for the primary gateway or the RPC endpoint are + stripped when this gateway is on a different origin, so failing over to + a public fallback cannot disclose a private gateway's token. """ url = f"{gateway_base_url}{cid}" client = self._loop_client() semaphore = self._gateway_semaphore(gateway_base_url) retry_count = 0 + # httpx merges client-level headers into every request and offers no way + # to drop one per-request (Client._merge_headers starts from + # self.headers and only update()s, so an omitted or blanked entry is + # reinstated). Building the Request explicitly and calling send() + # bypasses that merge, which is the only reliable way to withhold a + # credential from a foreign origin. + strip_credentials = ( + _origin_of(gateway_base_url) not in self._credentialed_origins + ) + request: httpx.Request | None = None + if strip_credentials: + safe_headers = { + name: value + for name, value in client.headers.items() + if name.lower() not in _SENSITIVE_HEADERS + } + safe_headers.update(headers) + request = httpx.Request("GET", url, headers=safe_headers) + while retry_count <= self.max_retries: try: async with semaphore: # Throttle each gateway attempt - response = await client.get(url, headers=headers or None) + if request is not None: + # auth=None also suppresses client-level httpx.Auth, + # which would otherwise re-add an Authorization header. + response = await client.send( + request, auth=None, follow_redirects=client.follow_redirects + ) + else: + response = await client.get(url, headers=headers or None) # An unsatisfiable range is answered with 416 by a compliant # gateway; return b"" to match Python-slice semantics (and # InMemoryCAS) instead of raising. @@ -1060,7 +1141,11 @@ async def _load_from_gateway( # A mismatch means this gateway served wrong bytes. Raising # here routes it through the caller's failover path like any # other per-gateway failure. - _verify_cid_content(cid, content) + try: + _verify_cid_content(cid, content) + except GatewayContentMismatch: + stats.status = "content_mismatch" + raise return content except httpx.RequestError: @@ -1149,7 +1234,7 @@ async def load( gateway_base_url, cid, headers, offset, length, suffix, stats ) except (httpx.HTTPError, GatewayContentMismatch) as error: - health.record_failure(asyncio.get_running_loop().time()) + health.record_failure(time.monotonic()) failures.append(error) if len(gateways) > 1: logger.debug( diff --git a/tests/test_k15_multi_gateway_failover.py b/tests/test_k15_multi_gateway_failover.py index 5e1c019..69a9636 100644 --- a/tests/test_k15_multi_gateway_failover.py +++ b/tests/test_k15_multi_gateway_failover.py @@ -2,6 +2,7 @@ import asyncio import threading +import time from collections.abc import Iterator from dataclasses import dataclass, field from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -25,6 +26,8 @@ class FakeGateway: url: str hits: list[str] = field(default_factory=list) + # Lower-cased request headers, one dict per request served. + headers_seen: list[dict[str, str]] = field(default_factory=list) # Set by tests to control responses; returns (status, body). responder: Callable[[str], tuple[int, bytes]] = lambda _cid: (200, BODY) max_concurrent: int = 0 @@ -46,6 +49,11 @@ def do_GET(self) -> None: cid = self.path.rsplit("/", 1)[-1].split("?", 1)[0] with gw._lock: gw.hits.append(cid) + # Header names are case-insensitive on the wire; normalize so + # assertions cannot pass or fail on casing alone. + gw.headers_seen.append({ + name.lower(): value for name, value in self.headers.items() + }) gw._inflight += 1 gw.max_concurrent = max(gw.max_concurrent, gw._inflight) try: @@ -297,10 +305,22 @@ def flaky(_cid: str) -> tuple[int, bytes]: assert await cas.load(GOOD_CID) == BODY health = cas._gateway_health[cas.gateway_base_urls[0]] assert health.consecutive_failures == 0 - # A further failure does not immediately trip a stale counter. - assert await cas.load(GOOD_CID) == BODY + + # Loads 1 and 2 failed over; load 3 was served by A itself. assert len(gateway_b.hits) == 2 + # A further failure must not trip the breaker on a stale count: the + # reset means this is failure 1 of 3, not 3 of 3. + responses.append((500, b"")) + assert await cas.load(GOOD_CID) == BODY + assert len(gateway_b.hits) == 3, "load 4 should have failed over" + assert health.consecutive_failures == 1 + assert health.tripped_at is None, "breaker tripped on a stale count" + + # Untripped, A stays in the rotation and serves the next read itself. + assert await cas.load(GOOD_CID) == BODY + assert len(gateway_b.hits) == 3, "A should have served this without failover" + @pytest.mark.asyncio async def test_all_gateways_unhealthy_still_attempts_every_one( @@ -484,6 +504,257 @@ async def test_range_reads_are_not_verified(gateway_a: FakeGateway) -> None: assert await cas.load(GOOD_CID, offset=0, length=5) == BODY[:5] +@pytest.mark.asyncio +async def test_failed_load_is_not_traced_as_a_success( + gateway_a: FakeGateway, monkeypatch: pytest.MonkeyPatch +) -> None: + """A raised load must not be recorded as ``status="ok"``. + + Verification runs outside the ``except`` blocks that set a failure status, + so without an explicit assignment a mismatch would raise to the caller + while the trace reported success. + """ + seen: list[str] = [] + original = store_module.instrumentation.end_cas_load + + def spy(trace: object, *, byte_count: int, retries: int, status: str) -> None: + seen.append(status) + return original(trace, byte_count=byte_count, retries=retries, status=status) + + monkeypatch.setattr(store_module.instrumentation, "end_cas_load", spy) + gateway_a.responder = lambda _cid: (200, b"corrupted payload") + + async with make_cas(gateway_a, verify_content=True, max_retries=0) as cas: + with pytest.raises(GatewayContentMismatch): + await cas.load(GOOD_CID) + + assert seen == ["content_mismatch"] + + +@pytest.mark.asyncio +async def test_recovery_after_a_failed_gateway_traces_as_success( + gateway_a: FakeGateway, gateway_b: FakeGateway, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failed leg followed by a successful one is one successful load.""" + seen: list[str] = [] + original = store_module.instrumentation.end_cas_load + + def spy(trace: object, *, byte_count: int, retries: int, status: str) -> None: + seen.append(status) + return original(trace, byte_count=byte_count, retries=retries, status=status) + + monkeypatch.setattr(store_module.instrumentation, "end_cas_load", spy) + gateway_a.responder = lambda _cid: (500, b"") + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert seen == ["ok"] + + +def test_gateway_health_uses_a_process_wide_clock() -> None: + """Health must be timed by a process-wide clock, not a per-loop one. + + ``_gateway_health`` is shared across event loops, so a trip recorded under + one loop is later compared against a reading taken under another. + ``asyncio`` documents ``loop.time()`` as having an epoch that "may differ + per event loop", so pairing it with this shared state would make the + cooldown expire instantly or never. + + CPython's default loop happens to implement ``time()`` as + ``time.monotonic()``, which would hide the bug, so this pins the property + against a loop whose epoch genuinely differs -- exactly what a custom or + uvloop-style event loop is permitted to do. + """ + cas = KuboCAS( + gateway_base_urls=["http://127.0.0.1:1", "http://127.0.0.1:2"], + rpc_base_url="http://127.0.0.1:1", + ) + tripped_url = cas.gateway_base_urls[0] + health = cas._gateway_health[tripped_url] + for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): + health.record_failure(time.monotonic()) + assert health.tripped_at is not None + + class SkewedEpochLoop(asyncio.SelectorEventLoop): + """A conforming loop whose clock epoch is far from ``monotonic()``.""" + + def time(self) -> float: + # Legal per the asyncio contract: monotonic, unspecified epoch. + return time.monotonic() + 1_000_000.0 + + loop = SkewedEpochLoop() + try: + ordering = loop.run_until_complete(_ordered_on(cas)) + finally: + loop.close() + asyncio.set_event_loop(None) + + # Under loop.time() the trip would look ~1e6 seconds old and the cooldown + # long expired, floating the dead gateway back to the front. + assert ordering == [cas.gateway_base_urls[1], tripped_url], ( + "cooldown must not be measured against a per-loop clock epoch" + ) + + +async def _ordered_on(cas: KuboCAS) -> list[str]: + return cas._ordered_gateways() + + +# --------------------------------------------------------------------------- # +# credential scoping # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_credentials_are_not_leaked_to_a_fallback_gateway( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """A private gateway's token must not reach a public fallback. + + Credentials live on the shared httpx client, so without explicit scoping a + 503 from the private gateway would hand its bearer token to whatever + gateway is next in the rotation. + """ + gateway_a.responder = lambda _cid: (503, b"") + + async with make_cas( + gateway_a, + gateway_b, + max_retries=0, + headers={"Authorization": "Bearer SECRET", "X-Trace": "keepme"}, + ) as cas: + assert await cas.load(GOOD_CID) == BODY + + primary = gateway_a.headers_seen[0] + fallback = gateway_b.headers_seen[0] + assert primary["authorization"] == "Bearer SECRET" + assert "authorization" not in fallback, "credential leaked to fallback gateway" + # Only credentialed headers are dropped; ordinary ones still travel. + assert fallback["x-trace"] == "keepme" + + +@pytest.mark.asyncio +async def test_credentials_are_withheld_on_every_retry_to_a_foreign_origin( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """Retries against the fallback must not re-add the credential.""" + gateway_a.responder = lambda _cid: (503, b"") + attempts = {"n": 0} + + def flaky(_cid: str) -> tuple[int, bytes]: + attempts["n"] += 1 + return (200, BODY) if attempts["n"] > 2 else (503, b"") + + gateway_b.responder = flaky + + async with make_cas( + gateway_a, + gateway_b, + max_retries=2, + headers={"Authorization": "Bearer SECRET"}, + ) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert len(gateway_b.headers_seen) == 3 + assert all("authorization" not in seen for seen in gateway_b.headers_seen) + + +@pytest.mark.asyncio +async def test_client_level_auth_is_also_withheld( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """``auth=`` builds an httpx.Auth flow, which must not run on a fallback.""" + gateway_a.responder = lambda _cid: (503, b"") + + async with make_cas( + gateway_a, gateway_b, max_retries=0, auth=("user", "hunter2") + ) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert "authorization" in gateway_a.headers_seen[0] + assert "authorization" not in gateway_b.headers_seen[0] + + +@pytest.mark.asyncio +async def test_credentials_are_kept_for_a_same_origin_gateway( + gateway_a: FakeGateway, +) -> None: + """Scoping must not strip credentials from the gateway they belong to.""" + async with make_cas(gateway_a, headers={"Authorization": "Bearer SECRET"}) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert gateway_a.headers_seen[0]["authorization"] == "Bearer SECRET" + + +# --------------------------------------------------------------------------- # +# digest computation # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_blake3_cids_are_actually_verified(gateway_a: FakeGateway) -> None: + """blake3 is this library's default hasher and must not skip verification. + + ``multihash.digest`` *raises* for variable-output functions when no size is + given, so computing at the default length would send every blake3 check + into the unsupported-function path and silently accept substituted content. + """ + body = b"blake3 addressed block" + cid = CID("base32", 1, "raw", multihash.digest(body, "blake3", size=32)) + gateway_a.responder = lambda _cid: (200, b"substituted by the gateway") + + async with make_cas(gateway_a, verify_content=True, max_retries=0) as cas: + with pytest.raises(GatewayContentMismatch): + await cas.load(cid) + + gateway_a.responder = lambda _cid: (200, body) + async with make_cas(gateway_a, verify_content=True) as cas: + assert await cas.load(cid) == body + + +@pytest.mark.asyncio +async def test_truncated_digests_are_verified_at_their_own_length( + gateway_a: FakeGateway, +) -> None: + """A legitimately truncated digest must not be rejected as a mismatch.""" + body = b"truncated digest block" + cid = CID("base32", 1, "raw", multihash.digest(body, "sha2-256", size=20)) + assert len(bytes(cid.raw_digest)) == 20 + + gateway_a.responder = lambda _cid: (200, body) + async with make_cas(gateway_a, verify_content=True) as cas: + assert await cas.load(cid) == body, "correct content rejected" + + gateway_a.responder = lambda _cid: (200, b"wrong") + async with make_cas(gateway_a, verify_content=True, max_retries=0) as cas: + with pytest.raises(GatewayContentMismatch): + await cas.load(cid) + + +@pytest.mark.asyncio +async def test_identity_cids_are_verified_against_inlined_content( + gateway_a: FakeGateway, +) -> None: + """load() returns gateway bytes for identity CIDs, so they must be checked. + + The content is inlined in the CID itself, so a mismatch is provable by + direct comparison -- skipping it would let a gateway substitute content + that the caller could have validated locally. + """ + body = b"inlined block" + cid = CID("base32", 1, "raw", multihash.digest(body, "identity")) + + gateway_a.responder = lambda _cid: (200, b"substituted") + async with make_cas(gateway_a, verify_content=True, max_retries=0) as cas: + with pytest.raises(GatewayContentMismatch, match="inlined"): + await cas.load(cid) + + gateway_a.responder = lambda _cid: (200, body) + async with make_cas(gateway_a, verify_content=True) as cas: + assert await cas.load(cid) == body + + def test_dag_pb_cids_are_not_verifiable() -> None: """Gateways return reassembled UnixFS files for dag-pb, not the block.""" dag_pb_cid = CID("base32", 1, "dag-pb", multihash.digest(BODY, "sha2-256")) @@ -492,6 +763,11 @@ def test_dag_pb_cids_are_not_verifiable() -> None: assert not store_module._cid_is_verifiable(GOOD_CID, 0, None) assert not store_module._cid_is_verifiable(GOOD_CID, None, 4) + # Identity CIDs inline their content but load() still returns gateway + # bytes, so the response is verifiable by direct comparison. + identity_cid = CID("base32", 1, "raw", multihash.digest(BODY, "identity")) + assert store_module._cid_is_verifiable(identity_cid, None, None) + def test_unsupported_hash_function_is_not_a_mismatch( monkeypatch: pytest.MonkeyPatch, From e31b9fc8135ea608c297d5febde5357da74d2e55 Mon Sep 17 00:00:00 2001 From: TheGreatAlgo <37487508+TheGreatAlgo@users.noreply.github.com> Date: Mon, 27 Jul 2026 10:10:56 -0400 Subject: [PATCH 3/5] fix: gateway forwarding --- py_hamt/store_httpx.py | 35 +++++++++--- tests/test_k15_multi_gateway_failover.py | 71 +++++++++++++++++++----- 2 files changed, 82 insertions(+), 24 deletions(-) diff --git a/py_hamt/store_httpx.py b/py_hamt/store_httpx.py index 3a5a832..957dbf8 100644 --- a/py_hamt/store_httpx.py +++ b/py_hamt/store_httpx.py @@ -33,11 +33,22 @@ _GATEWAY_FAILURE_THRESHOLD = 3 _GATEWAY_COOLDOWN_SECONDS = 30.0 -# Request headers that carry credentials. When a read falls over to a gateway -# on a different origin than the one the credentials were configured for, these -# are stripped: a token minted for a private gateway must not be handed to a -# public fallback simply because the private one returned 503. -_SENSITIVE_HEADERS = frozenset({"authorization", "proxy-authorization", "cookie"}) +# Headers forwarded to a gateway outside the credentialed origin. This is an +# allowlist, not a denylist: KuboCAS documents arbitrary headers as a supported +# way to authenticate ("set whatever headers ... they need"), so any header the +# caller configured may be a credential -- ``X-API-Key`` and ``X-Auth-Token`` +# name themselves, but a bearer token can live under any name at all. Only +# httpx's own content-negotiation defaults are safe to send to a foreign +# gateway; anything else is dropped rather than guessed about. +# +# ``Host`` is deliberately absent: it is derived from the request URL, and +# forwarding the primary's value would misroute the fallback request. +_FORWARDABLE_HEADERS = frozenset({ + "accept", + "accept-encoding", + "accept-language", + "user-agent", +}) def _origin_of(url: str) -> tuple[str, str, int | None]: @@ -508,9 +519,13 @@ class KuboCAS(ContentAddressedStore): gateways you do not control. Only full-body reads of non-`dag-pb` CIDs can be verified; Range reads and `dag-pb` reads are passed through unchecked because neither returns the exact bytes the CID commits to. - Credentials are never sent to a gateway outside the origin of - `gateway_base_url`/`rpc_base_url`, so a private primary can safely be - paired with public fallbacks. + Requests to a gateway outside the origin of + `gateway_base_url`/`rpc_base_url` forward only content-negotiation + headers (`Accept`, `Accept-Encoding`, `Accept-Language`, `User-Agent`) + and drop client-level `auth`. Because any header name may carry a + credential, everything else is withheld -- so a private primary can + safely be paired with public fallbacks, but a foreign gateway that needs + its own custom header will not receive one. - **chunker** (str): chunking algorithm specification for Kubo's `add` RPC. Accepted formats are `"size-"`, `"rabin"`, or `"rabin---"`. @@ -1086,8 +1101,10 @@ async def _load_from_gateway( safe_headers = { name: value for name, value in client.headers.items() - if name.lower() not in _SENSITIVE_HEADERS + if name.lower() in _FORWARDABLE_HEADERS } + # Range headers are computed by load() for this request, never + # caller-supplied credentials, so they are always safe to send. safe_headers.update(headers) request = httpx.Request("GET", url, headers=safe_headers) diff --git a/tests/test_k15_multi_gateway_failover.py b/tests/test_k15_multi_gateway_failover.py index 69a9636..f3c79ae 100644 --- a/tests/test_k15_multi_gateway_failover.py +++ b/tests/test_k15_multi_gateway_failover.py @@ -14,6 +14,7 @@ from py_hamt import GatewayContentMismatch, KuboCAS from py_hamt import store_httpx as store_module +from py_hamt.instrumentation import TraceContext BODY = b"multi gateway body" # raw + sha2-256 CID over BODY, so verify_content can check it for real. @@ -517,7 +518,9 @@ async def test_failed_load_is_not_traced_as_a_success( seen: list[str] = [] original = store_module.instrumentation.end_cas_load - def spy(trace: object, *, byte_count: int, retries: int, status: str) -> None: + def spy( + trace: TraceContext | None, *, byte_count: int, retries: int, status: str + ) -> None: seen.append(status) return original(trace, byte_count=byte_count, retries=retries, status=status) @@ -539,7 +542,9 @@ async def test_recovery_after_a_failed_gateway_traces_as_success( seen: list[str] = [] original = store_module.instrumentation.end_cas_load - def spy(trace: object, *, byte_count: int, retries: int, status: str) -> None: + def spy( + trace: TraceContext | None, *, byte_count: int, retries: int, status: str + ) -> None: seen.append(status) return original(trace, byte_count=byte_count, retries=retries, status=status) @@ -617,21 +622,30 @@ async def test_credentials_are_not_leaked_to_a_fallback_gateway( gateway is next in the rotation. """ gateway_a.responder = lambda _cid: (503, b"") - - async with make_cas( - gateway_a, - gateway_b, - max_retries=0, - headers={"Authorization": "Bearer SECRET", "X-Trace": "keepme"}, - ) as cas: + # KuboCAS documents arbitrary headers as a way to authenticate, so a + # credential can carry any name. A denylist of well-known header names + # would forward every one of these but the first. + secrets = { + "Authorization": "Bearer SECRET", + "Cookie": "session=abc", + "X-API-Key": "key-123", + "X-Auth-Token": "token-456", + "X-Custom-Corp-Secret": "nobody-guesses-this", + } + + async with make_cas(gateway_a, gateway_b, max_retries=0, headers=secrets) as cas: assert await cas.load(GOOD_CID) == BODY primary = gateway_a.headers_seen[0] fallback = gateway_b.headers_seen[0] - assert primary["authorization"] == "Bearer SECRET" - assert "authorization" not in fallback, "credential leaked to fallback gateway" - # Only credentialed headers are dropped; ordinary ones still travel. - assert fallback["x-trace"] == "keepme" + for name, value in secrets.items(): + assert primary[name.lower()] == value, f"{name} lost on its own gateway" + assert name.lower() not in fallback, f"{name} leaked to fallback gateway" + + # The fallback still gets ordinary content negotiation, and a Host derived + # from its own URL rather than the primary's. + assert "accept" in fallback + assert fallback["host"] not in gateway_a.url @pytest.mark.asyncio @@ -652,12 +666,14 @@ def flaky(_cid: str) -> tuple[int, bytes]: gateway_a, gateway_b, max_retries=2, - headers={"Authorization": "Bearer SECRET"}, + headers={"Authorization": "Bearer SECRET", "X-API-Key": "key-123"}, ) as cas: assert await cas.load(GOOD_CID) == BODY assert len(gateway_b.headers_seen) == 3 - assert all("authorization" not in seen for seen in gateway_b.headers_seen) + for seen in gateway_b.headers_seen: + assert "authorization" not in seen + assert "x-api-key" not in seen @pytest.mark.asyncio @@ -676,6 +692,31 @@ async def test_client_level_auth_is_also_withheld( assert "authorization" not in gateway_b.headers_seen[0] +def test_forwardable_headers_is_an_allowlist_of_non_credentials() -> None: + """The forwarding rule must be allow-by-name, not deny-by-name. + + A denylist cannot work here: a credential may use any header name, so + anything not explicitly known to be safe must be dropped. ``Host`` is + excluded too -- it is derived from the target URL, and forwarding the + primary's value would misroute the fallback. + """ + assert store_module._FORWARDABLE_HEADERS == frozenset({ + "accept", + "accept-encoding", + "accept-language", + "user-agent", + }) + for credential_name in ( + "authorization", + "cookie", + "proxy-authorization", + "x-api-key", + "x-auth-token", + "host", + ): + assert credential_name not in store_module._FORWARDABLE_HEADERS + + @pytest.mark.asyncio async def test_credentials_are_kept_for_a_same_origin_gateway( gateway_a: FakeGateway, From 2643c9d900a6ed71b26aa7c02b3881e7a0c48a69 Mon Sep 17 00:00:00 2001 From: TheGreatAlgo <37487508+TheGreatAlgo@users.noreply.github.com> Date: Mon, 27 Jul 2026 10:57:08 -0400 Subject: [PATCH 4/5] fix: content unveriabile --- py_hamt/__init__.py | 2 + py_hamt/store_httpx.py | 211 +++++++++++++++++------ tests/test_k15_multi_gateway_failover.py | 191 +++++++++++++++++++- 3 files changed, 350 insertions(+), 54 deletions(-) diff --git a/py_hamt/__init__.py b/py_hamt/__init__.py index 6ba81bf..a54e699 100644 --- a/py_hamt/__init__.py +++ b/py_hamt/__init__.py @@ -9,6 +9,7 @@ from .store_httpx import ( ContentAddressedStore, GatewayContentMismatch, + GatewayContentUnverifiable, InMemoryCAS, KuboCAS, ) @@ -19,6 +20,7 @@ "HAMT", "ContentAddressedStore", "GatewayContentMismatch", + "GatewayContentUnverifiable", "InMemoryCAS", "KuboCAS", "ZarrHAMTStore", diff --git a/py_hamt/store_httpx.py b/py_hamt/store_httpx.py index 957dbf8..8842d62 100644 --- a/py_hamt/store_httpx.py +++ b/py_hamt/store_httpx.py @@ -51,6 +51,24 @@ }) +def _carries_credentials(client: httpx.AsyncClient) -> bool: + """Whether this client would send anything worth withholding. + + A client with no auth and only content-negotiation headers has no secret to + leak to a foreign gateway, so reads through it can take httpx's ordinary + request path instead of the origin-scoping one. + """ + if client.auth is not None: + return True + # "connection" is a hop-by-hop header httpx sets itself. It is not + # forwardable (it describes this connection, not the request), but its + # presence does not mean the caller configured a credential. + benign = _FORWARDABLE_HEADERS | {"connection"} + return any( + name.decode("latin-1").lower() not in benign for name, _ in client.headers.raw + ) + + def _origin_of(url: str) -> tuple[str, str, int | None]: """Scheme/host/port triple used to decide if two URLs share an origin.""" parsed = urlsplit(url) @@ -85,31 +103,45 @@ class _GatewayHealth: state is touched from a second loop. """ - __slots__ = ("consecutive_failures", "tripped_at") + __slots__ = ("consecutive_failures", "probe_in_flight", "tripped_at") def __init__(self) -> None: self.consecutive_failures: int = 0 self.tripped_at: float | None = None + # True between handing out a post-cooldown probe slot and learning how + # that probe went. Keeps the gateway closed to everyone else meanwhile. + self.probe_in_flight: bool = False def record_success(self) -> None: self.consecutive_failures = 0 self.tripped_at = None + self.probe_in_flight = False def record_failure(self, now: float) -> None: self.consecutive_failures += 1 + self.probe_in_flight = False if self.consecutive_failures >= _GATEWAY_FAILURE_THRESHOLD: self.tripped_at = now def is_healthy(self, now: float) -> bool: + """Whether this gateway should be preferred, claiming a probe if due. + + Half-open, not merely time-based: exactly one caller past the cooldown + is let through as a probe, and the gateway stays deprioritized for + everyone else until that probe reports back via ``record_success`` or + ``record_failure``. Clearing the trip on a timer instead would let every + concurrent load in at once and dogpile a gateway that is still down. + """ if self.tripped_at is None: return True + if self.probe_in_flight: + # A probe is already out; nobody else gets through on its coattails. + return False if now - self.tripped_at >= _GATEWAY_COOLDOWN_SECONDS: - # Cooldown elapsed. Clear the trip so a single probe failure does - # not immediately re-trip on a stale counter, but keep the gateway - # on probation by leaving the failure count one short of the - # threshold: one more failure re-trips it right away. - self.tripped_at = None - self.consecutive_failures = _GATEWAY_FAILURE_THRESHOLD - 1 + # Claim the probe slot. tripped_at stays set so that if this probe + # fails, the gateway remains tripped without needing to re-cross the + # failure threshold; record_failure refreshes the cooldown window. + self.probe_in_flight = True return True return False @@ -164,6 +196,18 @@ class GatewayContentMismatch(Exception): """ +class GatewayContentUnverifiable(GatewayContentMismatch): + """Verification was requested but could not be carried out. + + Subclasses ``GatewayContentMismatch`` so it fails over and is caught by + existing handlers. Distinct because the cause differs: the content is not + known to be wrong, only unproven. ``verify_content=True`` must still fail + closed here -- returning unverified bytes would silently downgrade the + guarantee exactly when the hashing backend is broken or the algorithm is + unavailable. + """ + + def _cid_is_verifiable(cid: CID, offset: Optional[int], suffix: Optional[int]) -> bool: """Whether a gateway response for ``cid`` can be checked against its digest. @@ -195,11 +239,12 @@ def _verify_cid_content(cid: CID, data: bytes) -> None: default hasher here) *reject* a call that omits the size, and a legitimately truncated digest would never match one computed at full length. - Verification is skipped only when the local ``multiformats`` build cannot - compute the function at all -- we cannot prove the content wrong, so the - read is allowed through with a warning. That path must stay narrow: silently - skipping is indistinguishable from passing, which would make - ``verify_content`` worthless exactly when it matters. + If the digest cannot be computed at all -- an algorithm the local + ``multiformats`` build does not support, or a hashing backend that fails -- + this raises ``GatewayContentUnverifiable`` rather than returning the bytes. + Verification fails closed: a caller who asked for ``verify_content`` gets + either proven content or an error, never unverified bytes presented as + though they had been checked. """ raw_digest = bytes(cid.raw_digest) if cid.hashfun.name == "identity": @@ -213,15 +258,11 @@ def _verify_cid_content(cid: CID, data: bytes) -> None: try: computed = multihash.digest(data, cid.hashfun.name, size=len(raw_digest)) - except Exception: # pragma: no cover - depends on multiformats build - logger.warning( - "Cannot verify CID %s: local multiformats cannot compute %s at " - "%d bytes; returning unverified gateway content", - cid, - cid.hashfun.name, - len(raw_digest), - ) - return + except Exception as error: + raise GatewayContentUnverifiable( + f"cannot verify {cid}: computing {cid.hashfun.name} at " + f"{len(raw_digest)} bytes failed ({error})" + ) from error if bytes(multihash.unwrap(computed)) != raw_digest: raise GatewayContentMismatch( f"gateway returned {len(data)} bytes that do not hash to {cid}" @@ -1062,6 +1103,93 @@ async def save(self, data: bytes, codec: ContentAddressedStore.CodecInput) -> CI ) raise RuntimeError("Exited the retry loop unexpectedly.") # pragma: no cover + def _request_headers_for( + self, client: httpx.AsyncClient, url: str, headers: Dict[str, str] + ) -> Tuple[Dict[str, str], bool]: + """Build headers for ``url``, dropping credentials on a foreign origin. + + Returns the headers and whether credentials were withheld (which also + means client-level ``auth`` must be suppressed). + + httpx merges client-level headers into every request and offers no way + to drop one per-request: ``Client._merge_headers`` starts from + ``self.headers`` and only ``update()``s, so an omitted or blanked entry + is reinstated. Building the ``Request`` explicitly is the only reliable + way to withhold a credential. + """ + # Read from .raw: httpx.Headers.items() lower-cases names, and building + # a Request from that would silently rewrite every outgoing header name + # compared with the client.get() path this replaces. + client_headers = [ + (name.decode("latin-1"), value.decode("latin-1")) + for name, value in client.headers.raw + ] + + if _origin_of(url) in self._credentialed_origins: + merged = dict(client_headers) + merged.update(headers) + return merged, False + + safe_headers = { + name: value + for name, value in client_headers + if name.lower() in _FORWARDABLE_HEADERS + } + # Range headers are computed by load() for this request, never + # caller-supplied credentials, so they are always safe to send. + safe_headers.update(headers) + return safe_headers, True + + async def _get_with_origin_scoped_credentials( + self, client: httpx.AsyncClient, url: str, headers: Dict[str, str] + ) -> httpx.Response: + """GET ``url``, re-deciding credential scope at every redirect hop. + + Redirects are followed manually because httpx's own redirect handling + strips only ``Authorization`` and ``Cookie`` when crossing origins (see + ``Client._redirect_headers``). Custom credentials -- ``X-API-Key`` and + friends, which ``KuboCAS`` documents as a supported way to + authenticate -- survive its stripping, so a gateway could redirect to an + origin of its choosing and harvest them. Following each hop ourselves + re-applies the full origin check to the *redirect target*. + """ + if not _carries_credentials(client): + # Nothing to withhold: no credentialed header and no client auth, so + # no origin can harvest anything by redirecting. Keep the plain + # client.get() path, which preserves httpx's own redirect, auth, and + # header-casing behaviour for the overwhelmingly common case. + return await client.get(url, headers=headers or None) + + redirects_remaining = client.max_redirects if client.follow_redirects else 0 + current_url = url + + while True: + request_headers, stripped = self._request_headers_for( + client, current_url, headers + ) + # auth=None also suppresses client-level httpx.Auth, which would + # otherwise re-add an Authorization header after our filtering. + response = await client.send( + httpx.Request("GET", current_url, headers=request_headers), + auth=None if stripped else httpx.USE_CLIENT_DEFAULT, + follow_redirects=False, + ) + + location = response.headers.get("Location") + if not (response.is_redirect and location): + return response + if redirects_remaining <= 0: + # Mirror httpx's own behaviour rather than silently returning + # the 3xx as if it were the block. + raise httpx.TooManyRedirects( + "Exceeded maximum allowed redirects.", request=response.request + ) + + redirects_remaining -= 1 + await response.aread() + await response.aclose() + current_url = str(response.url.join(location)) + async def _load_from_gateway( self, gateway_base_url: str, @@ -1087,38 +1215,12 @@ async def _load_from_gateway( semaphore = self._gateway_semaphore(gateway_base_url) retry_count = 0 - # httpx merges client-level headers into every request and offers no way - # to drop one per-request (Client._merge_headers starts from - # self.headers and only update()s, so an omitted or blanked entry is - # reinstated). Building the Request explicitly and calling send() - # bypasses that merge, which is the only reliable way to withhold a - # credential from a foreign origin. - strip_credentials = ( - _origin_of(gateway_base_url) not in self._credentialed_origins - ) - request: httpx.Request | None = None - if strip_credentials: - safe_headers = { - name: value - for name, value in client.headers.items() - if name.lower() in _FORWARDABLE_HEADERS - } - # Range headers are computed by load() for this request, never - # caller-supplied credentials, so they are always safe to send. - safe_headers.update(headers) - request = httpx.Request("GET", url, headers=safe_headers) - while retry_count <= self.max_retries: try: async with semaphore: # Throttle each gateway attempt - if request is not None: - # auth=None also suppresses client-level httpx.Auth, - # which would otherwise re-add an Authorization header. - response = await client.send( - request, auth=None, follow_redirects=client.follow_redirects - ) - else: - response = await client.get(url, headers=headers or None) + response = await self._get_with_origin_scoped_credentials( + client, url, headers + ) # An unsatisfiable range is answered with 416 by a compliant # gateway; return b"" to match Python-slice semantics (and # InMemoryCAS) instead of raising. @@ -1243,9 +1345,11 @@ async def load( stats = _LoadStats() gateways = self._ordered_gateways() failures: list[Exception] = [] + attempted: set[str] = set() try: for gateway_base_url in gateways: health = self._gateway_health[gateway_base_url] + attempted.add(gateway_base_url) try: content = await self._load_from_gateway( gateway_base_url, cid, headers, offset, length, suffix, stats @@ -1277,6 +1381,13 @@ async def load( f"all {len(gateways)} gateways failed for CID {cid}", failures ) finally: + # Ranking claims a probe slot for any tripped gateway whose cooldown + # has elapsed, but an earlier gateway usually succeeds first and the + # rest are never tried. Release those unused claims, or the slot is + # never reported back and the gateway is locked out permanently. + for gateway_base_url in gateways: + if gateway_base_url not in attempted: + self._gateway_health[gateway_base_url].probe_in_flight = False instrumentation.end_cas_load( trace_started_at, byte_count=stats.response_bytes, diff --git a/tests/test_k15_multi_gateway_failover.py b/tests/test_k15_multi_gateway_failover.py index f3c79ae..d3120eb 100644 --- a/tests/test_k15_multi_gateway_failover.py +++ b/tests/test_k15_multi_gateway_failover.py @@ -31,6 +31,8 @@ class FakeGateway: headers_seen: list[dict[str, str]] = field(default_factory=list) # Set by tests to control responses; returns (status, body). responder: Callable[[str], tuple[int, bytes]] = lambda _cid: (200, BODY) + # When the responder returns a 3xx, redirect here (the CID is appended). + redirect_to: str | None = None max_concurrent: int = 0 _inflight: int = 0 _lock: threading.Lock = field(default_factory=threading.Lock) @@ -63,6 +65,8 @@ def do_GET(self) -> None: with gw._lock: gw._inflight -= 1 self.send_response(status) + if 300 <= status < 400 and gw.redirect_to: + self.send_header("Location", f"{gw.redirect_to}{cid}") self.send_header("Content-Length", str(len(body))) self.end_headers() if body: @@ -356,6 +360,60 @@ def test_health_cooldown_leaves_gateway_on_probation() -> None: assert not health.is_healthy(now=store_module._GATEWAY_COOLDOWN_SECONDS + 1.0) +def test_only_one_probe_is_admitted_after_cooldown() -> None: + """Half-open, not merely time-based. + + Clearing the trip on a timer would let every concurrent load through at + once and dogpile a gateway that is still down. Exactly one caller gets a + probe slot; the rest stay deprioritized until it reports back. + """ + health = store_module._GatewayHealth() + for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): + health.record_failure(now=0.0) + after = store_module._GATEWAY_COOLDOWN_SECONDS + 1.0 + + assert health.is_healthy(now=after), "the first caller should get the probe" + assert [health.is_healthy(now=after) for _ in range(8)] == [False] * 8 + + # A failed probe keeps it tripped without re-crossing the threshold. + health.record_failure(now=after) + assert not health.is_healthy(now=after + 1.0) + + # A successful probe restores it for everyone. + later = after + store_module._GATEWAY_COOLDOWN_SECONDS + 1.0 + assert health.is_healthy(now=later) + health.record_success() + assert all(health.is_healthy(now=later) for _ in range(5)) + + +@pytest.mark.asyncio +async def test_unused_probe_claim_is_released( + gateway_a: FakeGateway, gateway_b: FakeGateway, monkeypatch: pytest.MonkeyPatch +) -> None: + """A gateway ranked but never tried must not keep its probe slot. + + Ranking claims the slot, but the loop stops at the first success, so a + lower-ranked gateway is usually never attempted. Without releasing the + claim it would never report back and would be locked out forever. + """ + monkeypatch.setattr(store_module, "_GATEWAY_COOLDOWN_SECONDS", 0.0) + + async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: + health_b = cas._gateway_health[cas.gateway_base_urls[1]] + for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): + health_b.record_failure(time.monotonic()) + + # gateway_a serves this, so gateway_b is ranked but never attempted. + assert await cas.load(GOOD_CID) == BODY + assert gateway_b.hits == [] + assert not health_b.probe_in_flight, "probe slot leaked on an unused gateway" + + # gateway_b is therefore still reachable on a later read. + gateway_a.responder = lambda _cid: (500, b"") + assert await cas.load(GOOD_CID) == BODY + assert len(gateway_b.hits) == 1 + + # --------------------------------------------------------------------------- # # per-gateway concurrency # # --------------------------------------------------------------------------- # @@ -692,6 +750,95 @@ async def test_client_level_auth_is_also_withheld( assert "authorization" not in gateway_b.headers_seen[0] +@pytest.mark.asyncio +async def test_redirect_to_a_foreign_origin_does_not_carry_credentials( + gateway_a: FakeGateway, gateway_b: FakeGateway +) -> None: + """A gateway must not be able to harvest credentials via a redirect. + + httpx's own redirect handling strips only ``Authorization`` and ``Cookie`` + across origins, so custom credentials would survive a 302 that the gateway + itself chooses the target of -- bypassing the origin check entirely. This + affects single-gateway users too, not just multi-gateway ones. + """ + + # gateway_a is the configured (credentialed) gateway; it redirects to + # gateway_b, which is a different origin and must receive nothing. + def redirect(cid: str) -> tuple[int, bytes]: + return (302, b"") + + gateway_a.responder = redirect + gateway_a.redirect_to = f"{gateway_b.url}/ipfs/" + + async with KuboCAS( + gateway_base_urls=[gateway_a.url], + rpc_base_url=gateway_a.url, + max_retries=0, + headers={ + "Authorization": "Bearer SECRET", + "Cookie": "session=abc", + "X-API-Key": "key-123", + }, + ) as cas: + assert await cas.load(GOOD_CID) == BODY + + primary = gateway_a.headers_seen[0] + target = gateway_b.headers_seen[0] + assert primary["authorization"] == "Bearer SECRET" + assert primary["x-api-key"] == "key-123" + for name in ("authorization", "cookie", "x-api-key"): + assert name not in target, f"{name} leaked across a redirect" + + +@pytest.mark.asyncio +async def test_redirect_within_the_credentialed_origin_keeps_credentials( + gateway_a: FakeGateway, +) -> None: + """Same-origin redirects must not lose the credentials they need.""" + hits = {"n": 0} + + def redirect_once(_cid: str) -> tuple[int, bytes]: + hits["n"] += 1 + return (302, b"") if hits["n"] == 1 else (200, BODY) + + gateway_a.responder = redirect_once + gateway_a.redirect_to = f"{gateway_a.url}/ipfs/" + + async with KuboCAS( + gateway_base_urls=[gateway_a.url], + rpc_base_url=gateway_a.url, + max_retries=0, + headers={"X-API-Key": "key-123"}, + ) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert len(gateway_a.headers_seen) == 2 + assert all(seen["x-api-key"] == "key-123" for seen in gateway_a.headers_seen) + + +@pytest.mark.asyncio +async def test_redirect_loop_is_bounded(gateway_a: FakeGateway) -> None: + """Manual redirect following must still honour a redirect limit. + + Following hops ourselves means httpx's own cap no longer applies, so a + gateway that redirects to itself forever would otherwise hang the read. + """ + gateway_a.responder = lambda _cid: (302, b"") + gateway_a.redirect_to = f"{gateway_a.url}/ipfs/" + + async with KuboCAS( + gateway_base_urls=[gateway_a.url], + rpc_base_url=gateway_a.url, + max_retries=0, + headers={"X-API-Key": "key-123"}, + ) as cas: + with pytest.raises(httpx.TooManyRedirects): + await cas.load(GOOD_CID) + + # Bounded, not infinite: it gave up rather than looping forever. + assert len(gateway_a.hits) <= 21 + + def test_forwardable_headers_is_an_allowlist_of_non_credentials() -> None: """The forwarding rule must be allow-by-name, not deny-by-name. @@ -810,13 +957,49 @@ def test_dag_pb_cids_are_not_verifiable() -> None: assert store_module._cid_is_verifiable(identity_cid, None, None) -def test_unsupported_hash_function_is_not_a_mismatch( - monkeypatch: pytest.MonkeyPatch, +def test_uncomputable_hash_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + """Verification that cannot run must raise, not pass the bytes through. + + Returning unverified content here would be the worst outcome for the + feature: a caller who asked for ``verify_content`` would receive arbitrary + gateway bytes indistinguishable from checked ones, precisely when the + hashing backend is broken or the algorithm is unavailable. + """ + + def unsupported(*_args: object, **_kwargs: object) -> bytes: + raise KeyError("unsupported multihash") + + monkeypatch.setattr(store_module.multihash, "digest", unsupported) + with pytest.raises(store_module.GatewayContentUnverifiable): + store_module._verify_cid_content(GOOD_CID, b"anything at all") + + +def test_unverifiable_is_caught_as_a_content_mismatch() -> None: + """It must subclass GatewayContentMismatch so failover still catches it.""" + assert issubclass(store_module.GatewayContentUnverifiable, GatewayContentMismatch) + + +@pytest.mark.asyncio +async def test_uncomputable_hash_fails_over_then_raises( + gateway_a: FakeGateway, gateway_b: FakeGateway, monkeypatch: pytest.MonkeyPatch ) -> None: - """We cannot prove content wrong with a hash we cannot compute.""" + """An unverifiable read is a per-gateway failure, not a silent success.""" def unsupported(*_args: object, **_kwargs: object) -> bytes: raise KeyError("unsupported multihash") monkeypatch.setattr(store_module.multihash, "digest", unsupported) - store_module._verify_cid_content(GOOD_CID, b"anything at all") + + async with make_cas( + gateway_a, gateway_b, verify_content=True, max_retries=0 + ) as cas: + with pytest.raises(ExceptionGroup) as excinfo: + await cas.load(GOOD_CID) + + # Both gateways were tried; neither could be verified, so neither won. + assert len(gateway_a.hits) == 1 + assert len(gateway_b.hits) == 1 + assert all( + isinstance(exc, store_module.GatewayContentUnverifiable) + for exc in excinfo.value.exceptions + ) From 8320022bb0de3cfc3630e9ea8dd4213ec35a751a Mon Sep 17 00:00:00 2001 From: TheGreatAlgo <37487508+TheGreatAlgo@users.noreply.github.com> Date: Mon, 27 Jul 2026 11:24:19 -0400 Subject: [PATCH 5/5] fix: final changes --- py_hamt/store_httpx.py | 61 ++++++++++--------- tests/test_k15_multi_gateway_failover.py | 75 +++++++----------------- 2 files changed, 51 insertions(+), 85 deletions(-) diff --git a/py_hamt/store_httpx.py b/py_hamt/store_httpx.py index 8842d62..f5ea6a6 100644 --- a/py_hamt/store_httpx.py +++ b/py_hamt/store_httpx.py @@ -69,10 +69,22 @@ def _carries_credentials(client: httpx.AsyncClient) -> bool: ) +# Ports implied by a scheme, so ``https://h`` and ``https://h:443`` compare +# equal rather than looking like two different origins. +_DEFAULT_PORTS = {"http": 80, "https": 443} + + def _origin_of(url: str) -> tuple[str, str, int | None]: - """Scheme/host/port triple used to decide if two URLs share an origin.""" + """Scheme/host/port triple used to decide if two URLs share an origin. + + The port falls back to the scheme default so a gateway written without a + port in config and reached with an explicit one (or vice versa) is treated + as the same origin, rather than silently losing its credentials. + """ parsed = urlsplit(url) - return (parsed.scheme.lower(), (parsed.hostname or "").lower(), parsed.port) + scheme = parsed.scheme.lower() + port = parsed.port or _DEFAULT_PORTS.get(scheme) + return (scheme, (parsed.hostname or "").lower(), port) def _normalize_gateway_base_url(gateway_base_url: str) -> str: @@ -103,45 +115,41 @@ class _GatewayHealth: state is touched from a second loop. """ - __slots__ = ("consecutive_failures", "probe_in_flight", "tripped_at") + __slots__ = ("consecutive_failures", "tripped_at") def __init__(self) -> None: self.consecutive_failures: int = 0 self.tripped_at: float | None = None - # True between handing out a post-cooldown probe slot and learning how - # that probe went. Keeps the gateway closed to everyone else meanwhile. - self.probe_in_flight: bool = False def record_success(self) -> None: self.consecutive_failures = 0 self.tripped_at = None - self.probe_in_flight = False def record_failure(self, now: float) -> None: self.consecutive_failures += 1 - self.probe_in_flight = False if self.consecutive_failures >= _GATEWAY_FAILURE_THRESHOLD: self.tripped_at = now def is_healthy(self, now: float) -> bool: - """Whether this gateway should be preferred, claiming a probe if due. - - Half-open, not merely time-based: exactly one caller past the cooldown - is let through as a probe, and the gateway stays deprioritized for - everyone else until that probe reports back via ``record_success`` or - ``record_failure``. Clearing the trip on a timer instead would let every - concurrent load in at once and dogpile a gateway that is still down. + """Whether this gateway should still be preferred. + + Deliberately time-based rather than a single-probe half-open gate. Once + the cooldown elapses, concurrent loads may all retry a gateway that is + still down -- but that costs a handful of wasted requests once per + cooldown, and each still fails over. Gating it behind one in-flight + probe means tracking that probe's outcome across cancellation and + never-attempted gateways, which is materially more machinery than the + thundering herd it avoids is worth here. """ if self.tripped_at is None: return True - if self.probe_in_flight: - # A probe is already out; nobody else gets through on its coattails. - return False if now - self.tripped_at >= _GATEWAY_COOLDOWN_SECONDS: - # Claim the probe slot. tripped_at stays set so that if this probe - # fails, the gateway remains tripped without needing to re-cross the - # failure threshold; record_failure refreshes the cooldown window. - self.probe_in_flight = True + # Cooldown elapsed. Clear the trip so a single probe failure does + # not immediately re-trip on a stale counter, but keep the gateway + # on probation by leaving the failure count one short of the + # threshold: one more failure re-trips it right away. + self.tripped_at = None + self.consecutive_failures = _GATEWAY_FAILURE_THRESHOLD - 1 return True return False @@ -1345,11 +1353,9 @@ async def load( stats = _LoadStats() gateways = self._ordered_gateways() failures: list[Exception] = [] - attempted: set[str] = set() try: for gateway_base_url in gateways: health = self._gateway_health[gateway_base_url] - attempted.add(gateway_base_url) try: content = await self._load_from_gateway( gateway_base_url, cid, headers, offset, length, suffix, stats @@ -1381,13 +1387,6 @@ async def load( f"all {len(gateways)} gateways failed for CID {cid}", failures ) finally: - # Ranking claims a probe slot for any tripped gateway whose cooldown - # has elapsed, but an earlier gateway usually succeeds first and the - # rest are never tried. Release those unused claims, or the slot is - # never reported back and the gateway is locked out permanently. - for gateway_base_url in gateways: - if gateway_base_url not in attempted: - self._gateway_health[gateway_base_url].probe_in_flight = False instrumentation.end_cas_load( trace_started_at, byte_count=stats.response_bytes, diff --git a/tests/test_k15_multi_gateway_failover.py b/tests/test_k15_multi_gateway_failover.py index d3120eb..54fd448 100644 --- a/tests/test_k15_multi_gateway_failover.py +++ b/tests/test_k15_multi_gateway_failover.py @@ -360,60 +360,6 @@ def test_health_cooldown_leaves_gateway_on_probation() -> None: assert not health.is_healthy(now=store_module._GATEWAY_COOLDOWN_SECONDS + 1.0) -def test_only_one_probe_is_admitted_after_cooldown() -> None: - """Half-open, not merely time-based. - - Clearing the trip on a timer would let every concurrent load through at - once and dogpile a gateway that is still down. Exactly one caller gets a - probe slot; the rest stay deprioritized until it reports back. - """ - health = store_module._GatewayHealth() - for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): - health.record_failure(now=0.0) - after = store_module._GATEWAY_COOLDOWN_SECONDS + 1.0 - - assert health.is_healthy(now=after), "the first caller should get the probe" - assert [health.is_healthy(now=after) for _ in range(8)] == [False] * 8 - - # A failed probe keeps it tripped without re-crossing the threshold. - health.record_failure(now=after) - assert not health.is_healthy(now=after + 1.0) - - # A successful probe restores it for everyone. - later = after + store_module._GATEWAY_COOLDOWN_SECONDS + 1.0 - assert health.is_healthy(now=later) - health.record_success() - assert all(health.is_healthy(now=later) for _ in range(5)) - - -@pytest.mark.asyncio -async def test_unused_probe_claim_is_released( - gateway_a: FakeGateway, gateway_b: FakeGateway, monkeypatch: pytest.MonkeyPatch -) -> None: - """A gateway ranked but never tried must not keep its probe slot. - - Ranking claims the slot, but the loop stops at the first success, so a - lower-ranked gateway is usually never attempted. Without releasing the - claim it would never report back and would be locked out forever. - """ - monkeypatch.setattr(store_module, "_GATEWAY_COOLDOWN_SECONDS", 0.0) - - async with make_cas(gateway_a, gateway_b, max_retries=0) as cas: - health_b = cas._gateway_health[cas.gateway_base_urls[1]] - for _ in range(store_module._GATEWAY_FAILURE_THRESHOLD): - health_b.record_failure(time.monotonic()) - - # gateway_a serves this, so gateway_b is ranked but never attempted. - assert await cas.load(GOOD_CID) == BODY - assert gateway_b.hits == [] - assert not health_b.probe_in_flight, "probe slot leaked on an unused gateway" - - # gateway_b is therefore still reachable on a later read. - gateway_a.responder = lambda _cid: (500, b"") - assert await cas.load(GOOD_CID) == BODY - assert len(gateway_b.hits) == 1 - - # --------------------------------------------------------------------------- # # per-gateway concurrency # # --------------------------------------------------------------------------- # @@ -839,6 +785,27 @@ async def test_redirect_loop_is_bounded(gateway_a: FakeGateway) -> None: assert len(gateway_a.hits) <= 21 +def test_origin_comparison_normalizes_default_ports() -> None: + """``https://h`` and ``https://h:443`` are one origin, not two. + + Otherwise a gateway written without a port in config but reached with an + explicit one would look foreign and silently lose its credentials. + """ + assert store_module._origin_of("https://h/ipfs/") == store_module._origin_of( + "https://h:443/ipfs/" + ) + assert store_module._origin_of("http://h/ipfs/") == store_module._origin_of( + "http://h:80/ipfs/" + ) + # Genuinely different origins still differ. + assert store_module._origin_of("https://h/ipfs/") != store_module._origin_of( + "http://h/ipfs/" + ) + assert store_module._origin_of("https://h/ipfs/") != store_module._origin_of( + "https://h:8443/ipfs/" + ) + + def test_forwardable_headers_is_an_allowlist_of_non_credentials() -> None: """The forwarding rule must be allow-by-name, not deny-by-name.