From d4bc613407722c486538cde4bc65e6da7a0297f1 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Tue, 25 Aug 2026 07:33:42 +0100 Subject: [PATCH] feat(webhooks): support tenant-scoped signing senders --- docs/handler-authoring.md | 67 +++ src/adcp/__init__.py | 12 + src/adcp/decisioning/__init__.py | 14 + src/adcp/decisioning/dispatch.py | 10 + src/adcp/decisioning/pg/__init__.py | 7 +- src/adcp/decisioning/pg/task_registry.py | 83 +++- .../decisioning/pg/task_webhook_outbox.py | 215 ++++++++- .../decisioning/pg/task_webhook_outbox.sql | 5 + src/adcp/decisioning/task_registry.py | 5 + src/adcp/decisioning/webhook_emit.py | 14 +- src/adcp/webhook_sender.py | 63 ++- src/adcp/webhooks.py | 16 +- .../test_pg_task_webhook_outbox.py | 92 +++- tests/fixtures/public_api_snapshot.json | 4 + tests/test_decisioning_dispatch.py | 57 +++ tests/test_task_webhook_outbox_pg.py | 448 +++++++++++++++++- tests/test_webhook_signing_capabilities.py | 40 +- tests/type_checks/webhook_sender_resolver.py | 34 ++ 18 files changed, 1131 insertions(+), 55 deletions(-) create mode 100644 tests/type_checks/webhook_sender_resolver.py diff --git a/docs/handler-authoring.md b/docs/handler-authoring.md index d0f8d04a2..d975f79d8 100644 --- a/docs/handler-authoring.md +++ b/docs/handler-authoring.md @@ -1446,6 +1446,73 @@ copy of the callback token is cleared in that transaction. Workers use expiring leases and exact retries; the 1–7 day horizon begins on the first attempt and must exactly match the advertised value. +Multi-tenant sellers can resolve a different signing identity for each trusted +server-side tenant scope. Use `sender_resolver=` on the outbox and pair it with +`webhook_signing_scope_resolver=` on the registry: + +```python +from adcp.decisioning import ( + PgTaskRegistry, + PgTaskWebhookOutbox, + ScopePermanentlyUnknown, + ScopeTransientlyUnavailable, + WebhookSenderResolution, +) + +class TenantWebhookSenders: + async def resolve(self, signing_scope_id: str): + credential = await internal_key_store.active_for_scope(signing_scope_id) + if credential is None: + raise ScopePermanentlyUnknown + if credential.rotation_in_progress: + raise ScopeTransientlyUnavailable + return WebhookSenderResolution( + sender=credential.webhook_sender, # cached, lifecycle-managed sender + # Read from the same trusted record used by request-scoped capabilities. + advertised_algorithms=frozenset(credential.webhook_signing_algorithms), + ) + +def trusted_signing_scope(context): + # Use only internal tenant/platform metadata populated by the server. + return context.account.metadata.webhook_signing_scope_id + +outbox = PgTaskWebhookOutbox( + pool=pool, + sender_resolver=TenantWebhookSenders(), + encryption_key=task_webhook_encryption_key, + delivery_retry_horizon_seconds=86_400, +) +registry = PgTaskRegistry( + pool=pool, + task_webhook_outbox=outbox, + webhook_signing_scope_resolver=trusted_signing_scope, +) +``` + +The scope is encrypted at task issuance, persisted on the durable outbox row, +and authenticated as envelope AAD. The worker resolves a fresh sender on every +attempt, so key rotation changes the signature without changing the stored body +or idempotency key. `ScopeTransientlyUnavailable` releases the row for retry; +`ScopePermanentlyUnknown` quarantines it for operator reconciliation. Every +resolved sender is revalidated as an RFC 9421 sender using an SDK-owned, +IP-pinned transport with private destinations disabled. Its actual key +algorithm must also appear in the trusted scope's advertised algorithm set; +a mismatch is quarantined before any request is sent. + +The resolver owns sender lifecycle. Return cached senders whose clients are +closed during application shutdown; do not allocate a new `WebhookSender` (and +therefore a new connection pool) on every delivery attempt. + +Never derive the signing scope from `push_notification_config`, the request's +buyer-supplied `context`, or an unqualified buyer account id. It must be an +opaque identifier obtained from trusted internal tenant/platform metadata. +Pass exactly one of `sender=` or `sender_resolver=`; the fixed-sender path and +existing `NULL signing_scope_id` rows remain backward compatible while that +fixed-sender mode is retained. Before switching an existing deployment to +resolver mode, drain or reconcile its pre-migration `NULL` rows: the worker +cannot safely infer a tenant key for them and will quarantine them rather than +guess a signing identity. + Production adopters may set `auto_emit_task_webhooks=False` only when an external durable outbox owns publication, retries, immutable body/key retention, and reconciliation. Set `webhook_signing_managed_externally=True` in the corresponding diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index 6f9cbee89..74e8883e3 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -724,6 +724,8 @@ def _resolve_version() -> str: "LegacyHmacFallback", "MemoryBackend", "PreparedWebhook", + "ScopePermanentlyUnknown", + "ScopeTransientlyUnavailable", "WebhookChallengeError", "WebhookChallengeResult", "WebhookDedupStore", @@ -731,6 +733,8 @@ def _resolve_version() -> str: "WebhookReceiver", "WebhookReceiverConfig", "WebhookSender", + "WebhookSenderResolution", + "WebhookSenderResolver", "WebhookVerifyOptions", "challenge_webhook_destination", "create_a2a_webhook_payload", @@ -968,6 +972,10 @@ def get_adcp_version() -> str: "WebhookReceiver", "WebhookReceiverConfig", "WebhookSender", + "WebhookSenderResolution", + "WebhookSenderResolver", + "ScopePermanentlyUnknown", + "ScopeTransientlyUnavailable", "WebhookVerifyOptions", "WebhookDedupStore", "MemoryBackend", @@ -2140,6 +2148,8 @@ def get_adcp_version() -> str: LegacyHmacFallback, MemoryBackend, PreparedWebhook, + ScopePermanentlyUnknown, + ScopeTransientlyUnavailable, WebhookChallengeError, WebhookChallengeResult, WebhookDedupStore, @@ -2147,6 +2157,8 @@ def get_adcp_version() -> str: WebhookReceiver, WebhookReceiverConfig, WebhookSender, + WebhookSenderResolution, + WebhookSenderResolver, WebhookVerifyOptions, challenge_webhook_destination, create_a2a_webhook_payload, diff --git a/src/adcp/decisioning/__init__.py b/src/adcp/decisioning/__init__.py index 1c090e1c4..1361d0926 100644 --- a/src/adcp/decisioning/__init__.py +++ b/src/adcp/decisioning/__init__.py @@ -273,6 +273,12 @@ def create_media_buy( validate_capabilities_response_shape, validate_capabilities_response_shape_async, ) +from adcp.webhook_sender import ( + ScopePermanentlyUnknown, + ScopeTransientlyUnavailable, + WebhookSenderResolution, + WebhookSenderResolver, +) # Conditional import: PgTaskRegistry needs the [pg] extra. Always expose # the name — when psycopg isn't installed we fall through to a stub class whose @@ -289,6 +295,7 @@ def create_media_buy( PgTaskRegistry, PgTaskWebhookOutbox, PostgresTaskRegistry, + WebhookSigningScopeResolver, ) except ImportError: # pragma: no cover — exercised by the [pg] extra tests from typing import ClassVar as _ClassVar @@ -341,6 +348,8 @@ def __init__(self, *args: object, **kwargs: object) -> None: "(Poetry: `poetry add 'adcp[pg]'`)." ) + from adcp.decisioning.pg.task_registry import WebhookSigningScopeResolver + __all__ = [ "Account", @@ -446,6 +455,8 @@ def __init__(self, *args: object, **kwargs: object) -> None: "SalesResult", "SalesSpecialism", "ServiceUnavailableError", + "ScopePermanentlyUnknown", + "ScopeTransientlyUnavailable", "SignalsPlatform", "SingletonAccounts", "SELF_SERVE_UPDATE_ACTION_MODES", @@ -458,6 +469,9 @@ def __init__(self, *args: object, **kwargs: object) -> None: "TaskHandoffContext", "TaskRegistry", "TaskState", + "WebhookSenderResolver", + "WebhookSenderResolution", + "WebhookSigningScopeResolver", "TranslationMap", "UNKNOWN_UPDATE_ACTION", "UnsupportedFeatureError", diff --git a/src/adcp/decisioning/dispatch.py b/src/adcp/decisioning/dispatch.py index 1abc7c3b4..075bf9fe9 100644 --- a/src/adcp/decisioning/dispatch.py +++ b/src/adcp/decisioning/dispatch.py @@ -2092,10 +2092,15 @@ async def _project_handoff( and getattr(registry, "task_webhook_outbox", None) is not None ): push_url, push_token = push_target + signing_scope_id: str | None = None + signing_scope_resolver = getattr(registry, "resolve_webhook_signing_scope", None) + if signing_scope_resolver is not None: + signing_scope_id = await signing_scope_resolver(ctx) issue_kwargs.update( webhook_url=push_url, webhook_operation_id=_extract_push_operation_id(request_params), webhook_token=push_token, + webhook_signing_scope_id=signing_scope_id, ) task_id = await registry.issue(**issue_kwargs) @@ -2364,10 +2369,15 @@ async def _project_workflow_handoff( and getattr(registry, "task_webhook_outbox", None) is not None ): push_url, push_token = push_target + signing_scope_id: str | None = None + signing_scope_resolver = getattr(registry, "resolve_webhook_signing_scope", None) + if signing_scope_resolver is not None: + signing_scope_id = await signing_scope_resolver(ctx) issue_kwargs.update( webhook_url=push_url, webhook_operation_id=_extract_push_operation_id(request_params), webhook_token=push_token, + webhook_signing_scope_id=signing_scope_id, ) task_id = await registry.issue(**issue_kwargs) handoff_ctx = TaskHandoffContext(id=task_id, _registry=registry) diff --git a/src/adcp/decisioning/pg/__init__.py b/src/adcp/decisioning/pg/__init__.py index e9f9ec9ca..14c586fde 100644 --- a/src/adcp/decisioning/pg/__init__.py +++ b/src/adcp/decisioning/pg/__init__.py @@ -36,7 +36,11 @@ PgBuyerAgentRegistry, ) from adcp.decisioning.pg.proposal_store import PgProposalStore -from adcp.decisioning.pg.task_registry import PgTaskRegistry, PostgresTaskRegistry +from adcp.decisioning.pg.task_registry import ( + PgTaskRegistry, + PostgresTaskRegistry, + WebhookSigningScopeResolver, +) from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox __all__ = [ @@ -47,4 +51,5 @@ "PgTaskRegistry", "PgTaskWebhookOutbox", "PostgresTaskRegistry", + "WebhookSigningScopeResolver", ] diff --git a/src/adcp/decisioning/pg/task_registry.py b/src/adcp/decisioning/pg/task_registry.py index 246289e43..68c200e3f 100644 --- a/src/adcp/decisioning/pg/task_registry.py +++ b/src/adcp/decisioning/pg/task_registry.py @@ -62,17 +62,20 @@ async def main(): from __future__ import annotations +import inspect import json import re import time import uuid -from typing import TYPE_CHECKING, Any, ClassVar +from collections.abc import Awaitable, Callable +from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias from adcp.decisioning.account_projection import strip_credentials_from_wire_result if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool + from adcp.decisioning.context import RequestContext from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox try: @@ -95,6 +98,8 @@ async def main(): # the only protection against SQL injection or Unicode homoglyph substitution. _SAFE_IDENTIFIER_RE = re.compile(r"^[a-z_][a-z0-9_]{0,62}$") +WebhookSigningScopeResolver: TypeAlias = Callable[["RequestContext[Any]"], str | Awaitable[str]] + class PgTaskRegistry: """PostgreSQL-backed :class:`~adcp.decisioning.TaskRegistry` — v6.1. @@ -131,6 +136,7 @@ def __init__( *, pool: AsyncConnectionPool, task_webhook_outbox: PgTaskWebhookOutbox | None = None, + webhook_signing_scope_resolver: WebhookSigningScopeResolver | None = None, _table: str = _DEFAULT_TABLE, ) -> None: if not PG_AVAILABLE: @@ -141,9 +147,18 @@ def __init__( raise ValueError( "PgTaskRegistry and PgTaskWebhookOutbox must use the same connection pool" ) + uses_sender_resolver = ( + task_webhook_outbox is not None and task_webhook_outbox._sender_resolver is not None + ) + if uses_sender_resolver != (webhook_signing_scope_resolver is not None): + raise ValueError( + "webhook_signing_scope_resolver is required exactly when the " + "PgTaskWebhookOutbox uses sender_resolver" + ) self._pool = pool self._table = _table self.task_webhook_outbox = task_webhook_outbox + self._webhook_signing_scope_resolver = webhook_signing_scope_resolver self.atomic_task_webhook_outbox = task_webhook_outbox is not None # Pre-format queries at construction so the hot path avoids f-strings per call. @@ -247,6 +262,7 @@ async def issue( webhook_url: str | None = None, webhook_operation_id: str | None = None, webhook_token: str | None = None, + webhook_signing_scope_id: str | None = None, **_extra: Any, ) -> str: """Allocate a task_id, persist a ``submitted`` row, return the id. @@ -269,6 +285,8 @@ async def issue( raise ValueError("webhook_url must be non-empty when supplied") if webhook_operation_id is not None and not webhook_operation_id: raise ValueError("webhook_operation_id must be non-empty when supplied") + if webhook_url is None and webhook_signing_scope_id is not None: + raise ValueError("webhook_signing_scope_id requires webhook_url") outbox = self.task_webhook_outbox if webhook_url is not None: if outbox is None: @@ -287,6 +305,7 @@ async def issue( url=webhook_url, operation_id=webhook_operation_id, token=webhook_token, + signing_scope_id=webhook_signing_scope_id, ) now = time.time() async with self._pool.connection() as conn: @@ -305,6 +324,46 @@ async def issue( ) return task_id + async def resolve_webhook_signing_scope( + self, + context: RequestContext[Any], + ) -> str | None: + """Derive an opaque signing scope from trusted framework context. + + The callback is operator wiring and receives the hydrated + :class:`RequestContext`. It must use internal tenant/platform metadata, + never buyer request ``context``, ``push_notification_config``, or an + unqualified buyer account id. + """ + resolver = self._webhook_signing_scope_resolver + if resolver is None: + return None + from adcp.decisioning.types import AdcpError + + try: + value: object = resolver(context) + if inspect.isawaitable(value): + value = await value + except Exception: + raise AdcpError( + "INTERNAL_ERROR", + message="Webhook signing scope resolution failed", + recovery="terminal", + ) from None + if not isinstance(value, str): + raise AdcpError( + "INTERNAL_ERROR", + message="Webhook signing scope resolver returned an invalid value", + recovery="terminal", + ) + # Reuse the outbox's bounded opaque-ID validation before any task row + # is issued. This is trusted server state, but it still crosses a DB + # and authenticated-envelope boundary. + if self.task_webhook_outbox is None: + raise RuntimeError("signing scope resolver requires a task webhook outbox") + self.task_webhook_outbox._validate_signing_scope_id(value) + return value + async def update_progress( self, task_id: str, @@ -475,12 +534,14 @@ async def _enqueue_terminal_if_registered( ) if registration_nonce is None: raise RuntimeError(f"Task {task_id!r} has incomplete webhook registration") - url, operation_id, token = self.task_webhook_outbox.open_registration( - account_id=account_id, - task_id=task_id, - task_type=task_type, - encrypted_registration=bytes(encrypted_registration), - nonce=bytes(registration_nonce), + url, operation_id, token, signing_scope_id = ( + self.task_webhook_outbox._open_registration_with_scope( + account_id=account_id, + task_id=task_id, + task_type=task_type, + encrypted_registration=bytes(encrypted_registration), + nonce=bytes(registration_nonce), + ) ) await self.task_webhook_outbox.enqueue_terminal( conn, @@ -492,6 +553,7 @@ async def _enqueue_terminal_if_registered( url=url, operation_id=operation_id, token=token, + signing_scope_id=signing_scope_id, ) # The encrypted outbox envelope now owns the callback registration. # Clear the task-row copy in this same transaction. @@ -515,4 +577,9 @@ async def discard(self, task_id: str) -> None: PostgresTaskRegistry = PgTaskRegistry -__all__ = ["PG_AVAILABLE", "PgTaskRegistry", "PostgresTaskRegistry"] +__all__ = [ + "PG_AVAILABLE", + "PgTaskRegistry", + "PostgresTaskRegistry", + "WebhookSigningScopeResolver", +] diff --git a/src/adcp/decisioning/pg/task_webhook_outbox.py b/src/adcp/decisioning/pg/task_webhook_outbox.py index 421168f79..b4d765309 100644 --- a/src/adcp/decisioning/pg/task_webhook_outbox.py +++ b/src/adcp/decisioning/pg/task_webhook_outbox.py @@ -25,14 +25,20 @@ from cryptography.hazmat.primitives.ciphers.aead import AESGCM from adcp.signing.jwks import SSRFValidationError -from adcp.webhook_sender import PreparedWebhook, WebhookDeliveryResult +from adcp.webhook_sender import ( + PreparedWebhook, + ScopePermanentlyUnknown, + ScopeTransientlyUnavailable, + WebhookDeliveryResult, + WebhookSender, + WebhookSenderResolution, + WebhookSenderResolver, +) from adcp.webhook_supervisor import RetryPolicy if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool - from adcp.webhook_sender import WebhookSender - try: import psycopg_pool @@ -50,6 +56,7 @@ DEFAULT_TABLE = "adcp_task_webhook_outbox" MIN_RETRY_HORIZON_SECONDS = 86_400 MAX_RETRY_HORIZON_SECONDS = 604_800 +MAX_SIGNING_SCOPE_ID_BYTES = 255 class PgTaskWebhookOutbox: @@ -68,7 +75,8 @@ def __init__( self, *, pool: AsyncConnectionPool, - sender: WebhookSender | None, + sender: WebhookSender | None = None, + sender_resolver: WebhookSenderResolver | None = None, encryption_key: bytes, delivery_retry_horizon_seconds: int, retry: RetryPolicy | None = None, @@ -77,17 +85,12 @@ def __init__( ) -> None: if not PG_AVAILABLE: raise ImportError(_INSTALL_HINT) - if sender is None: - raise ValueError("PgTaskWebhookOutbox requires a non-None WebhookSender") + if (sender is None) == (sender_resolver is None): + raise ValueError("pass exactly one of sender or sender_resolver") if len(encryption_key) != 32: raise ValueError("encryption_key must be exactly 32 bytes for AES-256-GCM") - if not getattr(sender, "_owns_client", False) or getattr( - sender, "_allow_private_destinations", False - ): - raise ValueError( - "PgTaskWebhookOutbox requires a WebhookSender using the SDK-owned " - "IP-pinned transport with private destinations disabled" - ) + if sender is not None: + self._validate_delivery_sender(sender) if type(delivery_retry_horizon_seconds) is not int or not ( MIN_RETRY_HORIZON_SECONDS <= delivery_retry_horizon_seconds <= MAX_RETRY_HORIZON_SECONDS ): @@ -97,8 +100,8 @@ def __init__( ) if type(lease_seconds) is not int or lease_seconds <= 0: raise ValueError("lease_seconds must be a positive integer") - sender_timeout = float(getattr(sender, "_timeout", 0.0)) - if lease_seconds < sender_timeout + 5: + sender_timeout = float(getattr(sender, "_timeout", 0.0)) if sender is not None else 0.0 + if sender is not None and lease_seconds < sender_timeout + 5: raise ValueError( "lease_seconds must exceed the sender HTTP timeout by at least 5 seconds" ) @@ -119,6 +122,7 @@ def __init__( self._pool = pool self._sender = sender + self._sender_resolver = sender_resolver self._cipher = AESGCM(encryption_key) self.delivery_retry_horizon_seconds = delivery_retry_horizon_seconds self._retry = resolved_retry @@ -129,9 +133,9 @@ def __init__( self._sql_insert = ( # noqa: S608 f"INSERT INTO {table} (" "task_id, task_type, terminal_status, url, operation_id, " - "idempotency_key, account_id, encrypted_body, envelope_nonce, " + "idempotency_key, signing_scope_id, account_id, encrypted_body, envelope_nonce, " "retry_horizon_seconds" - ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id" + ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id" ) self._sql_expire = ( # noqa: S608 f"WITH expired AS (SELECT id FROM {table}" @@ -159,7 +163,8 @@ def __init__( " FROM candidate WHERE outbox.id = candidate.id" " RETURNING outbox.id, outbox.account_id, outbox.task_id, outbox.task_type," " outbox.terminal_status, outbox.url, outbox.operation_id," - " outbox.idempotency_key, outbox.encrypted_body, outbox.envelope_nonce," + " outbox.idempotency_key, outbox.signing_scope_id," + " outbox.encrypted_body, outbox.envelope_nonce," " outbox.attempt_count" ) self._sql_ack = ( # noqa: S608 @@ -202,6 +207,7 @@ async def create_schema(self) -> None: url TEXT NOT NULL, operation_id TEXT NOT NULL, idempotency_key TEXT COLLATE "C" NOT NULL UNIQUE, + signing_scope_id TEXT COLLATE "C", encrypted_body BYTEA NOT NULL, envelope_nonce BYTEA NOT NULL, state TEXT NOT NULL DEFAULT 'pending', @@ -225,6 +231,8 @@ async def create_schema(self) -> None: CHECK ((first_attempt_at IS NULL) = (retry_until IS NULL)), CHECK (retry_until IS NULL OR retry_until > first_attempt_at) )""", + f'''ALTER TABLE {self._table} + ADD COLUMN IF NOT EXISTS signing_scope_id TEXT COLLATE "C"''', f"""CREATE INDEX IF NOT EXISTS {self._table}_work_idx ON {self._table} (available_at, id) WHERE state IN ('pending', 'in_flight')""", @@ -247,11 +255,14 @@ async def enqueue_terminal( url: str, operation_id: str, token: str | None, + signing_scope_id: str | None = None, ) -> int: """Insert a terminal webhook using the caller's open transaction.""" if status not in {"completed", "failed"}: raise ValueError(f"terminal webhook status must be completed or failed, got {status!r}") - prepared = self._sender.prepare_mcp( + self._validate_scope_for_mode(signing_scope_id, require_resolver_scope=False) + preparer = self._sender or WebhookSender + prepared = preparer.prepare_mcp( url=url, task_id=task_id, task_type=task_type, @@ -270,6 +281,7 @@ async def enqueue_terminal( url=prepared.url, operation_id=operation_id, idempotency_key=prepared.idempotency_key, + signing_scope_id=signing_scope_id, ) encrypted_body = self._cipher.encrypt(nonce, prepared.body, aad) cursor = await conn.execute( @@ -281,6 +293,7 @@ async def enqueue_terminal( prepared.url, operation_id, prepared.idempotency_key, + signing_scope_id, account_id, encrypted_body, nonce, @@ -305,12 +318,19 @@ def protect_registration( url: str, operation_id: str, token: str | None, + signing_scope_id: str | None = None, ) -> tuple[bytes, bytes]: """Encrypt and authenticate callback registration at task issue time.""" self._validate_callback_url(url) + self._validate_scope_for_mode(signing_scope_id, require_resolver_scope=True) nonce = os.urandom(12) plaintext = json.dumps( - {"url": url, "operation_id": operation_id, "token": token}, + { + "url": url, + "operation_id": operation_id, + "token": token, + "signing_scope_id": signing_scope_id, + }, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") @@ -336,7 +356,30 @@ def open_registration( encrypted_registration: bytes, nonce: bytes, ) -> tuple[str, str, str | None]: - """Verify and decrypt callback registration at terminal transition.""" + """Verify and decrypt a callback registration. + + The three-item return shape is retained for compatibility. Durable + registry dispatch uses the private scope-aware decoder below. + """ + url, operation_id, token, _signing_scope_id = self._open_registration_with_scope( + account_id=account_id, + task_id=task_id, + task_type=task_type, + encrypted_registration=encrypted_registration, + nonce=nonce, + ) + return url, operation_id, token + + def _open_registration_with_scope( + self, + *, + account_id: str, + task_id: str, + task_type: str, + encrypted_registration: bytes, + nonce: bytes, + ) -> tuple[str, str, str | None, str | None]: + """Verify and decrypt callback registration with its trusted scope.""" try: plaintext = self._cipher.decrypt( nonce, @@ -355,12 +398,16 @@ def open_registration( url = value.get("url") operation_id = value.get("operation_id") token = value.get("token") + signing_scope_id = value.get("signing_scope_id") if not isinstance(url, str) or not isinstance(operation_id, str): raise ValueError("task webhook registration has invalid URL or operation_id") if token is not None and not isinstance(token, str): raise ValueError("task webhook registration token must be a string or null") + if signing_scope_id is not None and not isinstance(signing_scope_id, str): + raise ValueError("task webhook registration signing scope must be a string or null") self._validate_callback_url(url) - return url, operation_id, token + self._validate_scope_for_mode(signing_scope_id, require_resolver_scope=False) + return url, operation_id, token, signing_scope_id async def run_worker( self, @@ -416,6 +463,7 @@ async def process_one(self) -> bool: url, operation_id, idempotency_key, + signing_scope_id, encrypted_body, nonce, attempt_count, @@ -428,6 +476,7 @@ async def process_one(self) -> bool: url=str(url), operation_id=str(operation_id), idempotency_key=str(idempotency_key), + signing_scope_id=(str(signing_scope_id) if signing_scope_id is not None else None), ) try: body_bytes = self._cipher.decrypt(bytes(nonce), bytes(encrypted_body), aad) @@ -463,9 +512,22 @@ async def process_one(self) -> bool: error: BaseException | None = None try: delivery = await asyncio.wait_for( - self._sender.send_prepared(prepared), + self._deliver_prepared( + prepared, + str(signing_scope_id) if signing_scope_id is not None else None, + ), timeout=self._lease_seconds - 1, ) + except ScopePermanentlyUnknown: + await self._quarantine_permanent_delivery_error( + row_id=row_id, + lease_token=lease_token, + task_id=str(task_id), + error=ValueError("webhook signing scope is permanently unavailable"), + ) + return True + except ScopeTransientlyUnavailable: + error = RuntimeError("webhook signing scope is temporarily unavailable") except SSRFValidationError as exc: if exc.transient: error = exc @@ -506,7 +568,7 @@ async def process_one(self) -> bool: (error_message[:1000], row_id, lease_token), ) logger.error( - "[adcp.task_webhook_outbox] permanent HTTP failure for task %s; " "row quarantined", + "[adcp.task_webhook_outbox] permanent HTTP failure for task %s; row quarantined", task_id, ) return True @@ -546,7 +608,7 @@ async def _quarantine_permanent_delivery_error( (error_message[:1000], row_id, lease_token), ) logger.error( - "[adcp.task_webhook_outbox] permanent delivery failure for task %s; " "row quarantined", + "[adcp.task_webhook_outbox] permanent delivery failure for task %s; row quarantined", task_id, ) @@ -572,6 +634,92 @@ def _retry_delay(self, attempt_count: int) -> float: delay *= 0.5 + random.random() * 0.5 return delay + @staticmethod + def _validate_delivery_sender(sender: WebhookSender) -> None: + if not callable(getattr(sender, "send_prepared", None)): + raise ValueError("webhook sender resolver must return a WebhookSender") + if not getattr(sender, "_owns_client", False) or getattr( + sender, "_allow_private_destinations", False + ): + raise ValueError( + "PgTaskWebhookOutbox requires a WebhookSender using the SDK-owned " + "IP-pinned transport with private destinations disabled" + ) + if getattr(sender, "signs_with_rfc9421", False) is not True: + raise ValueError("PgTaskWebhookOutbox requires an RFC 9421 signing sender") + + @staticmethod + def _validate_signing_scope_id(signing_scope_id: str | None) -> None: + if signing_scope_id is None: + return + if not signing_scope_id or not signing_scope_id.isprintable(): + raise ValueError("signing_scope_id must be a non-empty printable string") + if len(signing_scope_id.encode("utf-8")) > MAX_SIGNING_SCOPE_ID_BYTES: + raise ValueError( + f"signing_scope_id must not exceed {MAX_SIGNING_SCOPE_ID_BYTES} UTF-8 bytes" + ) + + def _validate_scope_for_mode( + self, + signing_scope_id: str | None, + *, + require_resolver_scope: bool, + ) -> None: + self._validate_signing_scope_id(signing_scope_id) + if self._sender is not None and signing_scope_id is not None: + raise ValueError("fixed-sender outboxes must not carry a signing_scope_id") + if ( + require_resolver_scope + and self._sender_resolver is not None + and signing_scope_id is None + ): + raise ValueError("sender-resolver outboxes require a signing_scope_id") + + async def _resolve_delivery_sender(self, signing_scope_id: str | None) -> WebhookSender: + """Resolve and revalidate the sender at every delivery attempt.""" + self._validate_signing_scope_id(signing_scope_id) + if self._sender is not None: + if signing_scope_id is not None: + raise ScopePermanentlyUnknown + return self._sender + + resolver = self._sender_resolver + if resolver is None or signing_scope_id is None: + raise ScopePermanentlyUnknown + try: + resolution = await resolver.resolve(signing_scope_id) + except (ScopePermanentlyUnknown, ScopeTransientlyUnavailable): + raise + except Exception: + # Resolver diagnostics may contain key-service details. Keep the + # durable row and logs on a bounded local discriminator only. + raise ScopeTransientlyUnavailable from None + try: + if not isinstance(resolution, WebhookSenderResolution): + raise ValueError("resolver returned an invalid sender resolution") + sender = resolution.sender + self._validate_delivery_sender(sender) + sender_algorithm = getattr(getattr(sender, "_auth", None), "alg", None) + if sender_algorithm not in resolution.advertised_algorithms: + raise ValueError("resolved sender algorithm was not advertised for its scope") + sender_timeout = float(getattr(sender, "_timeout", 0.0)) + if self._lease_seconds < sender_timeout + 5: + raise ValueError("resolved sender timeout exceeds the outbox lease budget") + except Exception: + # Treat malformed/adversarial resolver output as a permanent local + # configuration error without persisting its diagnostics. + raise ScopePermanentlyUnknown from None + return sender + + async def _deliver_prepared( + self, + prepared: PreparedWebhook, + signing_scope_id: str | None, + ) -> WebhookDeliveryResult: + """Resolve, validate, and send within the caller's single lease budget.""" + sender = await self._resolve_delivery_sender(signing_scope_id) + return await sender.send_prepared(prepared) + @staticmethod def _envelope_aad( *, @@ -582,10 +730,25 @@ def _envelope_aad( url: str, operation_id: str, idempotency_key: str, + signing_scope_id: str | None = None, ) -> bytes: """Canonical associated data binding every routing/security field.""" + fields: list[str | None] = [ + account_id, + task_id, + task_type, + status, + url, + operation_id, + idempotency_key, + ] + # Preserve the exact seven-field AAD for rows written before the + # signing-scope migration. Scoped rows append the trusted scope and + # therefore fail authenticated decryption if the DB column is swapped. + if signing_scope_id is not None: + fields.append(signing_scope_id) return json.dumps( - [account_id, task_id, task_type, status, url, operation_id, idempotency_key], + fields, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") diff --git a/src/adcp/decisioning/pg/task_webhook_outbox.sql b/src/adcp/decisioning/pg/task_webhook_outbox.sql index 511ff6c85..4ae57e7e9 100644 --- a/src/adcp/decisioning/pg/task_webhook_outbox.sql +++ b/src/adcp/decisioning/pg/task_webhook_outbox.sql @@ -15,6 +15,8 @@ CREATE TABLE IF NOT EXISTS adcp_task_webhook_outbox ( url TEXT NOT NULL, operation_id TEXT NOT NULL, idempotency_key TEXT COLLATE "C" NOT NULL UNIQUE, + -- Trusted server-side tenant/key scope. NULL preserves fixed-sender rows. + signing_scope_id TEXT COLLATE "C", encrypted_body BYTEA NOT NULL, envelope_nonce BYTEA NOT NULL, state TEXT NOT NULL DEFAULT 'pending', @@ -39,6 +41,9 @@ CREATE TABLE IF NOT EXISTS adcp_task_webhook_outbox ( CHECK (retry_until IS NULL OR retry_until > first_attempt_at) ); +ALTER TABLE adcp_task_webhook_outbox + ADD COLUMN IF NOT EXISTS signing_scope_id TEXT COLLATE "C"; + CREATE INDEX IF NOT EXISTS adcp_task_webhook_outbox_work_idx ON adcp_task_webhook_outbox (available_at, id) WHERE state IN ('pending', 'in_flight'); diff --git a/src/adcp/decisioning/task_registry.py b/src/adcp/decisioning/task_registry.py index ca3a16f26..6ff677ffc 100644 --- a/src/adcp/decisioning/task_registry.py +++ b/src/adcp/decisioning/task_registry.py @@ -219,6 +219,7 @@ async def issue( webhook_url: str | None = None, webhook_operation_id: str | None = None, webhook_token: str | None = None, + webhook_signing_scope_id: str | None = None, **_extra: Any, ) -> str: """Allocate a fresh task_id, persist a ``submitted`` row, and @@ -248,6 +249,10 @@ async def issue( echo verbatim in every task webhook. Required with ``webhook_url``. :param webhook_token: Optional buyer validation token to echo in the webhook payload. Treat as sensitive callback registration data. + :param webhook_signing_scope_id: Framework-derived opaque scope used + by an SDK-managed tenant-aware webhook outbox. This value must + originate from trusted server-side ``RequestContext`` metadata, + never buyer request fields. Custom registries may ignore it. :param _extra: Forward-compat slot for kwargs added by future framework versions. Custom registry impls MUST include ``**_extra: Any`` on their ``issue()`` signature so the diff --git a/src/adcp/decisioning/webhook_emit.py b/src/adcp/decisioning/webhook_emit.py index 5856c5b0a..db18a8a0f 100644 --- a/src/adcp/decisioning/webhook_emit.py +++ b/src/adcp/decisioning/webhook_emit.py @@ -541,10 +541,16 @@ def validate_webhook_signing_for_capabilities( }, ) + outbox_sender_resolver = ( + getattr(task_outbox, "_sender_resolver", None) if internal_outbox_ready else None + ) resolved_sender: Any = ( getattr(task_outbox, "_sender", None) if internal_outbox_ready else sender ) - sender_introspectable = True + # Tenant-aware outboxes resolve and validate the active RFC 9421 sender + # on every attempt. There is intentionally no single boot-time key or + # algorithm to introspect because rotation is part of the contract. + sender_introspectable = outbox_sender_resolver is None if resolved_sender is None and supervisor is not None and not internal_outbox_ready: # Both reference supervisors store the underlying WebhookSender # on ``_sender``. Custom Protocol-only impls (Celery/Kafka @@ -706,6 +712,7 @@ def task_webhook_owner_ready( outbox = getattr(registry, "task_webhook_outbox", None) outbox_horizon = getattr(outbox, "delivery_retry_horizon_seconds", None) outbox_sender = getattr(outbox, "_sender", None) + outbox_sender_resolver = getattr(outbox, "_sender_resolver", None) return ( auto_emit_task_webhooks is True and sender is None @@ -718,7 +725,10 @@ def task_webhook_owner_ready( and getattr(outbox, "delivery_state_is_durable", False) is True and type(outbox_horizon) is int and outbox_horizon == advertised_horizon - and getattr(outbox_sender, "signs_with_rfc9421", False) is True + and ( + getattr(outbox_sender, "signs_with_rfc9421", False) is True + or outbox_sender_resolver is not None + ) ) diff --git a/src/adcp/webhook_sender.py b/src/adcp/webhook_sender.py index 47f1ad180..3cae3bb08 100644 --- a/src/adcp/webhook_sender.py +++ b/src/adcp/webhook_sender.py @@ -34,7 +34,7 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import Any, Protocol, runtime_checkable import httpx from cryptography.hazmat.primitives.asymmetric import ec, ed25519 @@ -227,6 +227,59 @@ class PreparedWebhook: extra_headers: Mapping[str, str] = field(default_factory=dict) +class ScopePermanentlyUnknown(RuntimeError): # noqa: N818 - public issue contract + """A signing scope can no longer resolve to a webhook sender. + + Durable outbox workers quarantine the affected row: retries cannot repair + a decommissioned or invalid scope without operator intervention. + """ + + +class ScopeTransientlyUnavailable(RuntimeError): # noqa: N818 - public issue contract + """A signing scope is temporarily unable to resolve a sender. + + Durable outbox workers release the row for retry, allowing credential + rotation or a temporarily unavailable key service to recover. + """ + + +@dataclass(frozen=True, slots=True) +class WebhookSenderResolution: + """A sender bound to the algorithms advertised for its trusted scope. + + Tenant resolvers must build this from the same internal credential record + used to produce request-scoped ``webhook_signing.algorithms``. The durable + outbox rejects a resolved key whose actual algorithm is outside this set. + """ + + sender: WebhookSender + advertised_algorithms: frozenset[str] + + def __post_init__(self) -> None: + algorithms = frozenset(self.advertised_algorithms) + if not algorithms or not algorithms.issubset({ALG_ED25519, ALG_ES256}): + raise ValueError( + "advertised_algorithms must be a non-empty set of supported " + "webhook-signing algorithms" + ) + object.__setattr__(self, "advertised_algorithms", algorithms) + + +@runtime_checkable +class WebhookSenderResolver(Protocol): + """Resolve the current sender for an opaque, trusted signing scope. + + Implementations typically load a tenant's active signing credential from + an internal key registry. They must not derive the scope from buyer input. + The outbox calls :meth:`resolve` for every delivery attempt so key rotation + can take effect without changing the immutable webhook body or delivery + idempotency key. + """ + + async def resolve(self, signing_scope_id: str) -> WebhookSenderResolution: + """Return the current sender and its scope's advertised algorithms.""" + + class WebhookSender: """Outbound signed-webhook delivery client. @@ -572,7 +625,7 @@ def __repr__(self) -> str: # Explicit repr so no future debug helper or error traceback auto- # renders self.__dict__ and pulls the private key (or HMAC secret / # bearer token) into logs. - return f"WebhookSender(auth={type(self._auth).__name__}, " f"key_id={self._key_id!r})" + return f"WebhookSender(auth={type(self._auth).__name__}, key_id={self._key_id!r})" @property def signs_with_rfc9421(self) -> bool: @@ -658,8 +711,8 @@ async def send_mcp( ) ) + @staticmethod def prepare_mcp( - self, *, url: str, task_id: str, @@ -1189,7 +1242,11 @@ async def _send_bytes( __all__ = [ "DockerLocalhostRewrite", "PreparedWebhook", + "ScopePermanentlyUnknown", + "ScopeTransientlyUnavailable", "TransportHook", "WebhookDeliveryResult", "WebhookSender", + "WebhookSenderResolution", + "WebhookSenderResolver", ] diff --git a/src/adcp/webhooks.py b/src/adcp/webhooks.py index 35c8bd24a..bc932d881 100644 --- a/src/adcp/webhooks.py +++ b/src/adcp/webhooks.py @@ -1226,9 +1226,7 @@ async def _send_legacy_webhook_challenge( ) -> httpx.Response: schemes_raw = authentication.get("schemes") if schemes_raw is not None and not isinstance(schemes_raw, (list, tuple)): - raise ValueError( - "authentication.schemes must be a list, got " f"{type(schemes_raw).__name__}" - ) + raise ValueError(f"authentication.schemes must be a list, got {type(schemes_raw).__name__}") schemes = list(schemes_raw or []) if len(schemes) != 1: raise ValueError("authentication.schemes must contain exactly one scheme") @@ -1412,7 +1410,7 @@ async def challenge_webhook_destination( field=field, url=error_url, suggestion=( - "Pass the seller's WebhookSender, or pass config.authentication " "for legacy auth." + "Pass the seller's WebhookSender, or pass config.authentication for legacy auth." ), ) try: @@ -1810,7 +1808,7 @@ def _extract_config_fields( schemes_raw = auth.get("schemes") if schemes_raw is not None and not isinstance(schemes_raw, (list, tuple)): raise ValueError( - "config.authentication.schemes must be a list, got " f"{type(schemes_raw).__name__}" + f"config.authentication.schemes must be a list, got {type(schemes_raw).__name__}" ) schemes = list(schemes_raw or []) if len(schemes) > 1: @@ -2006,8 +2004,12 @@ def _validate_header_value(name: str, value: Any) -> None: # such cycles without a third helper module. from adcp.webhook_sender import ( # noqa: E402 PreparedWebhook, + ScopePermanentlyUnknown, + ScopeTransientlyUnavailable, WebhookDeliveryResult, WebhookSender, + WebhookSenderResolution, + WebhookSenderResolver, ) from adcp.webhook_supervisor_pg import ( # noqa: E402 PgWebhookDeliverySupervisor, @@ -2032,8 +2034,12 @@ def _validate_header_value(name: str, value: Any) -> None: # Sender — one-call outbound helpers "deliver", "PreparedWebhook", + "ScopePermanentlyUnknown", + "ScopeTransientlyUnavailable", "WebhookDeliveryResult", "WebhookSender", + "WebhookSenderResolution", + "WebhookSenderResolver", "WebhookDestinationPolicy", "WebhookDestinationValidation", "WebhookDestinationValidationError", diff --git a/tests/conformance/decisioning/test_pg_task_webhook_outbox.py b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py index 0f1ba5c84..197b58b37 100644 --- a/tests/conformance/decisioning/test_pg_task_webhook_outbox.py +++ b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py @@ -28,7 +28,11 @@ ) from adcp.decisioning.pg import PgTaskRegistry, PgTaskWebhookOutbox # noqa: E402 -from adcp.webhook_sender import PreparedWebhook, WebhookDeliveryResult # noqa: E402 +from adcp.webhook_sender import ( # noqa: E402 + PreparedWebhook, + WebhookDeliveryResult, + WebhookSenderResolution, +) def _sender() -> MagicMock: @@ -37,6 +41,7 @@ def _sender() -> MagicMock: sender._allow_private_destinations = False sender._timeout = 10.0 sender.signs_with_rfc9421 = True + sender._auth.alg = "ed25519" def prepare_mcp(**kwargs: Any) -> PreparedWebhook: key = f"whk_{uuid.uuid4().hex}" @@ -124,7 +129,7 @@ async def test_terminal_state_and_encrypted_outbox_commit_together(stack) -> Non ).fetchone() outbox_row = await ( await conn.execute( - f"SELECT encrypted_body, first_attempt_at, retry_until " # noqa: S608 + f"SELECT encrypted_body, signing_scope_id, first_attempt_at, retry_until " # noqa: S608 f"FROM {outbox._table} WHERE task_id = %s", (task_id,), ) @@ -132,7 +137,7 @@ async def test_terminal_state_and_encrypted_outbox_commit_together(stack) -> Non assert task_row == ("completed", None, None) assert outbox_row is not None assert b"buyer-secret" not in bytes(outbox_row[0]) - assert outbox_row[1:] == (None, None) + assert outbox_row[1:] == (None, None, None) assert await outbox.process_one() is True async with pool.connection() as conn: @@ -149,6 +154,87 @@ async def test_terminal_state_and_encrypted_outbox_commit_together(stack) -> Non assert delivered[2] is not None +@pytest.mark.asyncio +async def test_tenant_scopes_survive_restart_and_resolve_current_sender() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + outbox_table = f"test_task_outbox_{suffix}" + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=2, + max_size=8, + open=False, + ) as pool: + await pool.open() + tenant_a_old = _sender() + tenant_a_current = _sender() + tenant_b = _sender() + active = {"scope-a": tenant_a_old, "scope-b": tenant_b} + resolver = MagicMock( + resolve=AsyncMock( + side_effect=lambda scope: WebhookSenderResolution( + sender=active[scope], + advertised_algorithms=frozenset({"ed25519"}), + ) + ) + ) + outbox = PgTaskWebhookOutbox( + pool=pool, + sender_resolver=resolver, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + table=outbox_table, + ) + registry = PgTaskRegistry( + pool=pool, + task_webhook_outbox=outbox, + webhook_signing_scope_resolver=lambda context: str(context.tenant_id), + _table=task_table, + ) + await registry.create_schema() + await outbox.create_schema() + try: + task_a = await registry.issue( + account_id="buyer-a", + task_type="create_media_buy", + webhook_url="https://buyer.example/a", + webhook_operation_id="op-a", + webhook_signing_scope_id="scope-a", + ) + task_b = await registry.issue( + account_id="buyer-b", + task_type="create_media_buy", + webhook_url="https://buyer.example/b", + webhook_operation_id="op-b", + webhook_signing_scope_id="scope-b", + ) + await registry.complete(task_a, {"media_buy_id": "mb-a"}) + await registry.complete(task_b, {"media_buy_id": "mb-b"}) + + async with pool.connection() as conn: + rows = await ( + await conn.execute( + f"SELECT task_id, signing_scope_id FROM {outbox_table} " # noqa: S608 + "ORDER BY task_id" + ) + ).fetchall() + assert sorted(rows) == sorted([(task_a, "scope-a"), (task_b, "scope-b")]) + + # Rotate after enqueue. The body/key remain stored, while delivery + # resolves the current tenant sender on the next worker attempt. + active["scope-a"] = tenant_a_current + assert await outbox.process_one() is True + assert await outbox.process_one() is True + + tenant_a_old.send_prepared.assert_not_awaited() + tenant_a_current.send_prepared.assert_awaited_once() + tenant_b.send_prepared.assert_awaited_once() + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {outbox_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + @pytest.mark.asyncio async def test_concurrent_claims_deliver_one_attempt(stack) -> None: _pool, registry, outbox, sender = stack diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index ab2841562..51e29a1a7 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -411,6 +411,8 @@ "ResolvedProperty", "ResponsePayloadJwsEnvelope", "SchemaValidationError", + "ScopePermanentlyUnknown", + "ScopeTransientlyUnavailable", "SegmentIdActivationKey", "SellerAgentReference", "SiSendActionResponseRequest", @@ -544,6 +546,8 @@ "WebhookReceiver", "WebhookReceiverConfig", "WebhookSender", + "WebhookSenderResolution", + "WebhookSenderResolver", "WebhookVerifyOptions", "WholesaleFeedEvent", "WholesaleFeedWebhook", diff --git a/tests/test_decisioning_dispatch.py b/tests/test_decisioning_dispatch.py index 916883da8..d184569f7 100644 --- a/tests/test_decisioning_dispatch.py +++ b/tests/test_decisioning_dispatch.py @@ -1098,6 +1098,63 @@ async def _handoff_fn(task_ctx): assert rec["result"] == {"media_buy_id": "mb_1"} +@pytest.mark.asyncio +async def test_handoff_uses_trusted_context_scope_not_buyer_fields( + executor: ThreadPoolExecutor, +) -> None: + class _PushConfig(BaseModel): + url: str + operation_id: str + token: str | None = None + + class _PushRequest(BaseModel): + push_notification_config: _PushConfig + context: dict[str, Any] + + class _ScopedRegistry(InMemoryTaskRegistry): + task_webhook_outbox = object() + + def __init__(self) -> None: + super().__init__() + self.issue_kwargs: dict[str, Any] = {} + self.seen_context = None + + async def resolve_webhook_signing_scope(self, context): + self.seen_context = context + return f"trusted:{context.tenant_id}" + + async def issue(self, **kwargs): + self.issue_kwargs = dict(kwargs) + return await super().issue(**kwargs) + + registry = _ScopedRegistry() + ctx = _build_request_context( + ToolContext(tenant_id="seller-tenant-a"), Account(id="buyer-account"), None + ) + request = _PushRequest( + push_notification_config=_PushConfig( + url="https://buyer.example/webhook", + operation_id="op-1", + ), + context={"signing_scope_id": "buyer-selected-scope"}, + ) + + await _project_handoff( + TaskHandoff(lambda _task_ctx: {"ok": True}), + ctx, + method_name="create_media_buy", + registry=registry, + executor=executor, + request_params=request, + webhook_auto_emit=False, + webhook_external_owner_ready=True, + ) + + assert registry.seen_context is ctx + assert registry.issue_kwargs["webhook_signing_scope_id"] == "trusted:seller-tenant-a" + assert registry.issue_kwargs["webhook_signing_scope_id"] != request.context["signing_scope_id"] + + @pytest.mark.asyncio async def test_handoff_async_fn_completes_via_registry( executor: ThreadPoolExecutor, diff --git a/tests/test_task_webhook_outbox_pg.py b/tests/test_task_webhook_outbox_pg.py index fc4f153d9..4ce7437c3 100644 --- a/tests/test_task_webhook_outbox_pg.py +++ b/tests/test_task_webhook_outbox_pg.py @@ -9,7 +9,13 @@ import pytest -from adcp.webhook_sender import PreparedWebhook, WebhookDeliveryResult +from adcp.webhook_sender import ( + PreparedWebhook, + ScopePermanentlyUnknown, + ScopeTransientlyUnavailable, + WebhookDeliveryResult, + WebhookSenderResolution, +) def _cursor(value: Any = None) -> AsyncMock: @@ -46,6 +52,7 @@ def _sender() -> MagicMock: sender._owns_client = True sender._allow_private_destinations = False sender._timeout = 10.0 + sender._auth.alg = "ed25519" body = json.dumps( { "idempotency_key": "whk_1234567890123456", @@ -75,6 +82,24 @@ def _sender() -> MagicMock: return sender +def _resolution( + sender: Any, + algorithms: frozenset[str] = frozenset({"ed25519"}), +) -> WebhookSenderResolution: + return WebhookSenderResolution(sender=sender, advertised_algorithms=algorithms) + + +@pytest.mark.parametrize( + "algorithms", + [frozenset(), frozenset({"future-secret-algorithm"})], +) +def test_sender_resolution_requires_a_safe_advertised_algorithm_set( + algorithms: frozenset[str], +) -> None: + with pytest.raises(ValueError, match="advertised_algorithms"): + _resolution(_sender(), algorithms) + + def _outbox(pool: Any, sender: Any, **kwargs: Any) -> Any: from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox @@ -88,6 +113,19 @@ def _outbox(pool: Any, sender: Any, **kwargs: Any) -> Any: ) +def _resolver_outbox(pool: Any, resolver: Any, **kwargs: Any) -> Any: + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + with patch("adcp.decisioning.pg.task_webhook_outbox.PG_AVAILABLE", True): + return PgTaskWebhookOutbox( + pool=pool, + sender_resolver=resolver, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + **kwargs, + ) + + def test_prepare_mcp_binds_operation_id_key_and_body_before_delivery() -> None: from adcp.webhook_sender import WebhookSender @@ -156,6 +194,94 @@ def test_outbox_rejects_sender_without_sdk_pinned_transport() -> None: ) +def test_outbox_requires_exactly_one_sender_mode() -> None: + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + with patch("adcp.decisioning.pg.task_webhook_outbox.PG_AVAILABLE", True): + with pytest.raises(ValueError, match="exactly one"): + PgTaskWebhookOutbox( + pool=MagicMock(), + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + ) + with pytest.raises(ValueError, match="exactly one"): + PgTaskWebhookOutbox( + pool=MagicMock(), + sender=_sender(), + sender_resolver=MagicMock(), + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + ) + + +def test_scoped_registration_is_encrypted_and_mode_bound() -> None: + resolver = MagicMock(resolve=AsyncMock()) + outbox = _resolver_outbox(MagicMock(), resolver) + encrypted, nonce = outbox.protect_registration( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + url="https://buyer.example/webhook", + operation_id="op_1", + token=None, + signing_scope_id="tenant-key-scope-a", + ) + assert b"tenant-key-scope-a" not in encrypted + assert outbox._open_registration_with_scope( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + encrypted_registration=encrypted, + nonce=nonce, + ) == ("https://buyer.example/webhook", "op_1", None, "tenant-key-scope-a") + assert outbox.open_registration( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + encrypted_registration=encrypted, + nonce=nonce, + ) == ("https://buyer.example/webhook", "op_1", None) + + with pytest.raises(ValueError, match="require a signing_scope_id"): + outbox.protect_registration( + account_id="acct_1", + task_id="task_2", + task_type="create_media_buy", + url="https://buyer.example/webhook", + operation_id="op_2", + token=None, + ) + + +@pytest.mark.parametrize("scope", ["", "tenant\nscope", "x" * 256, "é" * 128]) +def test_signing_scope_id_is_bounded_and_control_free(scope: str) -> None: + outbox = _resolver_outbox(MagicMock(), MagicMock(resolve=AsyncMock())) + with pytest.raises(ValueError, match="signing_scope_id"): + outbox.protect_registration( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + url="https://buyer.example/webhook", + operation_id="op_1", + token=None, + signing_scope_id=scope, + ) + + +def test_fixed_sender_rejects_scoped_registration() -> None: + outbox = _outbox(MagicMock(), _sender()) + with pytest.raises(ValueError, match="fixed-sender"): + outbox.protect_registration( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + url="https://buyer.example/webhook", + operation_id="op_1", + token=None, + signing_scope_id="tenant-a", + ) + + def test_registration_is_encrypted_and_bound_at_issue_time() -> None: outbox = _outbox(MagicMock(), _sender()) encrypted, nonce = outbox.protect_registration( @@ -218,8 +344,9 @@ async def test_enqueue_persists_prepared_bytes_and_horizon_on_callers_connection assert "retry_horizon_seconds" in conn.execute.await_args.args[0] assert "retry_until" not in conn.execute.await_args.args[0] assert params[5] == "whk_1234567890123456" - assert params[6] == "acct_1" - assert params[7] != sender.prepare_mcp.return_value.body + assert params[6] is None + assert params[7] == "acct_1" + assert params[8] != sender.prepare_mcp.return_value.body assert params[-1] == 86_400 @@ -268,6 +395,7 @@ async def test_worker_claims_outside_http_and_acknowledges_success() -> None: "https://buyer.example/webhook", "op_1", "whk_1234567890123456", + None, encrypted, nonce, 1, @@ -285,6 +413,269 @@ async def test_worker_claims_outside_http_and_acknowledges_success() -> None: assert "state = 'delivered'" in ack_conn.execute.await_args.args[0] +@pytest.mark.asyncio +async def test_resolver_selects_fresh_sender_each_attempt_without_body_drift() -> None: + first_sender = _sender() + first_sender.send_prepared.return_value = WebhookDeliveryResult( + status_code=503, + idempotency_key="whk_1234567890123456", + url="https://buyer.example/webhook", + response_headers={}, + response_body=b"retry", + sent_body=first_sender.prepare_mcp.return_value.body, + ) + rotated_sender = _sender() + resolver = MagicMock( + resolve=AsyncMock(side_effect=[_resolution(first_sender), _resolution(rotated_sender)]) + ) + outbox = _resolver_outbox(MagicMock(), resolver) + body = first_sender.prepare_mcp.return_value.body + nonce = b"s" * 12 + encrypted = outbox._cipher.encrypt( + nonce, + body, + outbox._envelope_aad( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + status="completed", + url="https://buyer.example/webhook", + operation_id="op_1", + idempotency_key="whk_1234567890123456", + signing_scope_id="tenant-a", + ), + ) + + def claim(attempt: int) -> tuple[Any, ...]: + return ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + "https://buyer.example/webhook", + "op_1", + "whk_1234567890123456", + "tenant-a", + encrypted, + nonce, + attempt, + ) + + first_claim = _connection(None, claim(1)) + first_release = _connection(None) + second_claim = _connection(None, claim(2)) + second_ack = _connection(None) + outbox._pool = _pool(first_claim, first_release, second_claim, second_ack) + + assert await outbox.process_one() is True + assert await outbox.process_one() is True + + assert resolver.resolve.await_args_list[0].args == ("tenant-a",) + assert resolver.resolve.await_args_list[1].args == ("tenant-a",) + first_prepared = first_sender.send_prepared.await_args.args[0] + rotated_prepared = rotated_sender.send_prepared.await_args.args[0] + assert first_prepared.body == rotated_prepared.body == body + assert first_prepared.idempotency_key == rotated_prepared.idempotency_key + assert "state = CASE" in first_release.execute.await_args.args[0] + assert "state = 'delivered'" in second_ack.execute.await_args.args[0] + + +@pytest.mark.asyncio +async def test_signing_scope_column_is_authenticated_before_resolution() -> None: + resolver = MagicMock(resolve=AsyncMock(return_value=_resolution(_sender()))) + outbox = _resolver_outbox(MagicMock(), resolver) + body = _sender().prepare_mcp.return_value.body + nonce = b"s" * 12 + encrypted = outbox._cipher.encrypt( + nonce, + body, + outbox._envelope_aad( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + status="completed", + url="https://buyer.example/webhook", + operation_id="op_1", + idempotency_key="whk_1234567890123456", + signing_scope_id="tenant-a", + ), + ) + claim_conn = _connection( + None, + ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + "https://buyer.example/webhook", + "op_1", + "whk_1234567890123456", + "tenant-b", # DB-level scope substitution + encrypted, + nonce, + 1, + ), + ) + quarantine_conn = _connection(None) + outbox._pool = _pool(claim_conn, quarantine_conn) + + assert await outbox.process_one() is True + resolver.resolve.assert_not_awaited() + assert "state = 'invalid'" in quarantine_conn.execute.await_args.args[0] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("resolution_error", "expected_sql"), + [ + (ScopeTransientlyUnavailable(), "state = CASE"), + (ScopePermanentlyUnknown(), "state = 'invalid'"), + ], +) +async def test_scope_resolution_errors_retry_or_quarantine( + resolution_error: Exception, + expected_sql: str, +) -> None: + resolver = MagicMock(resolve=AsyncMock(side_effect=resolution_error)) + outbox = _resolver_outbox(MagicMock(), resolver) + body = _sender().prepare_mcp.return_value.body + nonce = b"s" * 12 + encrypted = outbox._cipher.encrypt( + nonce, + body, + outbox._envelope_aad( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + status="completed", + url="https://buyer.example/webhook", + operation_id="op_1", + idempotency_key="whk_1234567890123456", + signing_scope_id="tenant-a", + ), + ) + claim_conn = _connection( + None, + ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + "https://buyer.example/webhook", + "op_1", + "whk_1234567890123456", + "tenant-a", + encrypted, + nonce, + 1, + ), + ) + settle_conn = _connection(None) + outbox._pool = _pool(claim_conn, settle_conn) + + assert await outbox.process_one() is True + assert expected_sql in settle_conn.execute.await_args.args[0] + + +@pytest.mark.asyncio +async def test_sender_resolution_is_bounded_by_the_delivery_lease() -> None: + async def never_resolves(_scope: str) -> Any: + await asyncio.Event().wait() + + resolver = MagicMock(resolve=AsyncMock(side_effect=never_resolves)) + outbox = _resolver_outbox(MagicMock(), resolver) + # Exercise the real aggregate lease timeout without making the test wait. + outbox._lease_seconds = 1.01 + body = _sender().prepare_mcp.return_value.body + nonce = b"s" * 12 + encrypted = outbox._cipher.encrypt( + nonce, + body, + outbox._envelope_aad( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + status="completed", + url="https://buyer.example/webhook", + operation_id="op_1", + idempotency_key="whk_1234567890123456", + signing_scope_id="tenant-a", + ), + ) + claim_conn = _connection( + None, + ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + "https://buyer.example/webhook", + "op_1", + "whk_1234567890123456", + "tenant-a", + encrypted, + nonce, + 1, + ), + ) + release_conn = _connection(None) + outbox._pool = _pool(claim_conn, release_conn) + + assert await asyncio.wait_for(outbox.process_one(), timeout=0.25) is True + resolver.resolve.assert_awaited_once_with("tenant-a") + assert "state = CASE" in release_conn.execute.await_args.args[0] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "invalid_mutation", + [ + lambda sender: setattr(sender, "_owns_client", False), + lambda sender: setattr(sender, "_allow_private_destinations", True), + lambda sender: setattr(sender, "signs_with_rfc9421", False), + lambda sender: setattr(sender, "_timeout", 60.0), + ], +) +async def test_resolved_sender_is_revalidated_on_every_attempt(invalid_mutation) -> None: + sender = _sender() + invalid_mutation(sender) + outbox = _resolver_outbox( + MagicMock(), MagicMock(resolve=AsyncMock(return_value=_resolution(sender))) + ) + + with pytest.raises(ScopePermanentlyUnknown): + await outbox._resolve_delivery_sender("tenant-a") + + +@pytest.mark.asyncio +async def test_resolved_sender_algorithm_must_match_scope_advertisement() -> None: + sender = _sender() + outbox = _resolver_outbox( + MagicMock(), + MagicMock( + resolve=AsyncMock(return_value=_resolution(sender, frozenset({"ecdsa-p256-sha256"}))) + ), + ) + + with pytest.raises(ScopePermanentlyUnknown): + await outbox._resolve_delivery_sender("tenant-a") + sender.send_prepared.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_untyped_resolver_failure_is_sanitized_as_transient() -> None: + resolver = MagicMock(resolve=AsyncMock(side_effect=RuntimeError("key vault secret"))) + outbox = _resolver_outbox(MagicMock(), resolver) + + with pytest.raises(ScopeTransientlyUnavailable) as exc_info: + await outbox._resolve_delivery_sender("tenant-a") + assert "key vault secret" not in str(exc_info.value) + + @pytest.mark.asyncio async def test_worker_releases_failed_delivery_for_horizon_retry() -> None: sender = _sender() @@ -322,6 +713,7 @@ async def test_worker_releases_failed_delivery_for_horizon_retry() -> None: "https://buyer.example/webhook", "op_1", "whk_1234567890123456", + None, encrypted, nonce, 1, @@ -353,6 +745,7 @@ async def test_worker_quarantines_body_that_breaks_authenticated_envelope() -> N "https://buyer.example/webhook", "op_1", "whk_1234567890123456", + None, body, b"n" * 12, 1, @@ -385,11 +778,13 @@ async def test_registry_completion_enqueues_on_same_transaction_connection() -> None, ) outbox = AsyncMock() - outbox.open_registration = MagicMock( + outbox._sender_resolver = None + outbox._open_registration_with_scope = MagicMock( return_value=( "https://buyer.example/webhook", "op_1", "buyer-token", + None, ) ) pool = _pool(conn) @@ -412,8 +807,9 @@ async def test_registry_completion_enqueues_on_same_transaction_connection() -> url="https://buyer.example/webhook", operation_id="op_1", token="buyer-token", + signing_scope_id=None, ) - outbox.open_registration.assert_called_once_with( + outbox._open_registration_with_scope.assert_called_once_with( account_id="acct_1", task_id="task_1", task_type="create_media_buy", @@ -439,11 +835,13 @@ async def test_registry_failure_enqueues_on_same_explicit_transaction() -> None: None, ) outbox = AsyncMock() - outbox.open_registration = MagicMock( + outbox._sender_resolver = None + outbox._open_registration_with_scope = MagicMock( return_value=( "https://buyer.example/webhook", "op_1", None, + None, ) ) pool = _pool(conn) @@ -464,11 +862,49 @@ async def test_registry_failure_enqueues_on_same_explicit_transaction() -> None: url="https://buyer.example/webhook", operation_id="op_1", token=None, + signing_scope_id=None, ) conn.transaction.assert_called_once_with() assert "webhook_registration = NULL" in conn.execute.await_args_list[-1].args[0] +@pytest.mark.asyncio +async def test_registry_derives_scope_only_from_hydrated_request_context() -> None: + from adcp.decisioning import RequestContext + from adcp.decisioning.pg.task_registry import PgTaskRegistry + + pool = MagicMock() + resolver_outbox = _resolver_outbox(pool, MagicMock(resolve=AsyncMock())) + scope_hook = AsyncMock(return_value="opaque-tenant-scope") + with patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True): + registry = PgTaskRegistry( + pool=pool, + task_webhook_outbox=resolver_outbox, + webhook_signing_scope_resolver=scope_hook, + ) + context = RequestContext(tenant_id="internal-tenant") + + assert await registry.resolve_webhook_signing_scope(context) == "opaque-tenant-scope" + scope_hook.assert_awaited_once_with(context) + + +def test_registry_requires_scope_hook_exactly_for_resolver_outbox() -> None: + from adcp.decisioning.pg.task_registry import PgTaskRegistry + + pool = MagicMock() + resolver_outbox = _resolver_outbox(pool, MagicMock(resolve=AsyncMock())) + fixed_outbox = _outbox(pool, _sender()) + with patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True): + with pytest.raises(ValueError, match="required exactly"): + PgTaskRegistry(pool=pool, task_webhook_outbox=resolver_outbox) + with pytest.raises(ValueError, match="required exactly"): + PgTaskRegistry( + pool=pool, + task_webhook_outbox=fixed_outbox, + webhook_signing_scope_resolver=lambda _context: "scope", + ) + + @pytest.mark.asyncio async def test_registry_strips_credentials_before_terminal_update() -> None: from adcp.decisioning.pg.task_registry import PgTaskRegistry diff --git a/tests/test_webhook_signing_capabilities.py b/tests/test_webhook_signing_capabilities.py index 25a03cdde..75ed68f59 100644 --- a/tests/test_webhook_signing_capabilities.py +++ b/tests/test_webhook_signing_capabilities.py @@ -19,7 +19,7 @@ import copy import json from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest @@ -223,6 +223,29 @@ def _sdk_registry_with_outbox(sender: WebhookSender, horizon: int = 86400): return PgTaskRegistry(pool=pool, task_webhook_outbox=outbox) +def _sdk_registry_with_sender_resolver(horizon: int = 86400): + from adcp.decisioning.pg.task_registry import PgTaskRegistry + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + pool = MagicMock() + resolver = MagicMock(resolve=AsyncMock()) + with ( + patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True), + patch("adcp.decisioning.pg.task_webhook_outbox.PG_AVAILABLE", True), + ): + outbox = PgTaskWebhookOutbox( + pool=pool, + sender_resolver=resolver, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=horizon, + ) + return PgTaskRegistry( + pool=pool, + task_webhook_outbox=outbox, + webhook_signing_scope_resolver=lambda context: str(context.tenant_id), + ) + + def test_boot_passes_when_capabilities_omit_webhook_signing() -> None: """No advertisement, no obligation — validator returns silently.""" validate_webhook_signing_for_capabilities( @@ -411,6 +434,21 @@ def test_boot_accepts_registry_backed_atomic_outbox() -> None: ) +def test_boot_accepts_registry_backed_tenant_sender_resolver() -> None: + validate_webhook_signing_for_capabilities( + capabilities=_Caps( + webhook_signing=WebhookSigning( + supported=True, + delivery_retry_horizon_seconds=86400, + algorithms=["ed25519"], + ) + ), + sender=None, + supervisor=None, + registry=_sdk_registry_with_sender_resolver(), + ) + + def test_boot_rejects_pg_registry_subclass_that_can_override_webhook_issue() -> None: """Only the audited exact registry type may claim SDK-managed delivery.""" from adcp.decisioning.pg.task_registry import PgTaskRegistry diff --git a/tests/type_checks/webhook_sender_resolver.py b/tests/type_checks/webhook_sender_resolver.py new file mode 100644 index 000000000..a31a2c1b7 --- /dev/null +++ b/tests/type_checks/webhook_sender_resolver.py @@ -0,0 +1,34 @@ +"""Adopter-facing type checks for tenant-scoped webhook sender resolution.""" + +from adcp import ( + ScopePermanentlyUnknown, + ScopeTransientlyUnavailable, + WebhookSender, + WebhookSenderResolution, + WebhookSenderResolver, +) +from adcp.decisioning import RequestContext, WebhookSigningScopeResolver + + +class TenantWebhookSenders: + async def resolve(self, signing_scope_id: str) -> WebhookSenderResolution: + if signing_scope_id == "decommissioned": + raise ScopePermanentlyUnknown + raise ScopeTransientlyUnavailable + + +resolver: WebhookSenderResolver = TenantWebhookSenders() + + +def resolve_signing_scope(context: RequestContext[object]) -> str: + return str(context.tenant_id) + + +scope_resolver: WebhookSigningScopeResolver = resolve_signing_scope + + +def bind_sender(sender: WebhookSender) -> WebhookSenderResolution: + return WebhookSenderResolution( + sender=sender, + advertised_algorithms=frozenset({"ed25519"}), + )