diff --git a/py_hamt/__init__.py b/py_hamt/__init__.py index 8bb5d59..a54e699 100644 --- a/py_hamt/__init__.py +++ b/py_hamt/__init__.py @@ -6,13 +6,21 @@ ShardedZarrV1DeprecationWarning, ShardReadMode, ) -from .store_httpx import ContentAddressedStore, InMemoryCAS, KuboCAS +from .store_httpx import ( + ContentAddressedStore, + GatewayContentMismatch, + GatewayContentUnverifiable, + InMemoryCAS, + KuboCAS, +) from .zarr_hamt_store import ZarrHAMTStore __all__ = [ "blake3_hashfn", "HAMT", "ContentAddressedStore", + "GatewayContentMismatch", + "GatewayContentUnverifiable", "InMemoryCAS", "KuboCAS", "ZarrHAMTStore", diff --git a/py_hamt/store_httpx.py b/py_hamt/store_httpx.py index ae0e950..f5ea6a6 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 @@ -24,6 +26,133 @@ # 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 + +# 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 _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 + ) + + +# 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. + + 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) + 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: + """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. + + 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") + + 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: + """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 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 +184,99 @@ 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. + """ + + +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. + + 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 + # 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``. + + 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. + + 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": + # 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, size=len(raw_digest)) + 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}" + ) + + def _slice_requested_range( data: bytes, offset: Optional[int], @@ -333,6 +555,26 @@ 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. + 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---"`. @@ -355,6 +597,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 +674,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 +693,47 @@ 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""" + # 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 + # 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 +814,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 +824,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 = 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]] = {} + 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 +885,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 +1001,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 +1034,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 +1057,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 +1111,199 @@ 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, + 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. + + 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 + + while retry_count <= self.max_retries: + try: + async with semaphore: # Throttle each gateway attempt + 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. + 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. + try: + _verify_cid_content(cid, content) + except GatewayContentMismatch: + stats.status = "content_mismatch" + raise + 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 +1319,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 +1350,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(time.monotonic()) + 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..54fd448 --- /dev/null +++ b/tests/test_k15_multi_gateway_failover.py @@ -0,0 +1,972 @@ +"""Multi-gateway failover, per-gateway concurrency, and content verification.""" + +import asyncio +import threading +import time +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 +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. +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) + # 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) + # 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) + + +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) + # 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: + status, body = gw.responder(cid) + finally: + 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: + 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 + + # 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( + 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] + + +@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: TraceContext | None, *, 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: TraceContext | None, *, 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"") + # 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] + 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 +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", "X-API-Key": "key-123"}, + ) as cas: + assert await cas.load(GOOD_CID) == BODY + + assert len(gateway_b.headers_seen) == 3 + for seen in gateway_b.headers_seen: + assert "authorization" not in seen + assert "x-api-key" not in 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_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_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. + + 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, +) -> 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")) + 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) + + # 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_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: + """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) + + 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 + )