diff --git a/src/adcp/canonical_formats/references.py b/src/adcp/canonical_formats/references.py index 9f15643d..292a7db4 100644 --- a/src/adcp/canonical_formats/references.py +++ b/src/adcp/canonical_formats/references.py @@ -42,6 +42,7 @@ DEFAULT_REFERENCE_TIMEOUT_SECONDS = 5.0 DEFAULT_REFERENCE_BODY_LIMIT_BYTES = 1024 * 1024 DEFAULT_MAX_SCHEMA_REFS = 256 +DEFAULT_MAX_SCHEMA_IDS = 32 DEFAULT_MAX_REF_DEPTH = 8 DEFAULT_MAX_SCHEMA_KEYWORDS = 10_000 DEFAULT_MAX_SCHEMA_DEPTH = 128 @@ -157,6 +158,7 @@ def __init__( timeout: float = DEFAULT_REFERENCE_TIMEOUT_SECONDS, max_body_bytes: int = DEFAULT_REFERENCE_BODY_LIMIT_BYTES, max_schema_refs: int = DEFAULT_MAX_SCHEMA_REFS, + max_schema_ids: int = DEFAULT_MAX_SCHEMA_IDS, max_ref_depth: int = DEFAULT_MAX_REF_DEPTH, max_schema_keywords: int = DEFAULT_MAX_SCHEMA_KEYWORDS, max_schema_depth: int = DEFAULT_MAX_SCHEMA_DEPTH, @@ -165,6 +167,7 @@ def __init__( self._timeout = timeout self._max_body_bytes = max_body_bytes self._max_schema_refs = max_schema_refs + self._max_schema_ids = max_schema_ids self._max_ref_depth = max_ref_depth self._max_schema_keywords = max_schema_keywords self._max_schema_depth = max_schema_depth @@ -318,6 +321,7 @@ def _validate_schema( document, base_uri=reference.uri, max_refs=self._max_schema_refs, + max_ids=self._max_schema_ids, max_ref_depth=self._max_ref_depth, max_keywords=self._max_schema_keywords, max_depth=self._max_schema_depth, @@ -534,11 +538,12 @@ def _validate_schema_refs( *, base_uri: str, max_refs: int, + max_ids: int, max_ref_depth: int, max_keywords: int, max_depth: int, ) -> str | None: - state = _SchemaRefState(max_refs=max_refs, max_keywords=max_keywords) + state = _SchemaRefState(max_refs=max_refs, max_ids=max_ids, max_keywords=max_keywords) return _walk_schema_refs( document, base_uri=base_uri, @@ -552,8 +557,10 @@ def _validate_schema_refs( @dataclass class _SchemaRefState: max_refs: int + max_ids: int max_keywords: int refs: int = 0 + ids: int = 0 keywords: int = 0 @@ -577,6 +584,9 @@ def _walk_schema_refs( if raw_id is not None: if not isinstance(raw_id, str): return "$id must be a string" + state.ids += 1 + if state.ids > state.max_ids: + return "format_schema exceeds $id count bound" id_error, resolved_id = _validate_schema_id_value(raw_id, base_uri=base_uri) if id_error is not None: return id_error @@ -711,5 +721,6 @@ def _trusted_aao_catalog_origin(parts: SplitResult) -> bool: "CanonicalReferenceStatus", "DEFAULT_REFERENCE_BODY_LIMIT_BYTES", "DEFAULT_REFERENCE_TIMEOUT_SECONDS", + "DEFAULT_MAX_SCHEMA_IDS", "parse_canonical_reference", ] diff --git a/src/adcp/server/idempotency/backends.py b/src/adcp/server/idempotency/backends.py index c74e07d8..ea1f9b01 100644 --- a/src/adcp/server/idempotency/backends.py +++ b/src/adcp/server/idempotency/backends.py @@ -4,12 +4,11 @@ 1. Retrieve a cached response by ``(principal_id, idempotency_key)``, honoring the seller's replay TTL. -2. Atomically commit ``(payload_hash, response)`` on a fresh key. Atomicity - with the handler's business writes is the backend's choice — - :class:`MemoryBackend` makes no such guarantee; :class:`PgBackend` shares - a connection pool so adopters with the same Postgres can compose their - handler's transaction with the cache write (v1 commits in a separate - pool connection — co-tx wiring is a v1.1 affordance). +2. Hold an execution lock across lookup, handler execution, and cache commit. +3. Atomically insert webhook-dedup markers with first-writer-wins semantics. + +The lock prevents concurrent duplicate execution. Atomicity with unrelated +business writes remains the adopter's responsibility. Backends expose async methods. The in-process :class:`MemoryBackend` is synchronous under the hood but wrapped in ``async`` signatures so the store @@ -22,8 +21,11 @@ import json import re import time +import weakref from abc import ABC, abstractmethod -from collections.abc import Callable +from collections.abc import AsyncIterator, Callable +from contextlib import asynccontextmanager +from contextvars import ContextVar from dataclasses import dataclass from datetime import datetime, timezone from typing import Any @@ -83,8 +85,8 @@ class IdempotencyBackend(ABC): """Abstract storage backend contract. All methods are async. Implementations MUST be safe to call concurrently - from multiple asyncio tasks — :class:`IdempotencyStore` does not serialize - access on the caller's behalf. + from multiple asyncio tasks. ``hold`` must coordinate all processes that + share the backend namespace. """ @abstractmethod @@ -109,6 +111,30 @@ async def put( or expired, so an overwrite in that window is a legitimate retry of the write itself.""" + def hold(self, scope_key: str, key: str) -> Any: + """Return an async context manager holding the key's execution lock. + + Custom backends must implement this operation to be usable by + :class:`IdempotencyStore`. It is concrete only to keep existing + backend subclasses importable while adopters migrate. + """ + raise NotImplementedError( + f"{type(self).__name__} must implement atomic hold(scope_key, key)" + ) + + async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool: + """Atomically insert a fresh/expired slot; return whether it won. + + The default composes ``hold`` with the legacy ``get``/``put`` API, + allowing custom backends to implement one locking primitive. Backends + with a native conditional insert should override this method. + """ + async with self.hold(scope_key, key): + if await self.get(scope_key, key) is not None: + return False + await self.put(scope_key, key, entry) + return True + @abstractmethod async def delete_expired(self, now_epoch: float | None = None) -> int: """Best-effort sweep of expired entries. Returns the count removed. @@ -139,6 +165,9 @@ class MemoryBackend(IdempotencyBackend): def __init__(self, *, clock: Callable[[], float] = time.time) -> None: self._store: dict[tuple[str, str], CachedResponse] = {} self._lock = asyncio.Lock() + self._key_locks: weakref.WeakValueDictionary[tuple[str, str], asyncio.Lock] = ( + weakref.WeakValueDictionary() + ) self._clock = clock async def get(self, scope_key: str, key: str) -> CachedResponse | None: @@ -162,6 +191,28 @@ async def put( async with self._lock: self._store[(scope_key, key)] = entry + @asynccontextmanager + async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]: + """Serialize one idempotent handler execution in this process.""" + slot = (scope_key, key) + async with self._lock: + key_lock = self._key_locks.get(slot) + if key_lock is None: + key_lock = asyncio.Lock() + self._key_locks[slot] = key_lock + async with key_lock: + yield + + async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool: + """Atomically claim a missing or expired slot.""" + slot = (scope_key, key) + async with self._lock: + existing = self._store.get(slot) + if existing is not None and existing.expires_at_epoch > self._clock(): + return False + self._store[slot] = entry + return True + async def delete_expired(self, now_epoch: float | None = None) -> int: cutoff = now_epoch if now_epoch is not None else self._clock() async with self._lock: @@ -198,15 +249,17 @@ class PgBackend(IdempotencyBackend): from adcp.server.idempotency import IdempotencyStore, PgBackend pool = AsyncConnectionPool("postgresql://...", min_size=2, max_size=10) - backend = PgBackend(pool=pool) + lock_pool = AsyncConnectionPool("postgresql://...", min_size=2, max_size=10) + backend = PgBackend(pool=pool, lock_pool=lock_pool) await backend.create_schema() # idempotent; safe to call on every boot store = IdempotencyStore(backend=backend, ttl_seconds=86400) - **Atomicity caveat (v1).** ``put`` commits on a fresh pool connection — - the cache write is NOT in the same transaction as the handler's - business writes. A crash between handler success and cache commit - leaves the slot empty; the next retry re-executes the handler. + **Atomicity caveat.** The backend holds a Postgres advisory lock and writes + the cache in its transaction, but the cache write is NOT automatically in + the same transaction as the handler's unrelated business writes. A crash + after an external side effect but before cache commit can still leave the + slot empty; the next retry re-executes the handler. Idempotent handlers absorb this without harm. **Handlers with non-idempotent side effects** (e.g., ``INSERT INTO media_buys`` without a unique constraint on the buyer's idempotency_key) need @@ -261,6 +314,10 @@ class PgBackend(IdempotencyBackend): :param pool: ``psycopg_pool.AsyncConnectionPool`` owned by the caller. Each operation acquires a short-lived connection. We don't open, own, or close the pool. + :param lock_pool: A distinct caller-owned pool reserved for advisory-lock + transactions. It MUST NOT be the business/cache ``pool``: ``hold`` + keeps one connection checked out while adopter code runs, and sharing + that pool with handler SQL can deadlock under saturation. :param table_name: Override the default table name. Useful for multi-tenant schema scoping. Default ``adcp_idempotency``. @@ -274,12 +331,19 @@ def __init__( self, *, pool: Any, # psycopg_pool.AsyncConnectionPool — Any avoids runtime psycopg import + lock_pool: Any, table_name: str = DEFAULT_IDEMPOTENCY_TABLE, ) -> None: if not _PG_AVAILABLE: raise ImportError(_PG_INSTALL_HINT) + if lock_pool is pool: + raise ValueError("lock_pool must be distinct from pool to prevent handler deadlocks") self._pool = pool + self._lock_pool = lock_pool self._table = _safe_identifier(table_name) + self._active_connection: ContextVar[tuple[Any, asyncio.Task[Any] | None] | None] = ( + ContextVar(f"adcp_idempotency_connection_{id(self)}", default=None) + ) # Pre-format SQL once. Validated identifier so f-string interpolation # is byte-safe; values always go through %s parameterization. Same @@ -309,6 +373,16 @@ def __init__( f"WHERE {t}.expires_at <= now()" ) self._sql_delete_expired = f"DELETE FROM {t} WHERE expires_at <= %s" # noqa: S608 + self._sql_lock = "SELECT pg_advisory_xact_lock(hashtextextended(%s, 6217))" + self._sql_put_if_absent = ( + f"INSERT INTO {t} " # noqa: S608 + f"(scope_key, key, payload_hash, response, expires_at) " + f"VALUES (%s, %s, %s, %s::jsonb, %s) " + f"ON CONFLICT (scope_key, key) DO UPDATE SET " + f" payload_hash = EXCLUDED.payload_hash, response = EXCLUDED.response, " + f" expires_at = EXCLUDED.expires_at " + f"WHERE {t}.expires_at <= now() RETURNING 1" + ) async def create_schema(self) -> None: """Bootstrap the table + index. Idempotent. @@ -341,17 +415,25 @@ async def get(self, scope_key: str, key: str) -> CachedResponse | None: sweeps them. ``get`` self-filters via ``expires_at > now()`` so a stale row never replays. """ + active = self._active_connection.get() + if active is not None and active[1] is asyncio.current_task(): + return await self._get_on_connection(active[0], scope_key, key) async with self._pool.connection() as conn: - cur = await conn.execute(self._sql_get, (scope_key, key)) - row = await cur.fetchone() - if row is None: - return None - payload_hash, response, expires_at = row - return CachedResponse( - payload_hash=payload_hash, - response=response if isinstance(response, dict) else json.loads(response), - expires_at_epoch=_to_epoch(expires_at), - ) + return await self._get_on_connection(conn, scope_key, key) + + async def _get_on_connection( + self, conn: Any, scope_key: str, key: str + ) -> CachedResponse | None: + cur = await conn.execute(self._sql_get, (scope_key, key)) + row = await cur.fetchone() + if row is None: + return None + payload_hash, response, expires_at = row + return CachedResponse( + payload_hash=payload_hash, + response=response if isinstance(response, dict) else json.loads(response), + expires_at_epoch=_to_epoch(expires_at), + ) async def put( self, @@ -366,9 +448,42 @@ async def put( that window is a legitimate retry of the write itself. """ expires_at_dt = datetime.fromtimestamp(entry.expires_at_epoch, tz=timezone.utc) + params = ( + scope_key, + key, + entry.payload_hash, + json.dumps(entry.response), + expires_at_dt, + ) + active = self._active_connection.get() + if active is not None and active[1] is asyncio.current_task(): + await active[0].execute(self._sql_put, params) + return + async with self._pool.connection() as conn: + await conn.execute(self._sql_put, params) + + @asynccontextmanager + async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]: + """Hold a cross-process transaction advisory lock for this key. + + The same pooled connection remains checked out while the handler runs; + nested ``get``/``put`` calls reuse it via a context-local binding. + """ + lock_identity = json.dumps([scope_key, key], separators=(",", ":")) + async with self._lock_pool.connection() as conn, conn.transaction(): + await conn.execute(self._sql_lock, (lock_identity,)) + token = self._active_connection.set((conn, asyncio.current_task())) + try: + yield + finally: + self._active_connection.reset(token) + + async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool: + """Atomically insert a webhook dedup marker, including stale replace.""" + expires_at_dt = datetime.fromtimestamp(entry.expires_at_epoch, tz=timezone.utc) async with self._pool.connection() as conn: - await conn.execute( - self._sql_put, + cur = await conn.execute( + self._sql_put_if_absent, ( scope_key, key, @@ -377,6 +492,7 @@ async def put( expires_at_dt, ), ) + return await cur.fetchone() is not None async def delete_expired(self, now_epoch: float | None = None) -> int: """Best-effort sweep of expired entries. Returns rows removed.""" diff --git a/src/adcp/server/idempotency/lazy.py b/src/adcp/server/idempotency/lazy.py index 5aaa0e75..7ae4722e 100644 --- a/src/adcp/server/idempotency/lazy.py +++ b/src/adcp/server/idempotency/lazy.py @@ -13,7 +13,8 @@ async def _resolve() -> IdempotencyBackend: pool = await app.get_pg_pool() - backend = PgBackend(pool=pool) + lock_pool = await app.get_idempotency_lock_pool() + backend = PgBackend(pool=pool, lock_pool=lock_pool) await backend.create_schema() return backend @@ -33,7 +34,8 @@ async def _resolve() -> IdempotencyBackend: from __future__ import annotations import asyncio -from collections.abc import Awaitable, Callable +from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import asynccontextmanager from adcp.server.idempotency.backends import CachedResponse, IdempotencyBackend @@ -108,6 +110,14 @@ async def get(self, scope_key: str, key: str) -> CachedResponse | None: async def put(self, scope_key: str, key: str, entry: CachedResponse) -> None: await (await self._resolve()).put(scope_key, key, entry) + @asynccontextmanager + async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]: + async with (await self._resolve()).hold(scope_key, key): + yield + + async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool: + return await (await self._resolve()).put_if_absent(scope_key, key, entry) + async def delete_expired(self, now_epoch: float | None = None) -> int: return await (await self._resolve()).delete_expired(now_epoch) diff --git a/src/adcp/server/idempotency/store.py b/src/adcp/server/idempotency/store.py index fbd55883..13e0c3fc 100644 --- a/src/adcp/server/idempotency/store.py +++ b/src/adcp/server/idempotency/store.py @@ -28,13 +28,15 @@ from __future__ import annotations +import asyncio import copy import hashlib import logging import time import warnings import weakref -from collections.abc import Awaitable, Callable +from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import asynccontextmanager from functools import wraps from typing import Any @@ -59,6 +61,25 @@ # only granted by IdempotencyStore.wrap itself. _WRAPPED_FUNCTIONS: weakref.WeakSet[Callable[..., Any]] = weakref.WeakSet() +# Keyed idempotency operations continue after the request task is cancelled: +# cancellation cannot stop a synchronous adopter thread, and dropping the lock +# early would allow a retry to execute concurrently. Keep strong references +# until these detached operations settle. +_SUPERVISED_OPERATIONS: set[asyncio.Task[Any]] = set() + + +def _finish_supervised_operation(task: asyncio.Task[Any]) -> None: + """Drop a detached operation and consume/log its terminal exception.""" + _SUPERVISED_OPERATIONS.discard(task) + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + logger.error( + "Idempotency operation failed after its request was cancelled", + exc_info=(type(exc), exc, exc.__traceback__), + ) + def is_wrapped(fn: Any) -> bool: """Return True if ``fn`` was produced by :meth:`IdempotencyStore.wrap`. @@ -111,6 +132,37 @@ def __init__( self.ttl_seconds = ttl_seconds self._hash_fn = hash_fn self._clock = clock + self._fallback_guard = asyncio.Lock() + self._fallback_locks: weakref.WeakValueDictionary[tuple[str, str], asyncio.Lock] = ( + weakref.WeakValueDictionary() + ) + self._warned_fallback_hold = False + + @asynccontextmanager + async def _hold(self, scope_key: str, key: str) -> AsyncIterator[None]: + """Use native backend locking or a deprecated process-local fallback.""" + if type(self.backend).hold is not IdempotencyBackend.hold: + async with self.backend.hold(scope_key, key): + yield + return + + if not self._warned_fallback_hold: + warnings.warn( + f"{type(self.backend).__name__} does not implement hold(); using " + "process-local idempotency locking. Implement hold() for " + "cross-process atomicity; this compatibility fallback is deprecated.", + DeprecationWarning, + stacklevel=3, + ) + self._warned_fallback_hold = True + slot = (scope_key, key) + async with self._fallback_guard: + lock = self._fallback_locks.get(slot) + if lock is None: + lock = asyncio.Lock() + self._fallback_locks[slot] = lock + async with lock: + yield def capability(self) -> dict[str, Any]: """Return the capabilities fragment declaring this store's replay window. @@ -178,75 +230,70 @@ async def _wrapped(*args: Any, **kwargs: Any) -> Any: payload_hash = self._hash_fn(params_dict) - cached = await self.backend.get(scope_key, idempotency_key) - if cached is not None: - if cached.payload_hash == payload_hash: - logger.debug( - "idempotency replay: scope=%s key_prefix=%s", - _scope_log_id(scope_key), - idempotency_key[:8], + async def _execute_locked() -> Any: + # The backend lock spans lookup, handler execution, and commit. + # This is the critical invariant: a concurrent request for the + # same scoped key waits, then observes the winner's cached result. + async with self._hold(scope_key, idempotency_key): + cached = await self.backend.get(scope_key, idempotency_key) + if cached is not None: + if cached.payload_hash == payload_hash: + logger.debug( + "idempotency replay: scope=%s key_prefix=%s", + _scope_log_id(scope_key), + idempotency_key[:8], + ) + # AdCP L1/security idempotency rule 4: the replay + # envelope MUST carry ``replayed: true`` so buyer + # agents can suppress side effects on retry. + replay = _clone_response(cached.response) + replay["replayed"] = True + return replay + # Same key, different payload — spec-defined conflict. + raise IdempotencyConflictError( + operation=operation, + errors=[ + { + "code": "IDEMPOTENCY_CONFLICT", + "message": ( + "idempotency_key reused with a different payload " + "(canonical hash mismatch)" + ), + } + ], + ) + + response = await handler(*args, **kwargs) + # Deep-copy when caching so post-return mutation of the caller's + # copy can't poison future replays. `_clone_response` also deep- + # copies on the hit path, giving independent objects per replay. + response_dict = copy.deepcopy(_to_dict(response)) + entry = CachedResponse( + payload_hash=payload_hash, + response=response_dict, + expires_at_epoch=self._clock() + self.ttl_seconds, ) - # AdCP L1/security idempotency rule 4: the replay - # envelope MUST carry ``replayed: true`` so buyer - # agents can suppress side effects (notifications, - # webhook dispatch, memory writes) on retry. The - # store owns this — sellers can't inject at the - # right point (cache lookup happens here, wire - # serialization happens later). The injection - # lands on the cloned dict, not ``cached.response``, - # so multiple replays of the same key all carry - # exactly one ``replayed: true`` without compounding. - replay = _clone_response(cached.response) - replay["replayed"] = True - return replay - # Same key, different payload — spec-defined conflict. - raise IdempotencyConflictError( - operation=operation, - errors=[ - { - "code": "IDEMPOTENCY_CONFLICT", - "message": ( - "idempotency_key reused with a different payload " - "(canonical hash mismatch)" - ), - } - ], - ) - - response = await handler(*args, **kwargs) - # Deep-copy when caching so post-return mutation of the caller's - # copy can't poison future replays. `_clone_response` also deep- - # copies on the hit path, giving independent objects per replay. - response_dict = copy.deepcopy(_to_dict(response)) - entry = CachedResponse( - payload_hash=payload_hash, - response=response_dict, - expires_at_epoch=self._clock() + self.ttl_seconds, - ) - # Commit cache AFTER handler returns. Atomicity with the handler's - # side effects depends on the backend: MemoryBackend is best-effort - # (no transactional relationship to external resources); PgBackend - # (follow-up) will commit in the same transaction when the handler - # uses the same engine. On put failure we log loudly and return - # the handler's response — swallowing the exception would be wrong - # (operators need the signal that caching is broken), and raising - # would look to the caller like the handler failed, triggering a - # retry that re-executes side effects. Best compromise: warn - # operators, return the result, and accept that the next retry - # with this key will re-execute. - try: - await self.backend.put(scope_key, idempotency_key, entry) - except Exception: - logger.warning( - "Idempotency cache put failed for scope=%s key_prefix=%s — " - "handler completed but a subsequent retry with this key will " - "re-execute rather than replay. This indicates an operational " - "issue with the idempotency backend.", - _scope_log_id(scope_key), - idempotency_key[:8], - exc_info=True, - ) - return response + # Commit while the execution lock is still held. This does not + # make unrelated business writes transactional with the cache, + # but it prevents a concurrent duplicate handler execution. + try: + await self.backend.put(scope_key, idempotency_key, entry) + except Exception: + logger.warning( + "Idempotency cache put failed for scope=%s key_prefix=%s — " + "handler completed but a subsequent retry with this key will " + "re-execute rather than replay. This indicates an operational " + "issue with the idempotency backend.", + _scope_log_id(scope_key), + idempotency_key[:8], + exc_info=True, + ) + return response + + operation = asyncio.create_task(_execute_locked()) + _SUPERVISED_OPERATIONS.add(operation) + operation.add_done_callback(_finish_supervised_operation) + return await asyncio.shield(operation) # Register the wrapper for the boot-time validator at # adcp.decisioning.validate_idempotency. WeakSet membership — diff --git a/src/adcp/server/idempotency/webhook_dedup.py b/src/adcp/server/idempotency/webhook_dedup.py index 5d3eef01..52218008 100644 --- a/src/adcp/server/idempotency/webhook_dedup.py +++ b/src/adcp/server/idempotency/webhook_dedup.py @@ -25,8 +25,11 @@ from __future__ import annotations +import asyncio import logging import time +import warnings +import weakref from collections.abc import Callable from adcp.server.idempotency.backends import CachedResponse, IdempotencyBackend @@ -80,6 +83,46 @@ def __init__( self.ttl_seconds = ttl_seconds self.namespace = namespace self._clock = clock + self._fallback_guard = asyncio.Lock() + self._fallback_locks: weakref.WeakValueDictionary[tuple[str, str], asyncio.Lock] = ( + weakref.WeakValueDictionary() + ) + self._warned_legacy_backend = False + + async def _put_if_absent( + self, + scope_key: str, + key: str, + entry: CachedResponse, + ) -> bool: + """Use backend atomicity or a warned process-local legacy fallback.""" + has_native_put = type(self.backend).put_if_absent is not IdempotencyBackend.put_if_absent + has_native_hold = type(self.backend).hold is not IdempotencyBackend.hold + if has_native_put or has_native_hold: + return await self.backend.put_if_absent(scope_key, key, entry) + + if not self._warned_legacy_backend: + warnings.warn( + f"{type(self.backend).__name__} implements neither put_if_absent() " + "nor hold(); using process-local webhook dedup locking. Implement " + "an atomic operation for cross-process safety; this compatibility " + "fallback is deprecated.", + DeprecationWarning, + stacklevel=3, + ) + self._warned_legacy_backend = True + + slot = (scope_key, key) + async with self._fallback_guard: + lock = self._fallback_locks.get(slot) + if lock is None: + lock = asyncio.Lock() + self._fallback_locks[slot] = lock + async with lock: + if await self.backend.get(scope_key, key) is not None: + return False + await self.backend.put(scope_key, key, entry) + return True async def check_and_record(self, sender_id: str, idempotency_key: str) -> bool: """Atomically check for first-seen and record if new. @@ -88,18 +131,8 @@ async def check_and_record(self, sender_id: str, idempotency_key: str) -> bool: processed), ``False`` on duplicate (caller MUST still return 2xx to the sender — the event was delivered successfully, it's just a retry). - Race note: the check-then-put pattern is not atomic across concurrent - callers unless the backend provides its own atomicity. MemoryBackend - serializes individual ``get`` and ``put`` under an ``asyncio.Lock`` but - does NOT bracket them together — two concurrent retries of the same - event CAN both observe "first-seen" and both process the event. That's - a tolerable failure mode: the ultimate guarantee is "at most once per - replay window in the common case"; a concurrent retry arriving in the - same few milliseconds is rare and, if it happens, produces the same - "duplicated side effect" outcome the at-least-once contract already - warns callers to tolerate. PgBackend implementations SHOULD use - ``INSERT ... ON CONFLICT DO NOTHING`` returning ``rowcount`` for - lock-free atomicity. + The backend performs a single atomic insert-or-reject operation, so + concurrent deliveries cannot both be reported as first-seen. """ if not sender_id: raise ValueError("sender_id must be a non-empty string") @@ -107,22 +140,13 @@ async def check_and_record(self, sender_id: str, idempotency_key: str) -> bool: raise ValueError("idempotency_key must be a non-empty string") scoped_sender = f"{self.namespace}:{sender_id}" - existing = await self.backend.get(scoped_sender, idempotency_key) - if existing is not None: - logger.debug( - "webhook dedup: duplicate sender=%s key_prefix=%s", - sender_id, - idempotency_key[:8], - ) - return False - entry = CachedResponse( payload_hash=_SENTINEL_HASH, response={}, expires_at_epoch=self._clock() + self.ttl_seconds, ) try: - await self.backend.put(scoped_sender, idempotency_key, entry) + inserted = await self._put_if_absent(scoped_sender, idempotency_key, entry) except Exception: # Same fail-open reasoning as the request-side store: log and # process. Swallowing the put failure means this event MIGHT @@ -135,7 +159,14 @@ async def check_and_record(self, sender_id: str, idempotency_key: str) -> bool: idempotency_key[:8], exc_info=True, ) - return True + return True + if not inserted: + logger.debug( + "webhook dedup: duplicate sender=%s key_prefix=%s", + sender_id, + idempotency_key[:8], + ) + return inserted __all__ = ["WebhookDedupStore"] diff --git a/src/adcp/signing/__init__.py b/src/adcp/signing/__init__.py index e9ea8328..0d42ead2 100644 --- a/src/adcp/signing/__init__.py +++ b/src/adcp/signing/__init__.py @@ -241,7 +241,12 @@ SigningAlgorithm, SigningProvider, ) -from adcp.signing.replay import InMemoryReplayStore, ReplayStore +from adcp.signing.replay import ( + AtomicReplayStore, + InMemoryReplayStore, + ReplayClaimResult, + ReplayStore, +) from adcp.signing.revocation import RevocationChecker, RevocationList from adcp.signing.revocation_fetcher import ( DEFAULT_GRACE_MULTIPLIER, @@ -315,6 +320,7 @@ def __init__(self, *args: object, **kwargs: object) -> None: "AsyncJwksFetcher", "AsyncJwksResolver", "AsyncRevocationListFetcher", + "AtomicReplayStore", "BrandAgentType", "BrandAuthorizationReason", "BrandAuthorizationResolver", @@ -377,6 +383,7 @@ def __init__(self, *args: object, **kwargs: object) -> None: "REQUEST_SIGNATURE_WINDOW_INVALID", "REVOCATION_LIST_TYP", "ReplayStore", + "ReplayClaimResult", "RevocationChecker", "RevocationList", "RevocationListFetchError", diff --git a/src/adcp/signing/_bounded_http.py b/src/adcp/signing/_bounded_http.py new file mode 100644 index 00000000..1fdaa344 --- /dev/null +++ b/src/adcp/signing/_bounded_http.py @@ -0,0 +1,57 @@ +"""Small streaming body readers shared by signing discovery fetchers.""" + +from __future__ import annotations + +import httpx + + +class ResponseTooLargeError(ValueError): + """A decoded HTTP response exceeded its configured byte budget.""" + + def __init__(self, *, limit: int, received: int) -> None: + super().__init__(f"response exceeds {limit} bytes (received at least {received})") + self.limit = limit + self.received = received + + +def read_limited_bytes(response: httpx.Response, *, limit: int) -> bytes: + """Stream at most ``limit`` decoded bytes from ``response``.""" + content_length = response.headers.get("content-length") + if content_length is not None: + try: + declared = int(content_length) + except ValueError: + declared = 0 + if declared > limit: + raise ResponseTooLargeError(limit=limit, received=declared) + + body = bytearray() + chunk_size = max(1, min(64 * 1024, limit + 1)) + for chunk in response.iter_bytes(chunk_size=chunk_size): + body.extend(chunk) + if len(body) > limit: + raise ResponseTooLargeError(limit=limit, received=len(body)) + return bytes(body) + + +async def async_read_limited_bytes(response: httpx.Response, *, limit: int) -> bytes: + """Async counterpart to :func:`read_limited_bytes`.""" + content_length = response.headers.get("content-length") + if content_length is not None: + try: + declared = int(content_length) + except ValueError: + declared = 0 + if declared > limit: + raise ResponseTooLargeError(limit=limit, received=declared) + + body = bytearray() + chunk_size = max(1, min(64 * 1024, limit + 1)) + async for chunk in response.aiter_bytes(chunk_size=chunk_size): + body.extend(chunk) + if len(body) > limit: + raise ResponseTooLargeError(limit=limit, received=len(body)) + return bytes(body) + + +__all__ = ["ResponseTooLargeError", "async_read_limited_bytes", "read_limited_bytes"] diff --git a/src/adcp/signing/agent_resolver.py b/src/adcp/signing/agent_resolver.py index 41edf005..8078c328 100644 --- a/src/adcp/signing/agent_resolver.py +++ b/src/adcp/signing/agent_resolver.py @@ -39,6 +39,7 @@ from __future__ import annotations import asyncio +import hashlib import json import time from collections.abc import Callable @@ -49,6 +50,7 @@ import httpx from pydantic import BaseModel, ConfigDict, Field +from adcp.signing._bounded_http import ResponseTooLargeError, async_read_limited_bytes from adcp.signing.brand_jwks import ( BrandAgentType, BrandJsonJwksResolver, @@ -60,6 +62,7 @@ StaticJwksResolver, async_default_jwks_fetcher, ) +from adcp.signing.replay import InMemoryReplayStore, ReplayClaimResult, ReplayStore #: Maximum capabilities response body in bytes. Capabilities documents #: are larger than brand.json (operator/agent declarations, supported @@ -230,7 +233,54 @@ async def _fetch_capabilities( try: async with client_cm as client: try: - response = await client.get(url, headers={"accept": "application/json"}) + request_cm = client.stream("GET", url, headers={"accept": "application/json"}) + async with request_cm as response: + if 300 <= response.status_code < 400 and "location" in response.headers: + if hop == max_redirects: + raise AgentResolverError( + "capabilities_unreachable", + f"capabilities fetch hit redirect limit ({max_redirects})", + ) + url = str(httpx.URL(url).join(response.headers["location"])) + try: + transport = build_async_ip_pinned_transport( + url, allow_private=allow_private + ) + except SSRFValidationError as exc: + raise AgentResolverError( + "capabilities_unreachable", + f"redirect target failed SSRF check: {exc}", + ) from exc + client_cm = httpx.AsyncClient( + transport=transport, + timeout=timeout_seconds, + follow_redirects=False, + trust_env=False, + ) + continue + + if response.status_code != 200: + raise AgentResolverError( + "capabilities_unreachable", + f"capabilities fetch returned HTTP {response.status_code}", + ) + + try: + body_bytes = await async_read_limited_bytes( + response, limit=max_body_bytes + ) + except ResponseTooLargeError as exc: + raise AgentResolverError( + "capabilities_invalid", f"capabilities {exc}" + ) from exc + + try: + parsed = json.loads(body_bytes) + except (ValueError, UnicodeDecodeError) as exc: + raise AgentResolverError( + "capabilities_invalid", + "capabilities response is not valid JSON", + ) from exc except SSRFValidationError as exc: raise AgentResolverError( "capabilities_unreachable", @@ -242,53 +292,6 @@ async def _fetch_capabilities( f"capabilities fetch failed: {exc}", ) from exc - if 300 <= response.status_code < 400 and "location" in response.headers: - if hop == max_redirects: - raise AgentResolverError( - "capabilities_unreachable", - f"capabilities fetch hit redirect limit ({max_redirects})", - ) - url = str(httpx.URL(url).join(response.headers["location"])) - # New host → new transport. Rebuild client_cm. - try: - transport = build_async_ip_pinned_transport( - url, allow_private=allow_private - ) - except SSRFValidationError as exc: - raise AgentResolverError( - "capabilities_unreachable", - f"redirect target failed SSRF check: {exc}", - ) from exc - client_cm = httpx.AsyncClient( - transport=transport, - timeout=timeout_seconds, - follow_redirects=False, - trust_env=False, - ) - continue - - if response.status_code != 200: - raise AgentResolverError( - "capabilities_unreachable", - f"capabilities fetch returned HTTP {response.status_code}", - ) - - body_bytes = response.content - if len(body_bytes) > max_body_bytes: - raise AgentResolverError( - "capabilities_invalid", - f"capabilities response exceeds {max_body_bytes} bytes " - f"(got {len(body_bytes)})", - ) - - try: - parsed = response.json() - except (ValueError, httpx.DecodingError, json.JSONDecodeError) as exc: - raise AgentResolverError( - "capabilities_invalid", - "capabilities response is not valid JSON", - ) from exc - if not isinstance(parsed, dict): raise AgentResolverError( "capabilities_invalid", @@ -578,6 +581,40 @@ def resolve_agent( # ---- verify factory ---- +_REPLAY_STORE_UNSET = object() +_VERIFY_FROM_AGENT_URL_REPLAY_STORE = InMemoryReplayStore() + + +class _NamespacedReplayStore: + """Scope a shared bounded replay cache to one resolved agent identity.""" + + def __init__(self, backing: ReplayStore, namespace: str) -> None: + self._backing = backing + self._prefix = hashlib.sha256(namespace.encode("utf-8")).hexdigest() + + def _keyid(self, keyid: str) -> str: + return f"{self._prefix}:{keyid}" + + def seen(self, keyid: str, nonce: str) -> bool: + return self._backing.seen(self._keyid(keyid), nonce) + + def remember(self, keyid: str, nonce: str, ttl_seconds: float) -> None: + self._backing.remember(self._keyid(keyid), nonce, ttl_seconds) + + def at_capacity(self, keyid: str) -> bool: + return self._backing.at_capacity(self._keyid(keyid)) + + def claim(self, keyid: str, nonce: str, ttl_seconds: float) -> ReplayClaimResult: + claim = getattr(self._backing, "claim", None) + if callable(claim): + result: ReplayClaimResult = claim(self._keyid(keyid), nonce, ttl_seconds) + return result + if self.seen(keyid, nonce): + return "replayed" + self.remember(keyid, nonce, ttl_seconds) + return "claimed" + + class _BrandJsonStaticJwksResolver(StaticJwksResolver): """A :class:`StaticJwksResolver` carrying the ``"brand_json"`` source discriminant AND the resolved ``jwks_uri``. @@ -622,7 +659,7 @@ async def verify_from_agent_url( brand_id: str | None = None, capability: Any = None, now: float | None = None, - replay_store: Any = None, + replay_store: Any = _REPLAY_STORE_UNSET, revocation_checker: Any = None, revocation_list: Any = None, allow_private_destinations: bool = False, @@ -659,6 +696,12 @@ async def verify_from_agent_url( :class:`AgentResolverError.code` directly — both exception hierarchies are preserved. + When ``replay_store`` is omitted, replay protection uses a bounded, + process-wide in-memory store namespaced by resolved agent URL so separate + counterparties may safely reuse a ``kid``. Pass ``None`` explicitly only + when replay protection is intentionally disabled, or provide a shared + store for multi-process deployments. + Returns ------- VerifiedSigner @@ -713,18 +756,24 @@ async def verify_from_agent_url( # marker the verifier would treat a bare ``StaticJwksResolver`` as a # publisher-pin-equivalent and skip the check — defeating the # production helper's defense against the shared-tenancy spoof. + if replay_store is _REPLAY_STORE_UNSET: + resolved_agent_url = str(resolution.agent_entry.get("url") or resolution.agent_url) + replay_store = _NamespacedReplayStore( + _VERIFY_FROM_AGENT_URL_REPLAY_STORE, + str(httpx.URL(resolved_agent_url)), + ) options = VerifyOptions( now=now if now is not None else _time.time(), capability=capability if capability is not None else VerifierCapability(supported=True), operation=operation, jwks_resolver=_BrandJsonStaticJwksResolver(resolution.jwks, jwks_uri=resolution.jwks_uri), - replay_store=replay_store, revocation_checker=revocation_checker, revocation_list=revocation_list, agent_url=resolution.agent_entry.get("url"), expected_key_origins=resolution.key_origins, signing_purpose=signing_purpose, posture=posture, + replay_store=replay_store, ) return await verify_starlette_request(request, options=options) diff --git a/src/adcp/signing/brand_authz.py b/src/adcp/signing/brand_authz.py index 04ec96a9..13ae6297 100644 --- a/src/adcp/signing/brand_authz.py +++ b/src/adcp/signing/brand_authz.py @@ -44,6 +44,7 @@ DEFAULT_MAX_AGE_SECONDS, DEFAULT_MAX_BRAND_JSON_BYTES, DEFAULT_MAX_REDIRECTS, + DEFAULT_MAX_STALE_SECONDS, DEFAULT_MIN_COOLDOWN_SECONDS, BrandAgentType, BrandJsonJwksResolver, @@ -144,6 +145,7 @@ def __init__( *, min_cooldown_seconds: float = DEFAULT_MIN_COOLDOWN_SECONDS, max_age_seconds: float = DEFAULT_MAX_AGE_SECONDS, + max_stale_seconds: float = DEFAULT_MAX_STALE_SECONDS, max_redirects: int = DEFAULT_MAX_REDIRECTS, max_body_bytes: int = DEFAULT_MAX_BRAND_JSON_BYTES, allow_private_destinations: bool = False, @@ -158,6 +160,7 @@ def __init__( brand_json_url, min_cooldown_seconds=min_cooldown_seconds, max_age_seconds=max_age_seconds, + max_stale_seconds=max_stale_seconds, max_redirects=max_redirects, max_body_bytes=max_body_bytes, allow_private_destinations=allow_private_destinations, @@ -294,10 +297,14 @@ async def _snapshot(self) -> _BrandJsonSnapshot | BrandJsonResolverError: if self._fetcher.is_stale(snap) and self._fetcher.can_refresh(snap): try: return await self._fetcher.refresh() - except BrandJsonResolverError: - # Stale-on-error: serve the prior snapshot. Matches - # the JWKS resolver's posture exactly. - return snap + except BrandJsonResolverError as exc: + if self._fetcher.can_serve_stale(snap): + return snap + return exc + if self._fetcher.is_stale(snap) and not self._fetcher.can_serve_stale(snap): + return self._fetcher.last_error or BrandJsonResolverError( + "fetch_failed", "expired brand.json authorization snapshot" + ) return snap @@ -312,6 +319,7 @@ def build_brand_json_resolvers( brand_id: str | None = None, min_cooldown_seconds: float = DEFAULT_MIN_COOLDOWN_SECONDS, max_age_seconds: float = DEFAULT_MAX_AGE_SECONDS, + max_stale_seconds: float = DEFAULT_MAX_STALE_SECONDS, max_redirects: int = DEFAULT_MAX_REDIRECTS, max_body_bytes: int = DEFAULT_MAX_BRAND_JSON_BYTES, allow_private_destinations: bool = False, @@ -333,6 +341,7 @@ def build_brand_json_resolvers( brand_json_url, min_cooldown_seconds=min_cooldown_seconds, max_age_seconds=max_age_seconds, + max_stale_seconds=max_stale_seconds, max_redirects=max_redirects, max_body_bytes=max_body_bytes, allow_private_destinations=allow_private_destinations, @@ -346,6 +355,7 @@ def build_brand_json_resolvers( brand_id=brand_id, min_cooldown_seconds=min_cooldown_seconds, max_age_seconds=max_age_seconds, + max_stale_seconds=max_stale_seconds, max_redirects=max_redirects, max_body_bytes=max_body_bytes, allow_private_destinations=allow_private_destinations, @@ -357,6 +367,7 @@ def build_brand_json_resolvers( brand_json_url, min_cooldown_seconds=min_cooldown_seconds, max_age_seconds=max_age_seconds, + max_stale_seconds=max_stale_seconds, max_redirects=max_redirects, max_body_bytes=max_body_bytes, allow_private_destinations=allow_private_destinations, diff --git a/src/adcp/signing/brand_jwks.py b/src/adcp/signing/brand_jwks.py index 3e82d441..e86bdd1a 100644 --- a/src/adcp/signing/brand_jwks.py +++ b/src/adcp/signing/brand_jwks.py @@ -36,6 +36,7 @@ from __future__ import annotations import asyncio +import json import re import time from collections.abc import Callable @@ -47,6 +48,7 @@ import httpx import idna +from adcp.signing._bounded_http import ResponseTooLargeError, async_read_limited_bytes from adcp.signing._idna_canonicalize import canonicalize_host from adcp.signing.jwks import ( AsyncCachingJwksResolver, @@ -95,6 +97,7 @@ DEFAULT_MIN_COOLDOWN_SECONDS = 30.0 DEFAULT_MAX_AGE_SECONDS = 3600.0 +DEFAULT_MAX_STALE_SECONDS = 3600.0 DEFAULT_MAX_REDIRECTS = 3 DEFAULT_BRAND_JSON_TIMEOUT_SECONDS = 10.0 @@ -187,6 +190,7 @@ def __init__( *, min_cooldown_seconds: float = DEFAULT_MIN_COOLDOWN_SECONDS, max_age_seconds: float = DEFAULT_MAX_AGE_SECONDS, + max_stale_seconds: float = DEFAULT_MAX_STALE_SECONDS, max_redirects: int = DEFAULT_MAX_REDIRECTS, max_body_bytes: int = DEFAULT_MAX_BRAND_JSON_BYTES, allow_private_destinations: bool = False, @@ -197,6 +201,7 @@ def __init__( self._url = brand_json_url self._min_cooldown = min_cooldown_seconds self._max_age = max_age_seconds + self._max_stale = max_stale_seconds self._max_redirects = max_redirects self._max_body_bytes = max_body_bytes self._allow_private = allow_private_destinations @@ -205,6 +210,8 @@ def __init__( self._client_factory = _client_factory self._snapshot: _BrandJsonSnapshot | None = None + self._last_attempt_at: float | None = None + self._last_error: BrandJsonResolverError | None = None # In-flight refresh future for single-flighting concurrent # callers — N tasks hitting a cold cache do ONE fetch, not N. # ``asyncio.Lock`` would also work but SERIALIZES (waiter N+1 @@ -242,7 +249,17 @@ def can_refresh(self, snapshot: _BrandJsonSnapshot | None = None) -> bool: snap = snapshot if snapshot is not None else self._snapshot if snap is None: return True - return self._clock() - snap.fetched_at >= self._min_cooldown + reference = max(snap.fetched_at, self._last_attempt_at or snap.fetched_at) + return self._clock() - reference >= self._min_cooldown + + def can_serve_stale(self, snapshot: _BrandJsonSnapshot | None = None) -> bool: + """Return whether an expired authorization snapshot is within grace.""" + snap = snapshot if snapshot is not None else self._snapshot + return snap is not None and self._clock() <= snap.expires_at + self._max_stale + + @property + def last_error(self) -> BrandJsonResolverError | None: + return self._last_error def clear(self) -> None: """Drop the cached snapshot. Next refresh will be unconditional.""" @@ -284,15 +301,21 @@ async def refresh(self) -> _BrandJsonSnapshot: self._refresh_in_flight = None async def _do_refresh(self) -> _BrandJsonSnapshot: - fetched = await _fetch_brand_json( - start_url=self._url, - current_etag=self._snapshot.etag if self._snapshot is not None else None, - max_redirects=self._max_redirects, - allow_private=self._allow_private, - timeout_seconds=self._timeout, - max_body_bytes=self._max_body_bytes, - client_factory=self._client_factory, - ) + self._last_attempt_at = self._clock() + try: + fetched = await _fetch_brand_json( + start_url=self._url, + current_etag=self._snapshot.etag if self._snapshot is not None else None, + max_redirects=self._max_redirects, + allow_private=self._allow_private, + timeout_seconds=self._timeout, + max_body_bytes=self._max_body_bytes, + client_factory=self._client_factory, + ) + except BrandJsonResolverError as exc: + self._last_error = exc + raise + self._last_error = None now = self._clock() if fetched.status == "not_modified" and self._snapshot is not None: @@ -360,6 +383,7 @@ def __init__( brand_id: str | None = None, min_cooldown_seconds: float = DEFAULT_MIN_COOLDOWN_SECONDS, max_age_seconds: float = DEFAULT_MAX_AGE_SECONDS, + max_stale_seconds: float = DEFAULT_MAX_STALE_SECONDS, max_redirects: int = DEFAULT_MAX_REDIRECTS, max_body_bytes: int = DEFAULT_MAX_BRAND_JSON_BYTES, allow_private_destinations: bool = False, @@ -385,6 +409,7 @@ def __init__( brand_json_url, min_cooldown_seconds=min_cooldown_seconds, max_age_seconds=max_age_seconds, + max_stale_seconds=max_stale_seconds, max_redirects=max_redirects, max_body_bytes=max_body_bytes, allow_private_destinations=allow_private_destinations, @@ -393,9 +418,8 @@ def __init__( _client_factory=_client_factory, ) - # Derived selector state. Recomputed whenever the fetcher's - # snapshot identity changes (final_url or etag); cheap to redo, - # so we don't bother diffing the body itself. + # Derived selector state. Recomputed for every successful body + # refresh; ETags are optional and may be reused incorrectly. self._selected: _SelectedAgent | None = None self._selected_for: tuple[str, str | None] | None = None self._inner: AsyncCachingJwksResolver | None = None @@ -419,8 +443,14 @@ async def resolve(self, kid: str) -> dict[str, Any] | None: try: await self._refresh() except BrandJsonResolverError: - # Keep stale on transient failure — same posture as JS. - pass + if not self._fetcher.can_serve_stale(snap): + self._selected = None + self._inner = None + return None + elif self._fetcher.is_stale(snap) and not self._fetcher.can_serve_stale(snap): + self._selected = None + self._inner = None + return None if self._inner is None: return None @@ -477,18 +507,21 @@ async def _refresh(self) -> None: self._sync_selector(snap) def _sync_selector(self, snap: _BrandJsonSnapshot) -> None: - """Reselect the agent if the brand.json snapshot identity changed.""" + """Reselect the agent from the current body, independent of validators.""" identity = (snap.final_url, snap.etag) - if self._selected is not None and self._selected_for == identity: - return - - agent = _select_agent( - snap.data, - snap.final_url, - agent_type=self._agent_type, - agent_id=self._agent_id, - brand_id=self._brand_id, - ) + try: + agent = _select_agent( + snap.data, + snap.final_url, + agent_type=self._agent_type, + agent_id=self._agent_id, + brand_id=self._brand_id, + ) + except BrandJsonResolverError: + self._selected = None + self._selected_for = identity + self._inner = None + raise if self._inner is None or ( self._selected is not None and self._selected.jwks_uri != agent.jwks_uri @@ -537,8 +570,7 @@ async def _fetch_brand_json( Body cap: each response is bounded to ``max_body_bytes`` (default 256 KiB). brand.json is small by design; an adversarial - multi-megabyte body would otherwise be buffered into memory by - ``response.json()``. + multi-megabyte body is stopped during streaming, before JSON parsing. ``client_factory`` is the test seam — production callers pass ``None`` to use the IP-pinned client; tests inject a factory that @@ -589,7 +621,38 @@ async def _fetch_brand_json( try: async with client_cm as client: try: - response = await client.get(url, headers=headers) + request_cm = client.stream("GET", url, headers=headers) + async with request_cm as response: + if hop == 0 and response.status_code == 304: + return _FetchedBrandJson( + status="not_modified", + final_url=url, + data=None, + etag=response.headers.get("etag"), + cache_control=response.headers.get("cache-control"), + ) + if response.status_code != 200: + raise BrandJsonResolverError( + "fetch_failed", + f"brand.json fetch returned HTTP {response.status_code}", + ) + + try: + body = await async_read_limited_bytes(response, limit=max_body_bytes) + except ResponseTooLargeError as exc: + raise BrandJsonResolverError( + "invalid_body", f"brand.json {exc}" + ) from exc + + try: + parsed = json.loads(body) + except (ValueError, UnicodeDecodeError) as exc: + raise BrandJsonResolverError( + "invalid_body", "brand.json response is not valid JSON" + ) from exc + + etag = response.headers.get("etag") + cache_control = response.headers.get("cache-control") except SSRFValidationError as exc: raise BrandJsonResolverError( "fetch_failed", f"brand.json URL failed SSRF check: {exc}" @@ -599,48 +662,12 @@ async def _fetch_brand_json( "fetch_failed", f"brand.json fetch failed: {exc}" ) from exc - if hop == 0 and response.status_code == 304: - return _FetchedBrandJson( - status="not_modified", - final_url=url, - data=None, - etag=response.headers.get("etag"), - cache_control=response.headers.get("cache-control"), - ) - if response.status_code != 200: - raise BrandJsonResolverError( - "fetch_failed", - f"brand.json fetch returned HTTP {response.status_code}", - ) - - # Body-size cap. ``response.content`` is already buffered - # by httpx (we're not streaming); reject if it exceeds - # the cap before paying the JSON-parse cost. - body = response.content - if len(body) > max_body_bytes: - raise BrandJsonResolverError( - "invalid_body", - f"brand.json response exceeds {max_body_bytes} bytes " f"(got {len(body)})", - ) - - try: - parsed = response.json() - except (ValueError, httpx.DecodingError) as exc: - raise BrandJsonResolverError( - "invalid_body", "brand.json response is not valid JSON" - ) from exc except BrandJsonResolverError: raise if not isinstance(parsed, dict): raise BrandJsonResolverError("invalid_body", "brand.json response is not an object") - # Capture response headers once before the client closes — - # used for both the `ok` return below and any 304 returns above - # (already returned by this point). - etag = response.headers.get("etag") - cache_control = response.headers.get("cache-control") - authoritative = parsed.get("authoritative_location") house = parsed.get("house") @@ -1029,5 +1056,6 @@ def _compute_lifetime(cache_control: str | None, max_age: float) -> float: "DEFAULT_BRAND_JSON_TIMEOUT_SECONDS", "DEFAULT_MAX_AGE_SECONDS", "DEFAULT_MAX_REDIRECTS", + "DEFAULT_MAX_STALE_SECONDS", "DEFAULT_MIN_COOLDOWN_SECONDS", ] diff --git a/src/adcp/signing/client.py b/src/adcp/signing/client.py index fcfa43fe..969e6809 100644 --- a/src/adcp/signing/client.py +++ b/src/adcp/signing/client.py @@ -53,6 +53,7 @@ from collections.abc import Awaitable, Callable, Iterator from contextlib import contextmanager from typing import TYPE_CHECKING, Any +from urllib.parse import urlsplit from adcp.signing.autosign import ( SigningConfig, @@ -81,12 +82,26 @@ """ +def _origin(url: Any) -> tuple[str, str, int] | None: + parsed = urlsplit(str(url)) + if not parsed.scheme or not parsed.hostname: + return None + try: + port = parsed.port + except ValueError as exc: + raise ValueError(f"invalid signing origin: {url!s}") from exc + if port is None: + port = 443 if parsed.scheme.lower() == "https" else 80 + return parsed.scheme.lower(), parsed.hostname.lower(), port + + def install_signing_event_hook( client: httpx.AsyncClient, *, signing: SigningConfig, seller_capability: RequestSigning | None = None, capability_provider: CapabilityProvider | None = None, + expected_origin: str | None = None, ) -> None: """Install an RFC 9421 request-signing event hook on ``client``. @@ -115,6 +130,12 @@ def install_signing_event_hook( capability needs lazy / re-resolved lookup. Sync and async are both supported. Returning ``None`` means "seller doesn't sign; skip every operation." + expected_origin: + Seller origin that signed requests are allowed to target. If omitted, + the client's ``base_url`` origin is used when available; otherwise the + first request inside :func:`signing_operation` binds the hook to its + origin. Cross-origin redirects and later cross-origin requests fail + before signing. Notes ----- @@ -128,7 +149,14 @@ def install_signing_event_hook( "`seller_capability` or `capability_provider`." ) + bound_origin = ( + _origin(expected_origin) if expected_origin is not None else _origin(client.base_url) + ) + if expected_origin is not None and bound_origin is None: + raise ValueError("expected_origin must be an absolute URL origin") + async def _hook(request: httpx.Request) -> None: + nonlocal bound_origin operation = current_operation.get() # Unset ContextVar → out-of-band call (health check, manual # probe). Skip without consulting capability. @@ -138,6 +166,19 @@ async def _hook(request: httpx.Request) -> None: if operation is None or operation == "get_adcp_capabilities": return + request_origin = _origin(request.url) + if request_origin is None: # pragma: no cover - httpx requests are absolute here + raise ValueError("cannot sign a request without an absolute origin") + if bound_origin is None: + # Bind before awaiting a dynamic provider so concurrent first-use + # requests cannot each establish a different signing origin. + bound_origin = request_origin + elif request_origin != bound_origin: + raise ValueError( + "refusing to sign a cross-origin request; redirects must be " + "handled and re-authorized outside signing_operation" + ) + capability: RequestSigning | None if seller_capability is not None: capability = seller_capability diff --git a/src/adcp/signing/jwks.py b/src/adcp/signing/jwks.py index 554cfd20..4663bf47 100644 --- a/src/adcp/signing/jwks.py +++ b/src/adcp/signing/jwks.py @@ -26,7 +26,9 @@ import asyncio import ipaddress +import json import socket +import threading import time from collections.abc import Callable from typing import Any, ClassVar, Literal, Protocol, runtime_checkable @@ -35,6 +37,7 @@ import httpx import idna +from adcp.signing._bounded_http import async_read_limited_bytes, read_limited_bytes from adcp.signing._idna_canonicalize import canonicalize_host from adcp.signing.errors import ( REQUEST_SIGNATURE_JWKS_UNAVAILABLE, @@ -43,7 +46,9 @@ ) DEFAULT_JWKS_COOLDOWN_SECONDS = 30.0 +DEFAULT_JWKS_MAX_AGE_SECONDS = 3600.0 DEFAULT_JWKS_TIMEOUT_SECONDS = 10.0 +DEFAULT_MAX_JWKS_BYTES = 1024 * 1024 # Cloud metadata endpoints that MUST be blocked even if somehow marked non-private BLOCKED_METADATA_IPS: frozenset[str] = frozenset( @@ -337,7 +342,12 @@ def resolve_and_validate_host( return host, accepted_ip, port -def default_jwks_fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: +def default_jwks_fetcher( + uri: str, + *, + allow_private: bool = False, + max_body_bytes: int = DEFAULT_MAX_JWKS_BYTES, +) -> dict[str, Any]: """Validate + resolve the URI once, then GET the JWKS over an IP-pinned transport. @@ -362,14 +372,38 @@ def default_jwks_fetcher(uri: str, *, allow_private: bool = False) -> dict[str, follow_redirects=False, trust_env=False, ) as client: - response = client.get(uri, headers={"Accept": "application/json"}) - response.raise_for_status() - body = response.json() + with client.stream("GET", uri, headers={"Accept": "application/json"}) as response: + response.raise_for_status() + body = json.loads(read_limited_bytes(response, limit=max_body_bytes)) if not isinstance(body, dict) or "keys" not in body: raise ValueError(f"JWKS document at {uri!r} has no 'keys' array") return body +def _index_jwks_keys(jwks: dict[str, Any], *, uri: str) -> dict[str, dict[str, Any]]: + """Validate the JWKS container and index object-shaped keys by ``kid``. + + Keys without a ``kid`` remain ignorable, matching the resolver's existing + behavior. Entries that cannot be JWK objects, and present ``kid`` values + that cannot be resolver identifiers, make the document malformed. + """ + keys = jwks.get("keys") + if not isinstance(keys, list): + raise ValueError(f"JWKS document at {uri!r} has no 'keys' array") + + indexed: dict[str, dict[str, Any]] = {} + for index, jwk in enumerate(keys): + if not isinstance(jwk, dict): + raise ValueError(f"JWKS key at index {index} in {uri!r} is not an object") + kid = jwk.get("kid") + if kid is None: + continue + if not isinstance(kid, str): + raise ValueError(f"JWKS key at index {index} in {uri!r} has a non-string 'kid'") + indexed[kid] = jwk + return indexed + + class CachingJwksResolver: """JWKS resolver with per-URI cache and refetch cooldown. @@ -389,46 +423,85 @@ def __init__( *, fetcher: JwksFetcher | None = None, cooldown_seconds: float = DEFAULT_JWKS_COOLDOWN_SECONDS, + max_age_seconds: float = DEFAULT_JWKS_MAX_AGE_SECONDS, allow_private: bool = False, clock: Callable[[], float] = time.monotonic, ) -> None: self._jwks_uri = jwks_uri self._fetcher = fetcher or default_jwks_fetcher self._cooldown = cooldown_seconds + self._max_age = max_age_seconds self._allow_private = allow_private self._clock = clock self._cache: dict[str, dict[str, Any]] = {} self._last_attempt: float | None = None + self._last_successful_refresh: float | None = None + self._last_failure: SignatureVerificationError | None = None self._primed = False + self._refresh_lock = threading.Lock() def __call__(self, keyid: str) -> dict[str, Any] | None: - if keyid in self._cache: - return self._cache[keyid] now = self._clock() - if not self._primed or ( + cache_expired = ( + self._primed + and self._last_successful_refresh is not None + and now - self._last_successful_refresh >= self._max_age + ) + miss_can_refresh = keyid not in self._cache and ( self._last_attempt is not None and now - self._last_attempt >= self._cooldown - ): - self._refresh(now) + ) + if not self._primed or cache_expired or miss_can_refresh: + with self._refresh_lock: + # Re-check after waiting: another thread may have refreshed a + # cold/expired cache or populated this kid while we blocked. + now = self._clock() + cache_expired = ( + self._primed + and self._last_successful_refresh is not None + and now - self._last_successful_refresh >= self._max_age + ) + miss_can_refresh = keyid not in self._cache and ( + self._last_attempt is not None and now - self._last_attempt >= self._cooldown + ) + if ( + cache_expired + and self._last_attempt is not None + and (now - self._last_attempt < self._cooldown) + ): + raise self._last_failure or SignatureVerificationError( + REQUEST_SIGNATURE_JWKS_UNAVAILABLE, + step=7, + message="cached JWKS is expired and refresh cooldown has not elapsed", + ) + if not self._primed or cache_expired or miss_can_refresh: + self._refresh(now) return self._cache.get(keyid) def _refresh(self, now: float) -> None: self._last_attempt = now try: jwks = self._fetcher(self._jwks_uri, allow_private=self._allow_private) + cache = _index_jwks_keys(jwks, uri=self._jwks_uri) except SSRFValidationError as exc: - raise SignatureVerificationError( + error = SignatureVerificationError( REQUEST_SIGNATURE_JWKS_UNTRUSTED, step=7, message=f"JWKS URI failed SSRF check: {exc}", - ) from exc + ) + self._last_failure = error + raise error from exc except (httpx.HTTPError, ValueError, OSError) as exc: - raise SignatureVerificationError( + error = SignatureVerificationError( REQUEST_SIGNATURE_JWKS_UNAVAILABLE, step=7, message=f"JWKS fetch failed: {exc}", - ) from exc + ) + self._last_failure = error + raise error from exc self._primed = True - self._cache = {jwk["kid"]: jwk for jwk in jwks.get("keys", []) if "kid" in jwk} + self._cache = cache + self._last_successful_refresh = now + self._last_failure = None class StaticJwksResolver: @@ -446,7 +519,12 @@ def __call__(self, keyid: str) -> dict[str, Any] | None: # --------------------------------------------------------------------------- -async def async_default_jwks_fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: +async def async_default_jwks_fetcher( + uri: str, + *, + allow_private: bool = False, + max_body_bytes: int = DEFAULT_MAX_JWKS_BYTES, +) -> dict[str, Any]: """Async counterpart to :func:`default_jwks_fetcher`. Uses :class:`httpx.AsyncClient` with an IP-pinned transport so @@ -464,9 +542,9 @@ async def async_default_jwks_fetcher(uri: str, *, allow_private: bool = False) - follow_redirects=False, trust_env=False, ) as client: - response = await client.get(uri, headers={"Accept": "application/json"}) - response.raise_for_status() - body = response.json() + async with client.stream("GET", uri, headers={"Accept": "application/json"}) as response: + response.raise_for_status() + body = json.loads(await async_read_limited_bytes(response, limit=max_body_bytes)) if not isinstance(body, dict) or "keys" not in body: raise ValueError(f"JWKS document at {uri!r} has no 'keys' array") return body @@ -492,16 +570,20 @@ def __init__( *, fetcher: AsyncJwksFetcher | None = None, cooldown_seconds: float = DEFAULT_JWKS_COOLDOWN_SECONDS, + max_age_seconds: float = DEFAULT_JWKS_MAX_AGE_SECONDS, allow_private: bool = False, clock: Callable[[], float] = time.monotonic, ) -> None: self._jwks_uri = jwks_uri self._fetcher = fetcher or async_default_jwks_fetcher self._cooldown = cooldown_seconds + self._max_age = max_age_seconds self._allow_private = allow_private self._clock = clock self._cache: dict[str, dict[str, Any]] = {} self._last_attempt: float | None = None + self._last_successful_refresh: float | None = None + self._last_failure: SignatureVerificationError | None = None self._primed = False # Construct the lock eagerly. Lazy init was racy: two tasks both # seeing ``self._lock is None`` would each construct a separate @@ -513,20 +595,38 @@ def __init__( self._lock: asyncio.Lock = asyncio.Lock() async def __call__(self, keyid: str) -> dict[str, Any] | None: - if keyid in self._cache: - return self._cache[keyid] now = self._clock() - if not self._primed or ( + cache_expired = ( + self._primed + and self._last_successful_refresh is not None + and now - self._last_successful_refresh >= self._max_age + ) + miss_can_refresh = keyid not in self._cache and ( self._last_attempt is not None and now - self._last_attempt >= self._cooldown + ) + if ( + cache_expired + and self._last_attempt is not None + and (now - self._last_attempt < self._cooldown) ): + raise self._last_failure or SignatureVerificationError( + REQUEST_SIGNATURE_JWKS_UNAVAILABLE, + step=7, + message="cached JWKS is expired and refresh cooldown has not elapsed", + ) + if not self._primed or cache_expired or miss_can_refresh: async with self._lock: # Re-check after acquiring: another task may have refreshed. - if keyid in self._cache: - return self._cache[keyid] now = self._clock() - if not self._primed or ( + cache_expired = ( + self._primed + and self._last_successful_refresh is not None + and now - self._last_successful_refresh >= self._max_age + ) + miss_can_refresh = keyid not in self._cache and ( self._last_attempt is not None and now - self._last_attempt >= self._cooldown - ): + ) + if not self._primed or cache_expired or miss_can_refresh: await self._refresh(now) return self._cache.get(keyid) @@ -534,20 +634,27 @@ async def _refresh(self, now: float) -> None: self._last_attempt = now try: jwks = await self._fetcher(self._jwks_uri, allow_private=self._allow_private) + cache = _index_jwks_keys(jwks, uri=self._jwks_uri) except SSRFValidationError as exc: - raise SignatureVerificationError( + error = SignatureVerificationError( REQUEST_SIGNATURE_JWKS_UNTRUSTED, step=7, message=f"JWKS URI failed SSRF check: {exc}", - ) from exc + ) + self._last_failure = error + raise error from exc except (httpx.HTTPError, ValueError, OSError) as exc: - raise SignatureVerificationError( + error = SignatureVerificationError( REQUEST_SIGNATURE_JWKS_UNAVAILABLE, step=7, message=f"JWKS fetch failed: {exc}", - ) from exc + ) + self._last_failure = error + raise error from exc self._primed = True - self._cache = {jwk["kid"]: jwk for jwk in jwks.get("keys", []) if "kid" in jwk} + self._cache = cache + self._last_successful_refresh = now + self._last_failure = None def as_async_resolver(resolver: JwksResolver) -> AsyncJwksResolver: @@ -573,7 +680,9 @@ async def resolve(keyid: str) -> dict[str, Any] | None: "CachingJwksResolver", "DEFAULT_ALLOWED_PORTS", "DEFAULT_JWKS_COOLDOWN_SECONDS", + "DEFAULT_JWKS_MAX_AGE_SECONDS", "DEFAULT_JWKS_TIMEOUT_SECONDS", + "DEFAULT_MAX_JWKS_BYTES", "JwksFetcher", "JwksResolver", "SSRFValidationError", diff --git a/src/adcp/signing/pg/replay_store.py b/src/adcp/signing/pg/replay_store.py index f8b46077..1508c108 100644 --- a/src/adcp/signing/pg/replay_store.py +++ b/src/adcp/signing/pg/replay_store.py @@ -79,6 +79,8 @@ async def sweep_forever(store: PgReplayStore, interval: float = 60.0) -> None: import re from typing import TYPE_CHECKING +from adcp.signing.replay import ReplayClaimResult + if TYPE_CHECKING: from psycopg_pool import ConnectionPool @@ -115,9 +117,8 @@ class PgReplayStore: ---------- pool: A :class:`psycopg_pool.ConnectionPool` owned by the caller. Each - operation acquires a short-lived connection, runs a single - statement, and returns the connection. No long-lived - transactions, no cross-operation state. + operation acquires a short-lived connection and returns it promptly. + ``claim`` runs its capacity check and insert in one short transaction. per_keyid_cap: Maximum number of live (non-expired) nonces per ``keyid``. Mirrors :class:`InMemoryReplayStore`; spec-recommended 1M. @@ -180,6 +181,14 @@ def __init__( f"WHERE keyid = %s AND expires_at > now()" ) self._sql_sweep = f"DELETE FROM {self._table} WHERE expires_at <= now()" # noqa: S608 + self._sql_claim_lock = "SELECT pg_advisory_xact_lock(hashtextextended(%s, 9173))" + self._sql_claim = ( + f"INSERT INTO {self._table} (keyid, nonce, expires_at) " # noqa: S608 + f"VALUES (%s, %s, now() + make_interval(secs => %s)) " + f"ON CONFLICT (keyid, nonce) DO UPDATE " + f"SET expires_at = EXCLUDED.expires_at " + f"WHERE {self._table}.expires_at <= now() RETURNING 1" + ) # -- schema bootstrap -------------------------------------------- @@ -252,6 +261,25 @@ def at_capacity(self, keyid: str) -> bool: row = cur.fetchone() return bool(row[0]) if row is not None else False + def claim(self, keyid: str, nonce: str, ttl_seconds: float) -> ReplayClaimResult: + """Atomically enforce the per-key cap and reserve a fresh nonce. + + A transaction-scoped advisory lock serializes claims for one key id + across all verifier processes. The primary key then provides the + exact nonce winner selection. + """ + with self._pool.connection() as conn, conn.transaction(), conn.cursor() as cur: + cur.execute(self._sql_claim_lock, (keyid,)) + cur.execute(self._sql_seen, (keyid, nonce)) + if cur.fetchone() is not None: + return "replayed" + cur.execute(self._sql_at_capacity, (self._per_keyid_cap, keyid)) + row = cur.fetchone() + if row is not None and bool(row[0]): + return "capacity" + cur.execute(self._sql_claim, (keyid, nonce, ttl_seconds)) + return "claimed" if cur.fetchone() is not None else "replayed" + # -- admin / cron ------------------------------------------------ def sweep_expired(self) -> int: diff --git a/src/adcp/signing/replay.py b/src/adcp/signing/replay.py index 2b275855..b686f6c4 100644 --- a/src/adcp/signing/replay.py +++ b/src/adcp/signing/replay.py @@ -1,8 +1,8 @@ """Replay dedup store for the AdCP request-signing profile. Stores `(keyid, nonce)` pairs that have already been accepted, with a TTL that -mirrors the signature's `expires` parameter plus skew. A per-keyid cap prevents -unbounded growth — when the cap is hit, new signatures for that keyid are +mirrors the signature's `expires` parameter plus skew. Per-keyid and global +caps prevent unbounded growth — when either cap is hit, new signatures are rejected with `request_signature_rate_abuse` rather than silently evicting older entries (which would create a replay window under attack). @@ -15,11 +15,19 @@ import threading import time from collections.abc import Callable -from typing import Protocol +from typing import Literal, Protocol + +ReplayClaimResult = Literal["claimed", "replayed", "capacity"] class ReplayStore(Protocol): - """Minimum interface a replay backend must expose.""" + """Legacy-compatible interface a replay backend must expose. + + New backends should additionally implement :class:`AtomicReplayStore`. + Keeping the atomic operation on a separate Protocol lets applications + written against the pre-6.6 replay API continue to type-check while the + verifier provides a visible, race-prone compatibility fallback. + """ def seen(self, keyid: str, nonce: str) -> bool: ... @@ -28,6 +36,14 @@ def remember(self, keyid: str, nonce: str, ttl_seconds: float) -> None: ... def at_capacity(self, keyid: str) -> bool: ... +class AtomicReplayStore(ReplayStore, Protocol): + """Replay backend that can reserve a nonce without a check/write race.""" + + def claim(self, keyid: str, nonce: str, ttl_seconds: float) -> ReplayClaimResult: + """Atomically reserve a nonce, or report why it cannot be reserved.""" + ... + + # Cap on the number of expired entries swept per mutating call. Bounded so that # a single `remember` / `seen` stays O(1) amortized on a well-behaved workload; # natural inserts and lookups sweep incrementally. @@ -37,17 +53,29 @@ def at_capacity(self, keyid: str) -> bool: ... class InMemoryReplayStore: """Process-local replay store. Uses a monotonic clock for TTL bookkeeping so wall-clock jumps (NTP adjustments, VM suspend/resume) don't race eviction. + + ``global_cap`` bounds attacker-controlled key rotation as well as nonce + volume. An indexed min-heap expires entries incrementally without copying + or scanning the nonce table on accepted requests. """ def __init__( self, *, per_keyid_cap: int = 1_000_000, + global_cap: int = 1_000_000, clock: Callable[[], float] = time.monotonic, ) -> None: + if per_keyid_cap <= 0: + raise ValueError("per_keyid_cap must be greater than zero") + if global_cap <= 0: + raise ValueError("global_cap must be greater than zero") self._per_keyid_cap = per_keyid_cap + self._global_cap = global_cap self._clock = clock self._entries: dict[tuple[str, str], float] = {} + self._expiry_heap: list[tuple[float, tuple[str, str]]] = [] + self._heap_positions: dict[tuple[str, str], int] = {} self._counts: dict[str, int] = {} self._cap_hit: set[str] = set() self._lock = threading.RLock() @@ -59,17 +87,50 @@ def seen(self, keyid: str, nonce: str) -> bool: def remember(self, keyid: str, nonce: str, ttl_seconds: float) -> None: with self._lock: - self._sweep_for_keyid(keyid) + now = self._clock() + self._purge_expired(now) key = (keyid, nonce) if key not in self._entries: + if ( + len(self._entries) >= self._global_cap + or self._counts.get(keyid, 0) >= self._per_keyid_cap + ): + return self._counts[keyid] = self._counts.get(keyid, 0) + 1 - self._entries[key] = self._clock() + ttl_seconds + expiry = now + ttl_seconds + self._entries[key] = expiry + self._push_expiry(key, expiry) def at_capacity(self, keyid: str) -> bool: with self._lock: + self._purge_expired(self._clock()) if keyid in self._cap_hit: return True - return self._counts.get(keyid, 0) >= self._per_keyid_cap + return ( + len(self._entries) >= self._global_cap + or self._counts.get(keyid, 0) >= self._per_keyid_cap + ) + + def claim(self, keyid: str, nonce: str, ttl_seconds: float) -> ReplayClaimResult: + """Atomically check capacity/replay state and reserve ``nonce``.""" + with self._lock: + now = self._clock() + self._expire_one(keyid, nonce) + if (keyid, nonce) in self._entries: + return "replayed" + self._purge_expired(now) + if ( + keyid in self._cap_hit + or len(self._entries) >= self._global_cap + or self._counts.get(keyid, 0) >= self._per_keyid_cap + ): + return "capacity" + key = (keyid, nonce) + expiry = now + ttl_seconds + self._entries[key] = expiry + self._counts[keyid] = self._counts.get(keyid, 0) + 1 + self._push_expiry(key, expiry) + return "claimed" def mark_cap_hit(self, keyid: str) -> None: """Test-harness hook — simulate the cap being reached for this keyid.""" @@ -81,22 +142,80 @@ def _expire_one(self, keyid: str, nonce: str) -> None: expiry = self._entries.get(key) if expiry is not None and expiry < self._clock(): del self._entries[key] + self._remove_expiry(key) self._counts[keyid] = self._counts.get(keyid, 1) - 1 if self._counts[keyid] <= 0: self._counts.pop(keyid, None) - def _sweep_for_keyid(self, keyid: str) -> None: - now = self._clock() - removed = 0 - # Scan only entries for this keyid to bound per-call work under load. - for key, expiry in list(self._entries.items()): - if key[0] != keyid: - continue - if expiry < now: - del self._entries[key] - self._counts[keyid] = self._counts.get(keyid, 1) - 1 - if self._counts[keyid] <= 0: - self._counts.pop(keyid, None) - removed += 1 - if removed >= _SWEEP_BATCH: - return + def _push_expiry(self, key: tuple[str, str], expiry: float) -> None: + position = self._heap_positions.get(key) + if position is None: + position = len(self._expiry_heap) + self._expiry_heap.append((expiry, key)) + self._heap_positions[key] = position + self._sift_up(position) + return + old_expiry = self._expiry_heap[position][0] + self._expiry_heap[position] = (expiry, key) + if expiry < old_expiry: + self._sift_up(position) + else: + self._sift_down(position) + + def _remove_expiry(self, key: tuple[str, str]) -> None: + position = self._heap_positions.pop(key, None) + if position is None: + return + last = self._expiry_heap.pop() + if position == len(self._expiry_heap): + return + self._expiry_heap[position] = last + self._heap_positions[last[1]] = position + if position > 0 and self._expiry_heap[position] < self._expiry_heap[(position - 1) // 2]: + self._sift_up(position) + else: + self._sift_down(position) + + def _sift_up(self, position: int) -> None: + while position > 0: + parent = (position - 1) // 2 + if self._expiry_heap[parent] <= self._expiry_heap[position]: + return + self._swap_heap(parent, position) + position = parent + + def _sift_down(self, position: int) -> None: + size = len(self._expiry_heap) + while (left := position * 2 + 1) < size: + right = left + 1 + child = ( + right + if right < size and self._expiry_heap[right] < self._expiry_heap[left] + else left + ) + if self._expiry_heap[position] <= self._expiry_heap[child]: + return + self._swap_heap(position, child) + position = child + + def _swap_heap(self, left: int, right: int) -> None: + self._expiry_heap[left], self._expiry_heap[right] = ( + self._expiry_heap[right], + self._expiry_heap[left], + ) + self._heap_positions[self._expiry_heap[left][1]] = left + self._heap_positions[self._expiry_heap[right][1]] = right + + def _purge_expired(self, now: float) -> None: + examined = 0 + while self._expiry_heap and examined < _SWEEP_BATCH: + expiry, key = self._expiry_heap[0] + if expiry >= now: + return + self._remove_expiry(key) + examined += 1 + del self._entries[key] + keyid = key[0] + self._counts[keyid] = self._counts.get(keyid, 1) - 1 + if self._counts[keyid] <= 0: + self._counts.pop(keyid, None) diff --git a/src/adcp/signing/revocation_fetcher.py b/src/adcp/signing/revocation_fetcher.py index 1112ecd1..34a38f25 100644 --- a/src/adcp/signing/revocation_fetcher.py +++ b/src/adcp/signing/revocation_fetcher.py @@ -38,12 +38,17 @@ import time from collections.abc import Callable from dataclasses import dataclass -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from typing import Any, Protocol import httpx import idna +from adcp.signing._bounded_http import ( + ResponseTooLargeError, + async_read_limited_bytes, + read_limited_bytes, +) from adcp.signing._idna_canonicalize import canonicalize_host from adcp.signing.jwks import ( DEFAULT_JWKS_TIMEOUT_SECONDS, @@ -69,6 +74,7 @@ # within `next_update + grace * last_interval`, all subsequent is_revoked # calls fail closed. Spec recommends 2×. DEFAULT_GRACE_MULTIPLIER = 2.0 +DEFAULT_MAX_REVOCATION_LIST_BYTES = 1024 * 1024 # Shape-validate `Last-Modified` before persisting it for the next # `If-Modified-Since` request. The value comes from a (potentially @@ -246,6 +252,7 @@ def default_revocation_list_fetcher( if_modified_since: str | None = None, allow_private: bool = False, timeout: float = DEFAULT_JWKS_TIMEOUT_SECONDS, + max_body_bytes: int = DEFAULT_MAX_REVOCATION_LIST_BYTES, ) -> FetchResult: """HTTPS GET the revocation list, honoring SSRF rules and conditional requests. @@ -271,19 +278,25 @@ def default_revocation_list_fetcher( follow_redirects=False, trust_env=False, ) as client: - response = client.get(uri, headers=headers) - except httpx.HTTPError as exc: + with client.stream("GET", uri, headers=headers) as response: + response_text = "" + if response.status_code == 200: + response_text = read_limited_bytes(response, limit=max_body_bytes).decode( + "utf-8" + ) + return _fetch_result_from_response( + uri, + response.status_code, + response_text, + response.headers, + if_none_match=if_none_match, + if_modified_since=if_modified_since, + ) + except ResponseTooLargeError as exc: + raise RevocationListFetchError(f"revocation list {uri!r} {exc}") from exc + except (httpx.HTTPError, UnicodeDecodeError) as exc: raise RevocationListFetchError(f"revocation list GET {uri!r} failed: {exc}") from exc - return _fetch_result_from_response( - uri, - response.status_code, - response.text, - response.headers, - if_none_match=if_none_match, - if_modified_since=if_modified_since, - ) - async def async_default_revocation_list_fetcher( uri: str, @@ -292,6 +305,7 @@ async def async_default_revocation_list_fetcher( if_modified_since: str | None = None, allow_private: bool = False, timeout: float = DEFAULT_JWKS_TIMEOUT_SECONDS, + max_body_bytes: int = DEFAULT_MAX_REVOCATION_LIST_BYTES, ) -> FetchResult: """Async counterpart to :func:`default_revocation_list_fetcher`. @@ -310,19 +324,25 @@ async def async_default_revocation_list_fetcher( follow_redirects=False, trust_env=False, ) as client: - response = await client.get(uri, headers=headers) - except httpx.HTTPError as exc: + async with client.stream("GET", uri, headers=headers) as response: + response_text = "" + if response.status_code == 200: + response_text = ( + await async_read_limited_bytes(response, limit=max_body_bytes) + ).decode("utf-8") + return _fetch_result_from_response( + uri, + response.status_code, + response_text, + response.headers, + if_none_match=if_none_match, + if_modified_since=if_modified_since, + ) + except ResponseTooLargeError as exc: + raise RevocationListFetchError(f"revocation list {uri!r} {exc}") from exc + except (httpx.HTTPError, UnicodeDecodeError) as exc: raise RevocationListFetchError(f"revocation list GET {uri!r} failed: {exc}") from exc - return _fetch_result_from_response( - uri, - response.status_code, - response.text, - response.headers, - if_none_match=if_none_match, - if_modified_since=if_modified_since, - ) - def _sanitize_last_modified(raw: str | None) -> str | None: """Validate a ``Last-Modified`` header value before persisting it. @@ -396,23 +416,6 @@ def _normalize_issuer(issuer: str) -> str: return urlunsplit((scheme, netloc, "", "", "")) -def _slide_next_update(current: RevocationList, polling_interval_seconds: float) -> RevocationList: - """Return ``current`` with ``next_update`` advanced by one polling interval. - - Used on a 304 response so the cached list's freshness window slides - forward without needing a fresh JWS. Preserves every other field. - """ - prior = _parse_iso8601(current.next_update) - new_next_update = prior + timedelta(seconds=polling_interval_seconds) - return RevocationList( - issuer=current.issuer, - updated=current.updated, - next_update=new_next_update.isoformat().replace("+00:00", "Z"), - revoked_kids=current.revoked_kids, - revoked_jtis=current.revoked_jtis, - ) - - def _post_jws_validation( payload: dict[str, Any], *, @@ -446,6 +449,10 @@ def _post_jws_validation( f"revocation list next_update {revocation_list.next_update!r} is not " f"after updated {revocation_list.updated!r}" ) + if next_update <= now_wall: + raise RevocationListParseError( + f"revocation list next_update {revocation_list.next_update!r} is already expired" + ) # Reject a freshly-fetched list whose `updated` is older than the # one we already have cached. Defense against CDN replay or @@ -502,19 +509,12 @@ def _init_state(self) -> None: self._last_refresh_attempt = None def _handle_not_modified(self, *, now_mono: float) -> None: - """Slide ``next_update`` forward on a 304 response. + """Record a successful conditional request without changing signed freshness. - Without this, subsequent calls past the original ``next_update`` - would re-enter the refresh branch on every verification (gated - only by the 60s cooldown). Advancing the cached - ``next_update`` by one polling interval lets the hot path - short-circuit cleanly. + A 304 authenticates no new JWS payload, so it cannot extend the + signed ``next_update`` authorization boundary. """ self._last_successful_refresh = now_mono - if self._current_list is not None and self._last_polling_interval_seconds: - self._current_list = _slide_next_update( - self._current_list, self._last_polling_interval_seconds - ) def _commit( self, @@ -794,12 +794,21 @@ def _ensure_fresh(self) -> None: else float("inf") ) if since_last_attempt >= MIN_POLLING_INTERVAL_SECONDS: + last_exc: Exception = RevocationListFetchError( + "another refresh completed without extending signed freshness" + ) try: - self._refresh(conditional=True, now_wall=now_wall, now_mono=now_mono) - return + installed_signed_payload = self._refresh( + conditional=True, now_wall=now_wall, now_mono=now_mono + ) + if installed_signed_payload: + return + last_exc = RevocationListFetchError( + "304 response did not extend signed next_update" + ) except (RevocationListFetchError, RevocationListParseError) as exc: # Fall through to the grace-window check below. - last_exc: Exception = exc + last_exc = exc else: last_exc = RevocationListFetchError( f"refresh cooldown not elapsed ({since_last_attempt:.0f}s < " @@ -815,7 +824,7 @@ def _ensure_fresh(self) -> None: ) from last_exc # Still within grace — serve the cached list. - def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: float) -> None: + def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: float) -> bool: self._last_refresh_attempt = now_mono if_none_match = self._current_etag if conditional else None if_modified_since = self._current_last_modified if conditional else None @@ -826,7 +835,7 @@ def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: float) -> ) if result.not_modified: self._handle_not_modified(now_mono=now_mono) - return + return False try: payload = verify_jws_document( @@ -846,6 +855,7 @@ def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: float) -> current_list=self._current_list, ) self._commit(result=result, revocation_list=revocation_list, now_mono=now_mono) + return True class AsyncCachingRevocationChecker(_CheckerState): @@ -981,6 +991,9 @@ async def _ensure_fresh(self) -> None: else float("inf") ) if since_last_attempt >= MIN_POLLING_INTERVAL_SECONDS: + last_exc: Exception = RevocationListFetchError( + "another refresh completed without extending signed freshness" + ) try: async with self._lock: # Re-check under the lock with fresh clock reads. @@ -989,14 +1002,18 @@ async def _ensure_fresh(self) -> None: now_mono_inside - self._last_refresh_attempt >= MIN_POLLING_INTERVAL_SECONDS ): now_wall_inside = self._wall_clock() - await self._refresh( + installed_signed_payload = await self._refresh( conditional=True, now_wall=now_wall_inside, now_mono=now_mono_inside, ) - return + if installed_signed_payload: + return + last_exc = RevocationListFetchError( + "304 response did not extend signed next_update" + ) except (RevocationListFetchError, RevocationListParseError) as exc: - last_exc: Exception = exc + last_exc = exc else: last_exc = RevocationListFetchError( f"refresh cooldown not elapsed ({since_last_attempt:.0f}s < " @@ -1011,7 +1028,7 @@ async def _ensure_fresh(self) -> None: f"last refresh error: {last_exc}" ) from last_exc - async def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: float) -> None: + async def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: float) -> bool: # Stamp the attempt BEFORE the awaitable. On CancelledError the # finally block rolls it back so a cancelled task doesn't burn # the 60s cooldown for the next caller — non-cancellation @@ -1036,7 +1053,7 @@ async def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: flo raise if result.not_modified: self._handle_not_modified(now_mono=now_mono) - return + return False try: payload = await averify_jws_document( @@ -1056,6 +1073,7 @@ async def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: flo current_list=self._current_list, ) self._commit(result=result, revocation_list=revocation_list, now_mono=now_mono) + return True __all__ = [ @@ -1063,6 +1081,7 @@ async def _refresh(self, *, conditional: bool, now_wall: datetime, now_mono: flo "AsyncRevocationListFetcher", "CachingRevocationChecker", "DEFAULT_GRACE_MULTIPLIER", + "DEFAULT_MAX_REVOCATION_LIST_BYTES", "FetchResult", "REVOCATION_LIST_TYP", "RevocationListFetchError", diff --git a/src/adcp/signing/verifier.py b/src/adcp/signing/verifier.py index e813896e..e4cefb06 100644 --- a/src/adcp/signing/verifier.py +++ b/src/adcp/signing/verifier.py @@ -58,7 +58,7 @@ ) from adcp.signing.jwks import JwksResolver from adcp.signing.key_origins import check_key_origin_consistency -from adcp.signing.replay import InMemoryReplayStore, ReplayStore +from adcp.signing.replay import InMemoryReplayStore, ReplayClaimResult, ReplayStore from adcp.signing.revocation import RevocationChecker, RevocationList CoversDigestPolicy = Literal["required", "forbidden", "either"] @@ -88,18 +88,13 @@ class VerifiedSigner: class VerifierCapability: """The `request_signing` block a verifier advertises on get_adcp_capabilities. - Defaults to ``covers_content_digest="either"`` per the AdCP 3.0 schema - (`get-adcp-capabilities-response.json` declares this as the default - explicitly). The schema rationale recommends `"required"` for - spend-committing operations in production, and AdCP 4.0 recommends - `"required"` more broadly. - - Operators who want body-integrity authentication end-to-end on - every request — closing the MITM-inside-TLS-termination case where - a reverse proxy or service mesh can swap bodies on unsigned-digest - requests — opt INTO ``covers_content_digest="required"`` explicitly, - or use ``required_for=frozenset({"create_media_buy", ...})`` to - promote spend-committing operations selectively. + Defaults to ``covers_content_digest="required"`` so constructing a + verifier without an explicit peer capability cannot authenticate a + state-changing request while leaving its body unsigned. This is a secure + local default, intentionally stricter than the AdCP 3.0 wire-schema + default of ``"either"``. Callers mirroring a peer's advertised capability + should pass its explicit value; ``"either"`` and ``"forbidden"`` remain + available for protocol compatibility. The webhook-signing profile (``adcp.signing.webhook_verifier``) hard- codes ``"required"`` regardless of this default — webhook bodies @@ -107,7 +102,7 @@ class VerifierCapability: """ supported: bool = True - covers_content_digest: CoversDigestPolicy = "either" + covers_content_digest: CoversDigestPolicy = "required" required_for: frozenset[str] = field(default_factory=frozenset) supported_for: frozenset[str] = field(default_factory=frozenset) @@ -116,15 +111,14 @@ class VerifierCapability: class VerifyOptions: """Options bag passed to ``verify_request_signature``. - ``replay_store`` defaults to a fresh :class:`InMemoryReplayStore` so the - verifier always enforces nonce uniqueness on every request — defaulting - to ``None`` would silently disable replay protection for callers who - forget to wire a store, the exact security regression the AdCP profile's - step 12 exists to prevent. Wire an explicit shared store (Redis, Postgres, - etc.) for multi-replica deployments where replay state must be - coordinated across processes; pass ``replay_store=None`` if you genuinely - need to bypass the check (uncommon — typically only short-lived - integration tests). + ``replay_store`` defaults to a fresh :class:`InMemoryReplayStore` so one + options instance enforces nonce uniqueness across every verification that + reuses it. Constructing a new :class:`VerifyOptions` per request also + creates a new store and therefore resets replay history; long-lived + verifiers must reuse the options instance or provide an explicit store. + Wire a shared store (Redis, Postgres, etc.) for multi-replica deployments + where replay state must be coordinated across processes; pass + ``replay_store=None`` only when you genuinely need to bypass the check. ``revocation_checker`` and ``revocation_list`` remain optional — most agents don't track key revocations at runtime, and the verifier @@ -152,9 +146,9 @@ class VerifyOptions: #: (``request_signing``, ``webhook_signing``, ...). When provided #: AND the JWKS resolver reports ``jwks_source == "brand_json"``, #: the verifier checks that the resolved ``jwks_uri`` host - #: matches the declared origin for ``signing_purpose``. ``None`` - #: (default) skips the check — adopters who don't yet plumb - #: capabilities through to the verifier see no behavior change. + #: matches the declared origin for ``signing_purpose``. For a + #: brand-sourced resolver, ``None`` is treated as a missing declaration + #: and fails closed; other resolver sources skip the check. expected_key_origins: Mapping[str, str] | None = None #: Purpose key used to look up ``expected_key_origins`` and to #: render error messages. Default ``"request_signing"`` matches @@ -298,9 +292,8 @@ def verify_request_signature( message=f"key {keyid!r} is revoked", ) - # Step 9a (per spec, after adcp#2342): per-keyid cap runs between JWKS - # resolution and crypto verify. A compromised or misconfigured signer - # hitting the cap must be rejected cheaply, not after Ed25519/ECDSA verify. + # Cheap early rejection; ``claim`` repeats the capacity check atomically + # after crypto verification so concurrent claims cannot exceed the cap. if options.replay_store is not None and options.replay_store.at_capacity(keyid): raise SignatureVerificationError( REQUEST_SIGNATURE_RATE_ABUSE, @@ -345,17 +338,23 @@ def verify_request_signature( ) if options.replay_store is not None: - if options.replay_store.seen(keyid, nonce): + ttl = max( + float(parsed.params["expires"]) - options.now + options.max_skew_seconds, + 0.0, + ) + claim = _claim_replay_nonce(options.replay_store, keyid, nonce, ttl) + if claim == "replayed": raise SignatureVerificationError( REQUEST_SIGNATURE_REPLAYED, step=12, message=f"nonce {nonce!r} already seen for keyid {keyid!r}", ) - ttl = max( - float(parsed.params["expires"]) - options.now + options.max_skew_seconds, - 0.0, - ) - options.replay_store.remember(keyid, nonce, ttl) + if claim == "capacity": + raise SignatureVerificationError( + REQUEST_SIGNATURE_RATE_ABUSE, + step="9a", + message=f"replay cache at capacity for keyid {keyid!r}", + ) return VerifiedSigner( key_id=keyid, @@ -366,6 +365,36 @@ def verify_request_signature( ) +def _claim_replay_nonce( + store: ReplayStore, + keyid: str, + nonce: str, + ttl_seconds: float, +) -> ReplayClaimResult: + """Reserve a nonce, retaining compatibility with pre-atomic stores. + + The legacy ``seen`` then ``remember`` sequence is necessarily racy. It is + retained only so upgrading the SDK does not turn a previously valid custom + backend into an ``AttributeError`` after signature verification. + """ + claim = getattr(store, "claim", None) + if callable(claim): + result: ReplayClaimResult = claim(keyid, nonce, ttl_seconds) + return result + + warnings.warn( + "ReplayStore does not implement atomic claim(); falling back to the " + "legacy seen()/remember() sequence, which cannot prevent concurrent " + "replays. Implement claim() before this compatibility path is removed.", + RuntimeWarning, + stacklevel=2, + ) + if store.seen(keyid, nonce): + return "replayed" + store.remember(keyid, nonce, ttl_seconds) + return "claimed" + + def _precheck_presence( *, sig_input_raw: str | None, @@ -530,15 +559,12 @@ def _maybe_check_key_origin( we fail closed via the mismatch path (``actual_origin`` becomes ``None``). - Skip + warn cases (both fire :func:`warnings.warn` so the - one-time message in the operator's log surfaces the misconfig): + Missing declarations fail closed: - * ``jwks_source == "brand_json"`` + ``expected_key_origins is None``: - the resolver IS brand-json-sourced but the caller didn't surface - the operator's declared ``identity.key_origins`` map, so the - spec-mandated check silently no-ops. ``UserWarning`` — the - adopter needs to thread ``expected_key_origins`` through - ``VerifyOptions``. + * ``jwks_source == "brand_json"`` + ``expected_key_origins is None`` + is treated as an empty declaration map and raises + ``request_signature_key_origin_missing``. Brand-sourced trust cannot + silently downgrade when capabilities omit the mandatory binding. * ``expected_key_origins`` set + resolver has no ``jwks_source``: adopter upgraded the SDK but their custom resolver predates the discriminant. ``DeprecationWarning`` — set @@ -547,20 +573,7 @@ def _maybe_check_key_origin( spec defense; without it the SDK silently downgrades to no-check. """ source = getattr(resolver, "jwks_source", None) - if expected_key_origins is None: - if source == "brand_json": - warnings.warn( - "Resolver advertises jwks_source='brand_json' but VerifyOptions " - "did not supply expected_key_origins — the spec-mandated " - "identity.key_origins consistency check (ADCP #3690 step 7) " - "is silently skipped. Thread the operator's " - "identity.key_origins map through VerifyOptions(expected_key_origins=...) " - "to engage the check; pass an empty dict if the operator " - "advertises no map and you want the missing-declaration " - "rejection (request_signature_key_origin_missing) to fire.", - UserWarning, - stacklevel=2, - ) + if expected_key_origins is None and source != "brand_json": return if source != "brand_json": if source is None: @@ -598,7 +611,7 @@ def _maybe_check_key_origin( ) check_key_origin_consistency( jwks_uri=jwks_uri, - key_origins=expected_key_origins, + key_origins=expected_key_origins or {}, purpose=signing_purpose, posture=posture, ) diff --git a/src/adcp/signing/webhook_verifier.py b/src/adcp/signing/webhook_verifier.py index 75fcc833..3acbaf85 100644 --- a/src/adcp/signing/webhook_verifier.py +++ b/src/adcp/signing/webhook_verifier.py @@ -22,7 +22,7 @@ import logging import time from collections.abc import Callable, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from adcp.signing.canonical import _lookup, parse_signature_input_header from adcp.signing.constants import ( @@ -42,7 +42,7 @@ logger = logging.getLogger(__name__) from adcp.signing.jwks import JwksResolver -from adcp.signing.replay import ReplayStore +from adcp.signing.replay import InMemoryReplayStore, ReplayStore from adcp.signing.revocation import RevocationChecker, RevocationList from adcp.signing.verifier import ( VerifiedSigner, @@ -72,10 +72,15 @@ class WebhookVerifyOptions: verifier stamps time-of-check itself, so the same :class:`WebhookVerifyOptions` instance can live for the lifetime of your receiver without a factory closure around it. Override via ``clock=`` for deterministic tests. + + ``replay_store`` defaults to a per-options in-memory store so captured + signatures are rejected without extra configuration. Multi-process + receivers should supply a shared store. Passing ``None`` explicitly is + the opt-out for specialized tests or externally enforced replay policy. """ jwks_resolver: JwksResolver - replay_store: ReplayStore | None = None + replay_store: ReplayStore | None = field(default_factory=InMemoryReplayStore) revocation_checker: RevocationChecker | None = None revocation_list: RevocationList | None = None max_skew_seconds: int = DEFAULT_SKEW_SECONDS diff --git a/tests/conformance/decisioning/test_pg_idempotency_backend.py b/tests/conformance/decisioning/test_pg_idempotency_backend.py index d6619683..dacaff35 100644 --- a/tests/conformance/decisioning/test_pg_idempotency_backend.py +++ b/tests/conformance/decisioning/test_pg_idempotency_backend.py @@ -42,8 +42,11 @@ async def isolated_backend() -> AsyncIterator[PgBackend]: """Fresh async pool + isolated table per test. Drops on teardown.""" table = f"test_adcp_idem_{secrets.token_hex(6)}" - async with psycopg_pool.AsyncConnectionPool(TEST_URL, min_size=2, max_size=8) as pool: - backend = PgBackend(pool=pool, table_name=table) + async with ( + psycopg_pool.AsyncConnectionPool(TEST_URL, min_size=2, max_size=8) as pool, + psycopg_pool.AsyncConnectionPool(TEST_URL, min_size=2, max_size=8) as lock_pool, + ): + backend = PgBackend(pool=pool, lock_pool=lock_pool, table_name=table) await backend.create_schema() try: yield backend @@ -235,3 +238,38 @@ class Ctx: # AdCP L1/security rule 4 (#714): replay envelope carries ``replayed: true``. assert r2.get("replayed") is True assert {k: v for k, v in r2.items() if k != "replayed"} == r1 + + +@pytest.mark.asyncio +async def test_concurrent_wrapped_calls_execute_once(isolated_backend: PgBackend) -> None: + import asyncio + + store = IdempotencyStore(backend=isolated_backend, ttl_seconds=3600) + entered = asyncio.Event() + release = asyncio.Event() + calls = 0 + + @store.wrap + async def handler(self, params, context=None): + nonlocal calls + calls += 1 + entered.set() + await release.wait() + return {"task_id": "only", "status": "ok"} + + class Ctx: + caller_identity = "buyer-acme" + tenant_id = "tenant-1" + + params = {"idempotency_key": "shared-key", "x": 42} + first = asyncio.create_task(handler(None, params, Ctx())) + await entered.wait() + second = asyncio.create_task(handler(None, params, Ctx())) + await asyncio.sleep(0.05) + assert calls == 1 + release.set() + + first_result, second_result = await asyncio.gather(first, second) + assert calls == 1 + assert first_result.get("replayed") is not True + assert second_result["replayed"] is True diff --git a/tests/conformance/signing/test_bounded_fetches.py b/tests/conformance/signing/test_bounded_fetches.py new file mode 100644 index 00000000..62725388 --- /dev/null +++ b/tests/conformance/signing/test_bounded_fetches.py @@ -0,0 +1,89 @@ +"""Remote signing documents are bounded while streaming, not after buffering.""" + +from __future__ import annotations + +import httpx +import pytest + +from adcp.signing.jwks import async_default_jwks_fetcher, default_jwks_fetcher +from adcp.signing.revocation_fetcher import ( + RevocationListFetchError, + async_default_revocation_list_fetcher, + default_revocation_list_fetcher, +) + + +class _SyncChunks(httpx.SyncByteStream): + def __init__(self) -> None: + self.read = 0 + + def __iter__(self): # type: ignore[no-untyped-def] + for chunk in (b"xxxx", b"yyyy", b"zzzz"): + self.read += 1 + yield chunk + + +class _AsyncChunks(httpx.AsyncByteStream): + def __init__(self) -> None: + self.read = 0 + + async def __aiter__(self): # type: ignore[no-untyped-def] + for chunk in (b"xxxx", b"yyyy", b"zzzz"): + self.read += 1 + yield chunk + + +def test_sync_jwks_stops_reading_chunked_oversize(monkeypatch: pytest.MonkeyPatch) -> None: + stream = _SyncChunks() + transport = httpx.MockTransport(lambda request: httpx.Response(200, stream=stream)) + monkeypatch.setattr( + "adcp.signing.ip_pinned_transport.build_ip_pinned_transport", + lambda uri, **kwargs: transport, + ) + with pytest.raises(ValueError, match="exceeds 5 bytes"): + default_jwks_fetcher("https://keys.example/jwks", max_body_bytes=5) + assert stream.read == 2 + + +@pytest.mark.asyncio +async def test_async_jwks_stops_reading_chunked_oversize( + monkeypatch: pytest.MonkeyPatch, +) -> None: + stream = _AsyncChunks() + transport = httpx.MockTransport(lambda request: httpx.Response(200, stream=stream)) + monkeypatch.setattr( + "adcp.signing.ip_pinned_transport.build_async_ip_pinned_transport", + lambda uri, **kwargs: transport, + ) + with pytest.raises(ValueError, match="exceeds 5 bytes"): + await async_default_jwks_fetcher("https://keys.example/jwks", max_body_bytes=5) + assert stream.read == 2 + + +def test_sync_revocation_stops_reading_chunked_oversize( + monkeypatch: pytest.MonkeyPatch, +) -> None: + stream = _SyncChunks() + transport = httpx.MockTransport(lambda request: httpx.Response(200, stream=stream)) + monkeypatch.setattr( + "adcp.signing.ip_pinned_transport.build_ip_pinned_transport", + lambda uri, **kwargs: transport, + ) + with pytest.raises(RevocationListFetchError, match="exceeds 5 bytes"): + default_revocation_list_fetcher("https://gov.example/list", max_body_bytes=5) + assert stream.read == 2 + + +@pytest.mark.asyncio +async def test_async_revocation_stops_reading_chunked_oversize( + monkeypatch: pytest.MonkeyPatch, +) -> None: + stream = _AsyncChunks() + transport = httpx.MockTransport(lambda request: httpx.Response(200, stream=stream)) + monkeypatch.setattr( + "adcp.signing.ip_pinned_transport.build_async_ip_pinned_transport", + lambda uri, **kwargs: transport, + ) + with pytest.raises(RevocationListFetchError, match="exceeds 5 bytes"): + await async_default_revocation_list_fetcher("https://gov.example/list", max_body_bytes=5) + assert stream.read == 2 diff --git a/tests/conformance/signing/test_install_signing_event_hook.py b/tests/conformance/signing/test_install_signing_event_hook.py index e7893d94..0379ef17 100644 --- a/tests/conformance/signing/test_install_signing_event_hook.py +++ b/tests/conformance/signing/test_install_signing_event_hook.py @@ -289,7 +289,7 @@ def provider() -> RequestSigning | None: @pytest.mark.asyncio async def test_forbidden_covers_content_digest_omits_digest_coverage() -> None: - """Capability with covers_content_digest='forbidden' ⇒ signature must NOT cover content-digest.""" + """A forbidden digest policy must not cover content-digest.""" body = b'{"plan_id":"p1"}' request = httpx.Request( method="POST", @@ -386,3 +386,52 @@ async def existing_hook(_request: httpx.Request) -> None: assert pre_existing_called == [True] assert "Signature" in request.headers + + +@pytest.mark.asyncio +async def test_cross_origin_redirect_is_rejected_before_second_request() -> None: + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(str(request.url)) + if request.url.host == "seller.example.com": + return httpx.Response( + 307, + headers={"Location": "https://attacker.example.net/capture"}, + ) + return httpx.Response(200, json={"captured": True}) + + client = httpx.AsyncClient( + follow_redirects=True, + transport=httpx.MockTransport(handler), + ) + install_signing_event_hook( + client, + signing=_config(), + seller_capability=_capability(required=["create_media_buy"]), + expected_origin="https://seller.example.com", + ) + + with signing_operation("create_media_buy"), pytest.raises(ValueError, match="cross-origin"): + await client.post("https://seller.example.com/mcp", content=b"{}") + + assert seen == ["https://seller.example.com/mcp"] + await client.aclose() + + +@pytest.mark.asyncio +async def test_first_scoped_request_binds_origin_when_not_configured() -> None: + capability = _capability(required=["create_media_buy"]) + client = httpx.AsyncClient() + install_signing_event_hook(client, signing=_config(), seller_capability=capability) + [hook] = client.event_hooks["request"] + + seller_request = httpx.Request("POST", "https://seller.example.com/mcp", content=b"{}") + attacker_request = httpx.Request("POST", "https://attacker.example.net/x", content=b"{}") + with signing_operation("create_media_buy"): + await hook(seller_request) + with pytest.raises(ValueError, match="cross-origin"): + await hook(attacker_request) + + assert "Signature" in seller_request.headers + assert "Signature" not in attacker_request.headers diff --git a/tests/conformance/signing/test_jwks.py b/tests/conformance/signing/test_jwks.py index 9aae2ef7..a6237374 100644 --- a/tests/conformance/signing/test_jwks.py +++ b/tests/conformance/signing/test_jwks.py @@ -3,6 +3,7 @@ from __future__ import annotations import socket +import threading from typing import Any from unittest.mock import patch @@ -10,6 +11,7 @@ from adcp.signing import ( DEFAULT_ALLOWED_PORTS, + AsyncCachingJwksResolver, CachingJwksResolver, SignatureVerificationError, SSRFValidationError, @@ -388,6 +390,120 @@ def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: assert calls == 2 +def test_caching_resolver_revalidates_known_kid_after_max_age() -> None: + calls = 0 + + def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: + nonlocal calls + calls += 1 + return _make_jwks("k1") if calls == 1 else _make_jwks("replacement") + + clock = {"t": 0.0} + resolver = CachingJwksResolver( + "https://example.com/jwks.json", + fetcher=fetcher, + max_age_seconds=60.0, + clock=lambda: clock["t"], + ) + assert resolver("k1") is not None + clock["t"] = 61.0 + assert resolver("k1") is None + assert calls == 2 + + +def test_sync_caching_resolver_single_flights_concurrent_expiry_refresh() -> None: + calls = 0 + refresh_started = threading.Event() + release_refresh = threading.Event() + + def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: + nonlocal calls + del uri, allow_private + calls += 1 + if calls == 1: + return _make_jwks("k1") + refresh_started.set() + assert release_refresh.wait(timeout=5) + return _make_jwks("replacement") + + clock = {"t": 0.0} + resolver = CachingJwksResolver( + "https://example.com/jwks.json", + fetcher=fetcher, + max_age_seconds=60.0, + clock=lambda: clock["t"], + ) + assert resolver("k1") is not None + + class _ObservedLock: + def __init__(self) -> None: + self._lock = threading.Lock() + self._guard = threading.Lock() + self._attempts = 0 + self.second_attempted = threading.Event() + + def __enter__(self) -> None: + with self._guard: + self._attempts += 1 + if self._attempts == 2: + self.second_attempted.set() + self._lock.acquire() + + def __exit__(self, *args: object) -> None: + self._lock.release() + + observed_lock = _ObservedLock() + resolver._refresh_lock = observed_lock + clock["t"] = 61.0 + start = threading.Barrier(3) + results: list[dict[str, Any] | None] = [] + errors: list[BaseException] = [] + + def resolve_expired() -> None: + try: + start.wait() + results.append(resolver("replacement")) + except BaseException as exc: # pragma: no cover - asserted below + errors.append(exc) + + threads = [threading.Thread(target=resolve_expired) for _ in range(2)] + for thread in threads: + thread.start() + start.wait() + assert refresh_started.wait(timeout=5) + assert observed_lock.second_attempted.wait(timeout=5) + release_refresh.set() + for thread in threads: + thread.join(timeout=5) + + assert errors == [] + assert all(not thread.is_alive() for thread in threads) + assert len(results) == 2 + assert all(result is not None for result in results) + assert calls == 2 + + +async def test_async_caching_resolver_revalidates_known_kid_after_max_age() -> None: + calls = 0 + + async def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: + nonlocal calls + calls += 1 + return _make_jwks("k1") if calls == 1 else _make_jwks("replacement") + + clock = {"t": 0.0} + resolver = AsyncCachingJwksResolver( + "https://example.com/jwks.json", + fetcher=fetcher, + max_age_seconds=60.0, + clock=lambda: clock["t"], + ) + assert await resolver("k1") is not None + clock["t"] = 61.0 + assert await resolver("k1") is None + assert calls == 2 + + def test_caching_resolver_wraps_ssrf_as_untrusted() -> None: def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: raise SSRFValidationError("blocked") @@ -410,6 +526,28 @@ def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: assert exc.value.code == "request_signature_jwks_unavailable" +def test_caching_resolver_wraps_malformed_key_as_unavailable() -> None: + def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: + return {"keys": [1]} + + resolver = CachingJwksResolver("https://example.com/jwks.json", fetcher=fetcher) + with pytest.raises(SignatureVerificationError) as exc: + resolver("k1") + assert exc.value.code == "request_signature_jwks_unavailable" + assert isinstance(exc.value.__cause__, ValueError) + + +async def test_async_caching_resolver_wraps_malformed_key_as_unavailable() -> None: + async def fetcher(uri: str, *, allow_private: bool = False) -> dict[str, Any]: + return {"keys": [1]} + + resolver = AsyncCachingJwksResolver("https://example.com/jwks.json", fetcher=fetcher) + with pytest.raises(SignatureVerificationError) as exc: + await resolver("k1") + assert exc.value.code == "request_signature_jwks_unavailable" + assert isinstance(exc.value.__cause__, ValueError) + + # ---- StaticJwksResolver ---- diff --git a/tests/conformance/signing/test_pg_replay_store.py b/tests/conformance/signing/test_pg_replay_store.py index eb0bf5ea..dfd0cdc0 100644 --- a/tests/conformance/signing/test_pg_replay_store.py +++ b/tests/conformance/signing/test_pg_replay_store.py @@ -174,19 +174,17 @@ def test_sweep_expired_returns_zero_when_clean(isolated_pool) -> None: # -- concurrency ----------------------------------------------------- -def test_concurrent_remember_same_nonce_is_idempotent(isolated_pool) -> None: - """Two workers racing on the same (keyid, nonce) MUST NOT error. - - ``ON CONFLICT ... DO UPDATE`` makes the second insert a no-op - (with refreshed TTL). Without it, the second worker would hit a - PK violation and blow up. - """ +def test_concurrent_claim_same_nonce_has_one_winner(isolated_pool) -> None: + """Only one verifier process may claim a nonce.""" store = _store(isolated_pool) errors: list[Exception] = [] + results: list[str] = [] + barrier = threading.Barrier(10) def worker() -> None: try: - store.remember("k", "shared", ttl_seconds=60) + barrier.wait() + results.append(store.claim("k", "shared", ttl_seconds=60)) except Exception as exc: # noqa: BLE001 errors.append(exc) @@ -197,6 +195,8 @@ def worker() -> None: t.join() assert errors == [] + assert results.count("claimed") == 1 + assert results.count("replayed") == 9 assert store.seen("k", "shared") is True assert store.live_count("k") == 1 diff --git a/tests/conformance/signing/test_replay.py b/tests/conformance/signing/test_replay.py index 7ff97d94..2a2dded3 100644 --- a/tests/conformance/signing/test_replay.py +++ b/tests/conformance/signing/test_replay.py @@ -38,6 +38,25 @@ def worker(tid: int) -> None: assert store._counts["kid"] == thread_count * nonces_per_thread +def test_concurrent_claim_has_exactly_one_winner() -> None: + store = InMemoryReplayStore() + barrier = threading.Barrier(16) + results: list[str] = [] + + def worker() -> None: + barrier.wait() + results.append(store.claim("kid", "shared", ttl_seconds=60.0)) + + threads = [threading.Thread(target=worker) for _ in range(16)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert results.count("claimed") == 1 + assert results.count("replayed") == 15 + + def test_monotonic_clock_ttl_expires_and_decrements_count() -> None: clock = _FakeClock(now=0.0) store = InMemoryReplayStore(per_keyid_cap=5, clock=clock) @@ -83,3 +102,36 @@ def test_mark_cap_hit_still_works() -> None: store = InMemoryReplayStore() store.mark_cap_hit("kid") assert store.at_capacity("kid") is True + + +def test_global_cap_bounds_rotating_keyids_and_recovers_after_expiry() -> None: + clock = _FakeClock(now=0.0) + store = InMemoryReplayStore(per_keyid_cap=10, global_cap=3, clock=clock) + assert store.claim("kid-1", "n", 10.0) == "claimed" + assert store.claim("kid-2", "n", 10.0) == "claimed" + assert store.claim("kid-3", "n", 10.0) == "claimed" + assert store.claim("kid-4", "n", 10.0) == "capacity" + assert len(store._entries) == 3 + + clock.now = 11.0 + assert store.claim("kid-4", "n", 10.0) == "claimed" + assert len(store._entries) == 1 + + +def test_claim_does_not_copy_the_entry_table() -> None: + class _NoItemsDict(dict[tuple[str, str], float]): + def items(self): # type: ignore[no-untyped-def] + raise AssertionError("accepted claim copied/scanned the full entry table") + + store = InMemoryReplayStore(global_cap=100) + store._entries = _NoItemsDict() + for index in range(50): + assert store.claim(f"kid-{index}", "nonce", 60.0) == "claimed" + + +def test_renewal_heap_remains_bounded() -> None: + store = InMemoryReplayStore(global_cap=3) + for index in range(50): + store.remember("kid", "nonce", float(index + 1)) + assert len(store._entries) == 1 + assert len(store._expiry_heap) <= 6 diff --git a/tests/conformance/signing/test_revocation_fetcher.py b/tests/conformance/signing/test_revocation_fetcher.py index aaf51271..5babe8a8 100644 --- a/tests/conformance/signing/test_revocation_fetcher.py +++ b/tests/conformance/signing/test_revocation_fetcher.py @@ -336,9 +336,7 @@ def test_accepts_future_version_with_forward_compat() -> None: # version=2 should NOT hard-reject: additive schema changes shouldn't # force every old SDK into fail-closed across their entire traffic. private, _, jwks_resolver = _key_and_jwks() - token = _sign_jws_compact( - _make_payload(version=2, revoked_kids=["rev"]), private=private - ) + token = _sign_jws_compact(_make_payload(version=2, revoked_kids=["rev"]), private=private) fetcher = _ScriptedFetcher() fetcher.enqueue(FetchResult(body=token, etag=None, not_modified=False)) @@ -546,12 +544,12 @@ def test_replay_older_list_rejected() -> None: revoked_kids=[], # attacker un-revokes the kid ) fetcher = _ScriptedFetcher() - fetcher.enqueue(FetchResult( - body=_sign_jws_compact(newer, private=private), etag='"v2"', not_modified=False - )) - fetcher.enqueue(FetchResult( - body=_sign_jws_compact(older, private=private), etag='"v1"', not_modified=False - )) + fetcher.enqueue( + FetchResult(body=_sign_jws_compact(newer, private=private), etag='"v2"', not_modified=False) + ) + fetcher.enqueue( + FetchResult(body=_sign_jws_compact(older, private=private), etag='"v1"', not_modified=False) + ) wall_clock, mono_clock, advance = _controllable_clock( datetime(2026, 4, 18, 14, 15, tzinfo=timezone.utc) @@ -693,12 +691,14 @@ def test_if_modified_since_threaded_to_fetcher() -> None: private, _, jwks_resolver = _key_and_jwks() token = _sign_jws_compact(_make_payload(), private=private) fetcher = _ScriptedFetcher() - fetcher.enqueue(FetchResult( - body=token, - etag='"v1"', - last_modified="Sat, 18 Apr 2026 14:00:00 GMT", - not_modified=False, - )) + fetcher.enqueue( + FetchResult( + body=token, + etag='"v1"', + last_modified="Sat, 18 Apr 2026 14:00:00 GMT", + not_modified=False, + ) + ) fetcher.enqueue(FetchResult(body="", etag='"v1"', not_modified=True)) wall_clock, mono_clock, advance = _controllable_clock( @@ -819,19 +819,14 @@ def test_last_modified_header_injection_rejected() -> None: assert _sanitize_last_modified(None) is None -def test_304_slides_next_update_forward() -> None: - """Round-2: successive 304s advance the cached next_update so the - checker doesn't hit the refresh-cooldown path on every call past the - original next_update.""" +def test_304_does_not_extend_signed_next_update() -> None: + """Transport validators cannot extend a signed authorization window.""" private, _, jwks_resolver = _key_and_jwks() token = _sign_jws_compact(_make_payload(), private=private) fetcher = _ScriptedFetcher() fetcher.enqueue(FetchResult(body=token, etag='"v1"', not_modified=False)) fetcher.enqueue(FetchResult(body="", etag='"v1"', not_modified=True)) - # If next_update wasn't advanced on 304, the next two calls past 14:30 - # would each try to refetch (subject to the 60s cooldown). We only - # queue ONE more fetcher response, so if the invariant breaks, one of - # the later calls raises AssertionError from the scripted fetcher. + fetcher.enqueue(FetchResult(body="", etag='"v1"', not_modified=True)) wall_clock, mono_clock, advance = _controllable_clock( datetime(2026, 4, 18, 14, 1, tzinfo=timezone.utc) @@ -846,16 +841,34 @@ def test_304_slides_next_update_forward() -> None: ) checker("k") # 1 fetch → initial advance(15 * 60 + 30) # 14:16:30, past original next_update (14:15) - checker("k") # 2 fetches → 304, should slide next_update to 14:30 + checker("k") # within grace, but the signed deadline remains unchanged - # Now at 14:16:30. Cached next_update was 14:15, now should be 14:30. assert checker._current_list is not None - assert checker._current_list.next_update.startswith("2026-04-18T14:30:00") + assert checker._current_list.next_update == "2026-04-18T14:15:00Z" - # Additional calls WITHIN the new window should NOT refetch. - advance(60) # 14:17:30 - checker("k") # still no fetch — we're before the new 14:30 next_update - assert len(fetcher.calls) == 2 + advance(29 * 60) # beyond signed next_update + 2x interval grace + with pytest.raises(RevocationListFreshnessError): + checker("k") + + +def test_cold_checker_rejects_already_expired_signed_list() -> None: + private, _, jwks_resolver = _key_and_jwks() + token = _sign_jws_compact(_make_payload(), private=private) + fetcher = _ScriptedFetcher() + fetcher.enqueue(FetchResult(body=token, etag='"v1"', not_modified=False)) + wall_clock, mono_clock, _ = _controllable_clock( + datetime(2026, 4, 18, 14, 16, tzinfo=timezone.utc) + ) + checker = CachingRevocationChecker( + revocation_uri=REVOCATION_URI, + issuer=ISSUER, + jwks_resolver=jwks_resolver, + fetcher=fetcher, + wall_clock=wall_clock, + clock=mono_clock, + ) + with pytest.raises(RevocationListParseError, match="already expired"): + checker("k") def test_clock_footgun_rejects_time_time() -> None: diff --git a/tests/conformance/signing/test_verifier_behaviors.py b/tests/conformance/signing/test_verifier_behaviors.py index 6432079c..0cf4bb49 100644 --- a/tests/conformance/signing/test_verifier_behaviors.py +++ b/tests/conformance/signing/test_verifier_behaviors.py @@ -15,6 +15,7 @@ import pytest from adcp.signing import ( + REQUEST_SIGNATURE_COMPONENTS_INCOMPLETE, REQUEST_SIGNATURE_COMPONENTS_UNEXPECTED, REQUEST_SIGNATURE_HEADER_MALFORMED, REQUEST_SIGNATURE_REVOCATION_STALE, @@ -346,11 +347,76 @@ def test_verify_options_rejects_positional() -> None: # ---- 6a: VerifierCapability default ---- -def test_verifier_capability_defaults_to_either_digest() -> None: - """The AdCP 3.0 schema declares ``covers_content_digest`` default as - ``"either"`` (``get-adcp-capabilities-response.json``); ``"required"`` - is opt-in for spend-committing operations. AdCP 4.0 is expected to - recommend ``"required"`` more broadly. Operators promote operations - selectively via ``required_for=frozenset({"create_media_buy", ...})``.""" +def test_verifier_capability_defaults_to_required_digest() -> None: + """Omitting policy must fail closed even though the 3.0 wire default is + ``"either"``; callers mirroring that capability pass it explicitly.""" cap = VerifierCapability() - assert cap.covers_content_digest == "either" + assert cap.covers_content_digest == "required" + + +def test_default_capability_rejects_signature_that_does_not_bind_body() -> None: + headers, body = _sign_basic() + options = VerifyOptions( + now=1776520800.0, + capability=VerifierCapability(), + operation="create_media_buy", + jwks_resolver=StaticJwksResolver({"keys": [ED25519_KEY]}), + ) + + with pytest.raises(SignatureVerificationError) as exc: + verify_request_signature( + method="POST", + url="https://seller.example.com/adcp/create_media_buy", + headers=headers, + body=body, + options=options, + ) + + assert exc.value.code == REQUEST_SIGNATURE_COMPONENTS_INCOMPLETE + assert exc.value.step == 6 + + +def test_legacy_replay_store_warns_and_remains_compatible() -> None: + class LegacyReplayStore: + def __init__(self) -> None: + self.entries: set[tuple[str, str]] = set() + + def seen(self, keyid: str, nonce: str) -> bool: + return (keyid, nonce) in self.entries + + def remember(self, keyid: str, nonce: str, ttl_seconds: float) -> None: + del ttl_seconds + self.entries.add((keyid, nonce)) + + def at_capacity(self, keyid: str) -> bool: + del keyid + return False + + headers, body = _sign_basic() + store = LegacyReplayStore() + options = VerifyOptions( + now=1776520800.0, + capability=VerifierCapability(covers_content_digest="either"), + operation="create_media_buy", + jwks_resolver=StaticJwksResolver({"keys": [ED25519_KEY]}), + replay_store=store, + ) + + with pytest.warns(RuntimeWarning, match=r"does not implement atomic claim\(\)"): + verify_request_signature( + method="POST", + url="https://seller.example.com/adcp/create_media_buy", + headers=headers, + body=body, + options=options, + ) + + with pytest.warns(RuntimeWarning), pytest.raises(SignatureVerificationError) as exc: + verify_request_signature( + method="POST", + url="https://seller.example.com/adcp/create_media_buy", + headers=headers, + body=body, + options=options, + ) + assert exc.value.code == "request_signature_replayed" diff --git a/tests/conformance/signing/test_verifier_key_origins.py b/tests/conformance/signing/test_verifier_key_origins.py index 2a00eebd..ca707d29 100644 --- a/tests/conformance/signing/test_verifier_key_origins.py +++ b/tests/conformance/signing/test_verifier_key_origins.py @@ -291,11 +291,8 @@ def test_resolver_without_source_attribute_skips_check() -> None: # ----- check does not fire without expected_key_origins ----- -def test_brand_json_source_skips_check_when_no_expected_origins() -> None: - """``expected_key_origins=None`` (default) → check skips even on a - brand-json-sourced resolver. Adopters who haven't yet plumbed - capabilities through the verifier see no behavior change. - """ +def test_brand_json_source_rejects_when_no_expected_origins() -> None: + """A brand-sourced resolver cannot omit its mandatory origin binding.""" headers, body = _sign_basic() resolver = _BrandJsonStaticResolver( # Even with a mismatched jwks_uri, the check skips when the @@ -304,13 +301,16 @@ def test_brand_json_source_skips_check_when_no_expected_origins() -> None: jwks_uri="https://different.example/.well-known/jwks.json", ) options = _options_with(resolver, expected_key_origins=None) - verify_request_signature( - method="POST", - url="https://seller.example.com/adcp/create_media_buy", - headers=headers, - body=body, - options=options, - ) + with pytest.raises(SignatureVerificationError) as exc_info: + verify_request_signature( + method="POST", + url="https://seller.example.com/adcp/create_media_buy", + headers=headers, + body=body, + options=options, + ) + assert exc_info.value.code == REQUEST_SIGNATURE_KEY_ORIGIN_MISSING + assert exc_info.value.step == 7 # ----- earlier failure codes still surface ----- diff --git a/tests/conformance/signing/test_webhook_dedup.py b/tests/conformance/signing/test_webhook_dedup.py index b35255c9..b2ca3f87 100644 --- a/tests/conformance/signing/test_webhook_dedup.py +++ b/tests/conformance/signing/test_webhook_dedup.py @@ -2,9 +2,16 @@ from __future__ import annotations +import asyncio + import pytest -from adcp.server.idempotency import MemoryBackend, WebhookDedupStore +from adcp.server.idempotency import ( + CachedResponse, + IdempotencyBackend, + MemoryBackend, + WebhookDedupStore, +) @pytest.fixture @@ -23,6 +30,45 @@ async def test_repeat_returns_false(store: WebhookDedupStore) -> None: assert await store.check_and_record("sender-1", "whk_abc") is False +@pytest.mark.asyncio +async def test_legacy_backend_warns_and_preserves_repeat_dedup() -> None: + class LegacyBackend(IdempotencyBackend): + def __init__(self) -> None: + self.entries: dict[tuple[str, str], CachedResponse] = {} + + async def get(self, scope_key: str, key: str) -> CachedResponse | None: + return self.entries.get((scope_key, key)) + + async def put(self, scope_key: str, key: str, entry: CachedResponse) -> None: + self.entries[(scope_key, key)] = entry + + async def delete_expired(self, now_epoch: float | None = None) -> int: + return 0 + + store = WebhookDedupStore(LegacyBackend()) + with pytest.warns(DeprecationWarning, match="process-local webhook dedup locking"): + assert await store.check_and_record("sender-1", "whk_legacy") is True + assert await store.check_and_record("sender-1", "whk_legacy") is False + + +@pytest.mark.asyncio +async def test_concurrent_deliveries_have_exactly_one_first_seen( + store: WebhookDedupStore, +) -> None: + gate = asyncio.Event() + + async def deliver() -> bool: + await gate.wait() + return await store.check_and_record("sender-1", "whk_shared") + + tasks = [asyncio.create_task(deliver()) for _ in range(20)] + gate.set() + results = await asyncio.gather(*tasks) + + assert results.count(True) == 1 + assert results.count(False) == 19 + + @pytest.mark.asyncio async def test_different_senders_independent(store: WebhookDedupStore) -> None: """Per-sender scoping: the same key from a different sender is fresh.""" diff --git a/tests/conformance/signing/test_webhook_receiver.py b/tests/conformance/signing/test_webhook_receiver.py index 661f8d21..4de12645 100644 --- a/tests/conformance/signing/test_webhook_receiver.py +++ b/tests/conformance/signing/test_webhook_receiver.py @@ -107,7 +107,10 @@ async def test_duplicate_detected() -> None: receiver = _build_receiver() first = await receiver.receive(method="POST", url=URL, headers=headers, body=body) - second = await receiver.receive(method="POST", url=URL, headers=headers, body=body) + # A retry is freshly signed (new signature nonce) while retaining the + # payload idempotency key. Reusing the captured signature itself is now + # rejected by the verifier before payload dedup. + second = await receiver.receive(method="POST", url=URL, headers=_sign_webhook(body), body=body) assert first.duplicate is False assert second.duplicate is True diff --git a/tests/conformance/signing/test_webhook_signer.py b/tests/conformance/signing/test_webhook_signer.py index c48b5cce..26cfd690 100644 --- a/tests/conformance/signing/test_webhook_signer.py +++ b/tests/conformance/signing/test_webhook_signer.py @@ -21,6 +21,7 @@ ) from adcp.signing.errors import ( WEBHOOK_SIGNATURE_KEY_PURPOSE_INVALID, + WEBHOOK_SIGNATURE_REPLAYED, WEBHOOK_SIGNATURE_REQUIRED, WEBHOOK_SIGNATURE_TAG_INVALID, SignatureVerificationError, @@ -86,6 +87,47 @@ def test_sign_then_verify_roundtrip() -> None: assert result.alg == "ed25519" +def test_default_replay_store_rejects_captured_signature() -> None: + body = b'{"idempotency_key":"whk_replay","task_id":"t1"}' + headers = _sign_and_headers(body) + options = _webhook_verify_options([WEBHOOK_ED25519]) + + verify_webhook_signature( + method="POST", + url="https://buyer.example.com/webhooks/adcp", + headers=headers, + body=body, + options=options, + ) + with pytest.raises(SignatureVerificationError) as exc_info: + verify_webhook_signature( + method="POST", + url="https://buyer.example.com/webhooks/adcp", + headers=headers, + body=body, + options=options, + ) + assert exc_info.value.code == WEBHOOK_SIGNATURE_REPLAYED + + +def test_explicit_none_opts_out_of_signature_replay_check() -> None: + body = b'{"idempotency_key":"whk_external_dedup","task_id":"t1"}' + headers = _sign_and_headers(body) + options = WebhookVerifyOptions( + jwks_resolver=StaticJwksResolver({"keys": [WEBHOOK_ED25519]}), + replay_store=None, + ) + + for _ in range(2): + verify_webhook_signature( + method="POST", + url="https://buyer.example.com/webhooks/adcp", + headers=headers, + body=body, + options=options, + ) + + def test_rejects_request_signing_key() -> None: """adcp_use='request-signing' MUST NOT verify as a webhook.""" body = b'{"idempotency_key":"whk_abc","task_id":"t1"}' diff --git a/tests/test_agent_resolver.py b/tests/test_agent_resolver.py index 10279c6d..764c7a06 100644 --- a/tests/test_agent_resolver.py +++ b/tests/test_agent_resolver.py @@ -46,6 +46,12 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: if url not in self.responses: return httpx.Response(404, content=b"") spec = self.responses[url] + if "stream" in spec: + return httpx.Response( + spec.get("status", 200), + stream=spec["stream"], + headers=spec.get("headers", {}), + ) return httpx.Response( spec.get("status", 200), content=spec.get("body", b""), @@ -53,6 +59,17 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: ) +class _ChunkedStream(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]) -> None: + self.chunks = chunks + self.read = 0 + + async def __aiter__(self): # type: ignore[no-untyped-def] + for chunk in self.chunks: + self.read += 1 + yield chunk + + @pytest.fixture def patch_resolver(monkeypatch: pytest.MonkeyPatch): """Wire a single ``_MockTransport`` into every hop of the resolver. @@ -355,6 +372,28 @@ async def test_resolve_rejects_oversize_capabilities_body(patch_resolver) -> Non assert "exceeds" in exc.value.message +@pytest.mark.asyncio +async def test_resolve_stops_streaming_oversize_capabilities_body(patch_resolver) -> None: + stream = _ChunkedStream([b"xxxx", b"yyyy", b"zzzz"]) + _, factory = patch_resolver( + { + "https://buyer.example.com/mcp": { + "stream": stream, + "headers": {"content-type": "application/json"}, + } + } + ) + with pytest.raises(AgentResolverError) as exc: + await async_resolve_agent( + "https://buyer.example.com/mcp", + agent_type="sales", + max_capabilities_bytes=5, + _capabilities_client_factory=factory, + ) + assert exc.value.code == "capabilities_invalid" + assert stream.read == 2 + + # ---- Sync wrapper ---- diff --git a/tests/test_brand_authz.py b/tests/test_brand_authz.py index 530f8ab8..76b6b90f 100644 --- a/tests/test_brand_authz.py +++ b/tests/test_brand_authz.py @@ -117,6 +117,36 @@ async def test_authz_etld1_match_authorizes_same_origin_agent() -> None: assert result.matched_agent_type == "signals" +@pytest.mark.asyncio +async def test_authz_stale_on_error_is_bounded() -> None: + url = "https://brand.com/.well-known/brand.json" + body = _brand_json({"agents": [{"type": "signals", "url": "https://ads.brand.com/signals"}]}) + transport = _MockTransport({url: {"body": body}}) + clock = {"t": 0.0} + resolver = BrandJsonAuthorizationResolver( + url, + max_age_seconds=10.0, + max_stale_seconds=20.0, + min_cooldown_seconds=0.0, + clock=lambda: clock["t"], + _client_factory=_factory(transport), + ) + kwargs = { + "agent_url": "https://ads.brand.com/signals", + "brand_domain": "brand.com", + } + assert (await resolver.check(**kwargs)).authorized + + transport.responses[url] = {"status": 503} + clock["t"] = 11.0 + assert (await resolver.check(**kwargs)).authorized + + clock["t"] = 31.0 + result = await resolver.check(**kwargs) + assert result.authorized is False + assert result.reason == "brand_json_unavailable" + + @pytest.mark.asyncio async def test_authz_etld1_match_with_subdomain_brand_url() -> None: body = _brand_json( diff --git a/tests/test_brand_jwks.py b/tests/test_brand_jwks.py index ed8747b7..78e3b6fa 100644 --- a/tests/test_brand_jwks.py +++ b/tests/test_brand_jwks.py @@ -47,6 +47,12 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: if url not in self.responses: return httpx.Response(404, content=b"") spec = self.responses[url] + if "stream" in spec: + return httpx.Response( + spec.get("status", 200), + stream=spec["stream"], + headers=spec.get("headers", {}), + ) # Return 304 when the request's If-None-Match matches the spec. if spec.get("etag") is not None and request.headers.get("if-none-match") == spec["etag"]: return httpx.Response( @@ -60,6 +66,17 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: ) +class _ChunkedStream(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]) -> None: + self.chunks = chunks + self.read = 0 + + async def __aiter__(self): # type: ignore[no-untyped-def] + for chunk in self.chunks: + self.read += 1 + yield chunk + + @pytest.fixture def patch_httpx(monkeypatch): """Inject a fake-transport ``client_factory`` into every @@ -453,6 +470,49 @@ async def test_resolver_fetches_brand_json_and_inner_jwks(patch_httpx) -> None: assert resolver.agent_url == "https://x.example/" +@pytest.mark.asyncio +async def test_resolver_reselects_when_body_changes_without_etag(patch_httpx) -> None: + url = "https://example.com/.well-known/brand.json" + responses = { + url: { + "body": _brand_json("https://x.example/", "https://x.example/old-jwks"), + "headers": {"content-type": "application/json"}, + } + } + transport = patch_httpx(responses) + clock = {"t": 0.0} + resolver = BrandJsonJwksResolver( + url, + agent_type="brand", + max_age_seconds=10.0, + min_cooldown_seconds=0.0, + clock=lambda: clock["t"], + jwks_fetcher=_jwks_fetcher_for( + { + "https://x.example/old-jwks": { + "kty": "OKP", + "crv": "Ed25519", + "x": "old", + "kid": "k1", + }, + "https://x.example/new-jwks": { + "kty": "OKP", + "crv": "Ed25519", + "x": "new", + "kid": "k1", + }, + } + ), + ) + assert (await resolver("k1"))["x"] == "old" # type: ignore[index] + + transport.responses[url]["body"] = _brand_json( + "https://x.example/", "https://x.example/new-jwks" + ) + clock["t"] = 11.0 + assert (await resolver("k1"))["x"] == "new" # type: ignore[index] + + @pytest.mark.asyncio async def test_resolver_returns_none_for_unknown_kid(patch_httpx) -> None: patch_httpx( @@ -727,6 +787,27 @@ async def test_resolver_rejects_oversized_brand_json(patch_httpx) -> None: assert "exceeds" in str(exc.value) +@pytest.mark.asyncio +async def test_resolver_stops_streaming_oversized_brand_json(patch_httpx) -> None: + stream = _ChunkedStream([b"xxxx", b"yyyy", b"zzzz"]) + patch_httpx( + { + "https://example.com/.well-known/brand.json": { + "stream": stream, + "headers": {"content-type": "application/json"}, + } + } + ) + resolver = BrandJsonJwksResolver( + "https://example.com/.well-known/brand.json", + agent_type="brand", + max_body_bytes=5, + ) + with pytest.raises(BrandJsonResolverError, match="exceeds 5 bytes"): + await resolver("k1") + assert stream.read == 2 + + @pytest.mark.asyncio async def test_resolver_loop_detection_handles_case_aliasing(patch_httpx) -> None: """Review finding #2 — without host lowercase + port-strip, diff --git a/tests/test_canonical_reference_resolver.py b/tests/test_canonical_reference_resolver.py index de4514a8..7c9e761e 100644 --- a/tests/test_canonical_reference_resolver.py +++ b/tests/test_canonical_reference_resolver.py @@ -443,6 +443,39 @@ def test_excessive_ref_count_is_rejected(monkeypatch: pytest.MonkeyPatch) -> Non assert result.message == "format_schema exceeds $ref count bound" +def test_excessive_schema_ids_stop_before_dns_amplification( + monkeypatch: pytest.MonkeyPatch, +) -> None: + dns_calls = 0 + + def counted_getaddrinfo(host: str, port: int, *args: Any, **kwargs: Any) -> list[Any]: + nonlocal dns_calls + del host, args, kwargs + dns_calls += 1 + return [(2, 1, 6, "", ("93.184.216.34", port))] + + monkeypatch.setattr("socket.getaddrinfo", counted_getaddrinfo) + document = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "allOf": [{"$id": f"child-{index}"} for index in range(100)], + } + body = json.dumps(document).encode() + resolver = CanonicalReferenceResolver( + max_schema_ids=2, + transport_factory=lambda _host, _ip: httpx.MockTransport( + lambda _request: httpx.Response(200, content=body) + ), + ) + + result = resolver.resolve_format_schema(_reference(body)) + + assert result.status is CanonicalReferenceStatus.INVALID_SCHEMA + assert result.message == "format_schema exceeds $id count bound" + # One lookup validates the fetched document URL, then at most the two + # configured $id values are resolved before the walker stops. + assert dns_calls <= 3 + + def test_deep_schema_nesting_returns_structured_invalid_schema( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_pg_idempotency_backend.py b/tests/test_pg_idempotency_backend.py index 453c6f2e..8c608948 100644 --- a/tests/test_pg_idempotency_backend.py +++ b/tests/test_pg_idempotency_backend.py @@ -16,6 +16,8 @@ ``expires_at > now()``). * ``put`` upserts with ``ON CONFLICT DO UPDATE``; serializes response via json.dumps; converts epoch to tz-aware datetime. +* ``hold`` uses a distinct advisory-lock pool so handler SQL cannot deadlock + against the ordinary cache/business pool. * ``delete_expired`` returns the rowcount. """ @@ -72,6 +74,11 @@ async def _connection(): return pool +def _pg_backend(pool: Any, **kwargs: Any) -> PgBackend: + """Build with a distinct lock pool for tests that do not exercise hold().""" + return PgBackend(pool=pool, lock_pool=MagicMock(), **kwargs) + + # --------------------------------------------------------------------------- # Construction # --------------------------------------------------------------------------- @@ -80,27 +87,32 @@ async def _connection(): class TestConstruction: def test_default_table_name(self) -> None: pool = MagicMock() - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) assert backend._table == DEFAULT_IDEMPOTENCY_TABLE def test_custom_table_name_accepted(self) -> None: pool = MagicMock() - backend = PgBackend(pool=pool, table_name="my_idem_cache") + backend = _pg_backend(pool, table_name="my_idem_cache") assert backend._table == "my_idem_cache" def test_invalid_identifier_rejected(self) -> None: pool = MagicMock() with pytest.raises(ValueError, match="Table name must match"): - PgBackend(pool=pool, table_name="bad-name") + _pg_backend(pool, table_name="bad-name") def test_uppercase_identifier_rejected(self) -> None: with pytest.raises(ValueError, match="Table name must match"): - PgBackend(pool=MagicMock(), table_name="MyTable") + _pg_backend(MagicMock(), table_name="MyTable") def test_satisfies_idempotency_backend_protocol(self) -> None: - backend = PgBackend(pool=MagicMock()) + backend = _pg_backend(MagicMock()) assert isinstance(backend, IdempotencyBackend) + def test_rejects_shared_lock_pool(self) -> None: + pool = MagicMock() + with pytest.raises(ValueError, match="lock_pool must be distinct"): + PgBackend(pool=pool, lock_pool=pool) + # --------------------------------------------------------------------------- # create_schema @@ -113,7 +125,7 @@ async def test_create_schema_executes_create_table_and_index() -> None: separate ``execute()`` call (psycopg does not split on ``;``).""" conn = _make_conn(_cursor(), _cursor()) pool = _make_pool(conn) - backend = PgBackend(pool=pool, table_name="adcp_idempotency") + backend = _pg_backend(pool, table_name="adcp_idempotency") await backend.create_schema() @@ -130,7 +142,7 @@ async def test_create_schema_executes_create_table_and_index() -> None: async def test_create_schema_uses_custom_table_name() -> None: conn = _make_conn(_cursor(), _cursor()) pool = _make_pool(conn) - backend = PgBackend(pool=pool, table_name="alt_idem") + backend = _pg_backend(pool, table_name="alt_idem") await backend.create_schema() assert "CREATE TABLE IF NOT EXISTS alt_idem" in conn.execute.call_args_list[0].args[0] @@ -148,7 +160,7 @@ async def test_create_schema_uses_custom_table_name() -> None: async def test_get_returns_none_on_miss() -> None: conn = _make_conn(_cursor(fetchone_value=None)) pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) assert await backend.get("scope-x", "key-y") is None sql = conn.execute.call_args.args[0] @@ -161,7 +173,7 @@ async def test_get_parses_dict_response() -> None: expires = datetime(2030, 1, 1, tzinfo=timezone.utc) conn = _make_conn(_cursor(fetchone_value=("hash-1", {"k": "v"}, expires))) pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) cached = await backend.get("scope-a", "key-1") assert cached is not None @@ -176,7 +188,7 @@ async def test_get_parses_json_string_response() -> None: expires = datetime(2030, 1, 1, tzinfo=timezone.utc) conn = _make_conn(_cursor(fetchone_value=("hash-1", '{"k":"v"}', expires))) pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) cached = await backend.get("scope-a", "key-1") assert cached is not None @@ -191,7 +203,7 @@ async def test_get_raises_on_naive_timestamp() -> None: naive = datetime(2030, 1, 1) conn = _make_conn(_cursor(fetchone_value=("hash", {}, naive))) pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) with pytest.raises(ValueError, match="naive datetime"): await backend.get("scope", "key") @@ -206,7 +218,7 @@ async def test_get_raises_on_naive_timestamp() -> None: async def test_put_upserts_with_on_conflict() -> None: conn = _make_conn(_cursor()) pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) entry = CachedResponse( payload_hash="hash-1", @@ -231,6 +243,43 @@ async def test_put_upserts_with_on_conflict() -> None: assert expires_at.tzinfo is not None # tz-aware +@pytest.mark.asyncio +async def test_put_if_absent_reports_atomic_insert_result() -> None: + conn = _make_conn(_cursor(fetchone_value=(1,))) + backend = _pg_backend(_make_pool(conn)) + entry = CachedResponse("hash", {}, time.time() + 3600) + + assert await backend.put_if_absent("scope", "key", entry) is True + sql = conn.execute.call_args.args[0] + assert "ON CONFLICT (scope_key, key) DO UPDATE" in sql + assert "RETURNING 1" in sql + + +@pytest.mark.asyncio +async def test_hold_reuses_locked_connection_for_get_and_put() -> None: + expires = datetime(2030, 1, 1, tzinfo=timezone.utc) + conn = _make_conn( + _cursor(), # advisory lock + _cursor(fetchone_value=None), # get + _cursor(), # put + ) + + @asynccontextmanager + async def transaction(): + yield + + conn.transaction = transaction + lock_pool = _make_pool(conn) + backend = PgBackend(pool=MagicMock(), lock_pool=lock_pool) + + async with backend.hold("scope", "key"): + assert await backend.get("scope", "key") is None + await backend.put("scope", "key", CachedResponse("hash", {}, expires.timestamp())) + + assert conn.execute.call_count == 3 + assert "pg_advisory_xact_lock" in conn.execute.call_args_list[0].args[0] + + # --------------------------------------------------------------------------- # delete_expired # --------------------------------------------------------------------------- @@ -240,7 +289,7 @@ async def test_put_upserts_with_on_conflict() -> None: async def test_delete_expired_uses_supplied_cutoff() -> None: conn = _make_conn(_cursor(rowcount=7)) pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) cutoff_epoch = 1_000_000_000.0 deleted = await backend.delete_expired(cutoff_epoch) @@ -258,7 +307,7 @@ async def test_delete_expired_uses_supplied_cutoff() -> None: async def test_delete_expired_defaults_to_wall_clock() -> None: conn = _make_conn(_cursor(rowcount=0)) pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) before = time.time() deleted = await backend.delete_expired() @@ -276,6 +325,6 @@ async def test_delete_expired_returns_zero_on_no_rowcount() -> None: backend coerces to 0.""" conn = _make_conn(_cursor(rowcount=None)) # type: ignore[arg-type] pool = _make_pool(conn) - backend = PgBackend(pool=pool) + backend = _pg_backend(pool) assert await backend.delete_expired() == 0 diff --git a/tests/test_server_idempotency.py b/tests/test_server_idempotency.py index f288cea1..90986701 100644 --- a/tests/test_server_idempotency.py +++ b/tests/test_server_idempotency.py @@ -407,7 +407,7 @@ def test_construction_without_pg_extra_raises_import_error(self) -> None: with patch("adcp.server.idempotency.backends._PG_AVAILABLE", False): with pytest.raises(ImportError, match="adcp\\[pg\\]"): - PgBackend(pool=MagicMock()) + PgBackend(pool=MagicMock(), lock_pool=MagicMock()) class TestScopeKeySeparatorValidation: @@ -518,6 +518,97 @@ async def test_cache_hit_replays_without_handler_call(self) -> None: # Everything else about the response is identical. assert {k: v for k, v in r2.items() if k != "replayed"} == r1 + @pytest.mark.asyncio + async def test_concurrent_same_key_executes_handler_once(self) -> None: + store = self._make_store() + entered = asyncio.Event() + release = asyncio.Event() + calls = 0 + + async def handler( + _self: object, params: dict[str, Any], context: ToolContext | None = None + ) -> dict[str, Any]: + nonlocal calls + calls += 1 + entered.set() + await release.wait() + return {"media_buy_id": "mb_only", "status": "completed"} + + wrapped = store.wrap(handler) + params = {"idempotency_key": str(uuid.uuid4()), "brand": "A"} + ctx = ToolContext(caller_identity="principal-a") + first = asyncio.create_task(wrapped(object(), params, ctx)) + await entered.wait() + second = asyncio.create_task(wrapped(object(), params, ctx)) + await asyncio.sleep(0) + assert calls == 1 + + release.set() + first_result, second_result = await asyncio.gather(first, second) + assert calls == 1 + assert first_result.get("replayed") is not True + assert second_result["replayed"] is True + + @pytest.mark.asyncio + async def test_cancelled_request_keeps_lock_until_sync_work_is_cached(self) -> None: + store = self._make_store() + entered = asyncio.Event() + release = asyncio.Event() + calls = 0 + + async def handler( + _self: object, params: dict[str, Any], context: ToolContext | None = None + ) -> dict[str, Any]: + nonlocal calls + calls += 1 + entered.set() + await release.wait() + return {"media_buy_id": "mb_only", "status": "completed"} + + wrapped = store.wrap(handler) + params = {"idempotency_key": str(uuid.uuid4()), "brand": "A"} + ctx = ToolContext(caller_identity="principal-a") + first = asyncio.create_task(wrapped(object(), params, ctx)) + await entered.wait() + first.cancel("client disconnected") + with pytest.raises(asyncio.CancelledError, match="client disconnected"): + await first + + retry = asyncio.create_task(wrapped(object(), params, ctx)) + await asyncio.sleep(0) + assert calls == 1 + release.set() + retry_result = await retry + assert calls == 1 + assert retry_result["replayed"] is True + + @pytest.mark.asyncio + async def test_legacy_backend_uses_deprecated_process_local_hold(self) -> None: + class LegacyBackend(IdempotencyBackend): + def __init__(self) -> None: + self.entry: CachedResponse | None = None + + async def get(self, scope_key: str, key: str) -> CachedResponse | None: + return self.entry + + async def put(self, scope_key: str, key: str, entry: CachedResponse) -> None: + self.entry = entry + + async def delete_expired(self, now_epoch: float | None = None) -> int: + return 0 + + store = IdempotencyStore(LegacyBackend()) + handler = _FakeHandler() + wrapped = store.wrap(_FakeHandler.create_media_buy) + params = {"idempotency_key": str(uuid.uuid4()), "brand": "A"} + ctx = ToolContext(caller_identity="principal-a") + + with pytest.warns(DeprecationWarning, match="process-local idempotency locking"): + await wrapped(handler, params, ctx) + replay = await wrapped(handler, params, ctx) + assert replay["replayed"] is True + assert handler.call_count == 1 + @pytest.mark.asyncio async def test_replay_flag_does_not_poison_cached_entry(self) -> None: """The cached ``CachedResponse.response`` MUST stay clean — the diff --git a/tests/test_verify_from_agent_url.py b/tests/test_verify_from_agent_url.py index 3d567423..e1c4eab2 100644 --- a/tests/test_verify_from_agent_url.py +++ b/tests/test_verify_from_agent_url.py @@ -290,6 +290,74 @@ async def fake_verify_starlette(request, *, options): # type: ignore[no-untyped assert seen["options"].signing_purpose == "request_signing" +@pytest.mark.asyncio +async def test_factory_uses_secure_replay_store_default_when_omitted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The convenience factory reuses one bounded cache across invocations.""" + from adcp.signing.replay import InMemoryReplayStore + + seen: list[Any] = [] + monkeypatch.setattr( + agent_resolver, + "_VERIFY_FROM_AGENT_URL_REPLAY_STORE", + InMemoryReplayStore(global_cap=10), + ) + + async def fake_resolve(*args, **kwargs): + return _resolved_with_origins(None) + + async def fake_verify_starlette(request, *, options): # type: ignore[no-untyped-def] + seen.append(options.replay_store.claim("shared-kid", "shared-nonce", 60.0)) + return "ok" + + monkeypatch.setattr(agent_resolver, "async_resolve_agent", fake_resolve) + monkeypatch.setattr("adcp.signing.middleware.verify_starlette_request", fake_verify_starlette) + + await verify_from_agent_url( + _FakeStarletteRequest(), + "https://buyer.example.com/mcp", + agent_type="sales", + operation="get_products", + ) + await verify_from_agent_url( + _FakeStarletteRequest(), + "https://buyer.example.com/mcp", + agent_type="sales", + operation="get_products", + ) + + assert seen == ["claimed", "replayed"] + + +@pytest.mark.asyncio +async def test_factory_preserves_explicit_replay_store_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Callers can still explicitly opt out for compatibility or tests.""" + seen: dict[str, Any] = {} + + async def fake_resolve(*args, **kwargs): + return _resolved_with_origins(None) + + async def fake_verify_starlette(request, *, options): # type: ignore[no-untyped-def] + seen["options"] = options + return "ok" + + monkeypatch.setattr(agent_resolver, "async_resolve_agent", fake_resolve) + monkeypatch.setattr("adcp.signing.middleware.verify_starlette_request", fake_verify_starlette) + + await verify_from_agent_url( + _FakeStarletteRequest(), + "https://buyer.example.com/mcp", + agent_type="sales", + operation="get_products", + replay_store=None, + ) + + assert seen["options"].replay_store is None + + @pytest.mark.asyncio async def test_factory_passes_signing_purpose_through( monkeypatch: pytest.MonkeyPatch, @@ -536,33 +604,26 @@ def test_static_jwks_resolver_does_not_satisfy_brand_sourced_protocol() -> None: # ---- Misconfig warnings (Argus first-pass follow-ups) ---- -def test_brand_json_source_without_expected_origins_emits_user_warning() -> None: - """A resolver advertising ``jwks_source='brand_json'`` paired with - ``expected_key_origins=None`` is an observable misconfig — the - spec's identity.key_origins consistency check (ADCP #3690 step 7) - silently no-ops. Surface as a :class:`UserWarning` so adopters - catch it in operator logs and thread the origins map through - ``VerifyOptions``.""" - import warnings as _w - +def test_brand_json_source_without_expected_origins_fails_closed() -> None: + """Missing capabilities binding rejects brand-sourced trust.""" + from adcp.signing import REQUEST_SIGNATURE_KEY_ORIGIN_MISSING from adcp.signing.agent_resolver import _BrandJsonStaticJwksResolver + from adcp.signing.errors import SignatureVerificationError from adcp.signing.verifier import _maybe_check_key_origin resolver = _BrandJsonStaticJwksResolver( {"keys": []}, jwks_uri="https://keys.brand.example/jwks.json", ) - with _w.catch_warnings(record=True) as caught: - _w.simplefilter("always") + with pytest.raises(SignatureVerificationError) as exc_info: _maybe_check_key_origin( resolver=resolver, expected_key_origins=None, signing_purpose="request_signing", posture=None, ) - user_warnings = [w for w in caught if issubclass(w.category, UserWarning)] - assert len(user_warnings) == 1 - assert "jwks_source='brand_json'" in str(user_warnings[0].message) + assert exc_info.value.code == REQUEST_SIGNATURE_KEY_ORIGIN_MISSING + assert exc_info.value.step == 7 def test_legacy_resolver_with_expected_origins_emits_deprecation_warning() -> None: