From dc0a4bbf71ff72cc6e340e3d46e79775767bc08d Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 23 Aug 2026 05:50:05 +0200 Subject: [PATCH 1/3] feat(webhooks): add durable task webhook outbox --- .github/workflows/ci.yml | 1 + CHANGELOG.md | 5 + MIGRATION_v7_to_v8.md | 68 ++ README.md | 2 +- docs/decisioning-capabilities.md | 9 + docs/handler-authoring.md | 64 +- .../migration-from-fragmented-senders.md | 10 +- src/adcp/__init__.py | 3 + src/adcp/decisioning/__init__.py | 15 + src/adcp/decisioning/dispatch.py | 66 +- src/adcp/decisioning/handler.py | 34 +- src/adcp/decisioning/pg/__init__.py | 5 + src/adcp/decisioning/pg/decisioning_tasks.sql | 10 + src/adcp/decisioning/pg/task_registry.py | 174 ++++- .../decisioning/pg/task_webhook_outbox.py | 641 ++++++++++++++++++ .../decisioning/pg/task_webhook_outbox.sql | 47 ++ src/adcp/decisioning/serve.py | 20 +- src/adcp/decisioning/task_registry.py | 10 + src/adcp/decisioning/webhook_emit.py | 153 ++++- src/adcp/signing/jwks.py | 6 +- src/adcp/webhook_sender.py | 160 ++++- src/adcp/webhooks.py | 2 + .../test_pg_task_webhook_outbox.py | 202 ++++++ tests/fixtures/public_api_snapshot.json | 1 + tests/test_decisioning_webhook_emit.py | 51 ++ tests/test_decisioning_workflow_handoff.py | 42 ++ tests/test_task_webhook_outbox_pg.py | 479 +++++++++++++ tests/test_webhook_signing_capabilities.py | 89 ++- 28 files changed, 2276 insertions(+), 93 deletions(-) create mode 100644 src/adcp/decisioning/pg/task_webhook_outbox.py create mode 100644 src/adcp/decisioning/pg/task_webhook_outbox.sql create mode 100644 tests/conformance/decisioning/test_pg_task_webhook_outbox.py create mode 100644 tests/test_task_webhook_outbox_pg.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c935a84a4..9739ff306 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -129,6 +129,7 @@ jobs: tests/conformance/signing/test_pg_replay_store_e2e.py \ tests/conformance/decisioning/test_pg_buyer_agent_registry.py \ tests/conformance/decisioning/test_pg_idempotency_backend.py \ + tests/conformance/decisioning/test_pg_task_webhook_outbox.py \ -v conventional-commits: diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bb2dd9d3..c94eee709 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ ### ⚠ BREAKING CHANGES * **protocol:** add AdCP 3.2 beta.5 parity ([#1064](https://github.com/adcontextprotocol/adcp-client-python/issues/1064)) +* **task webhooks:** MCP push registrations now require a buyer-supplied + `operation_id`. Publishers advertising webhook delivery must retain an + immutable payload/idempotency binding and retry state for the advertised + 1–7 day horizon; Submitted responses must be registry-backed. See + `MIGRATION_v7_to_v8.md` for the beta.6 → beta.7 rollout paths. ### Features diff --git a/MIGRATION_v7_to_v8.md b/MIGRATION_v7_to_v8.md index 62c360b52..2ba418fd8 100644 --- a/MIGRATION_v7_to_v8.md +++ b/MIGRATION_v7_to_v8.md @@ -7,6 +7,74 @@ AdCP 3.2.0-beta.4 and adds the compact product/media-buy lifecycle. The old for lifecycle selection, capability declarations, and the compatibility test matrix. +## 8.0.0-beta.6 to beta.7: task webhook durability + +Beta.7 enforces the AdCP 3.2 task-webhook contract at runtime. This is a +behavioral migration, not only a generated-type update: + +- A push-configured MCP task must include buyer-supplied + `push_notification_config.operation_id`; sellers echo it verbatim and never + recover it from the callback URL. +- A seller advertising `webhook_signing.supported=True` must advertise a + `delivery_retry_horizon_seconds` from 86,400 through 604,800 seconds. +- Every Submitted response must be created through `TaskHandoff` or + `WorkflowHandoff`, so the task is registry-backed and pollable. Hand-rolled + `{"task_id": ..., "status": "submitted"}` responses are rejected. +- Receivers use claim/acknowledge/release deduplication. A claim is + acknowledged only after application processing succeeds and is released on + failure so an exact retry remains processable. + +For SDK-managed publication, install the PostgreSQL extra and couple the task +registry to the durable outbox: + +```python +pool = AsyncConnectionPool(database_url, open=False) +outbox = PgTaskWebhookOutbox( + pool=pool, + sender=WebhookSender.from_jwk(private_webhook_jwk), + # Load a stable 32-byte secret from your secret manager. Do not rotate it + # until all rows encrypted under the old value have passed their horizon. + encryption_key=task_webhook_encryption_key, + delivery_retry_horizon_seconds=86_400, +) +registry = PgTaskRegistry(pool=pool, task_webhook_outbox=outbox) + +async def startup(): + await pool.open() + await registry.create_schema() + await outbox.create_schema() + +async def shutdown(): + await pool.close() + +serve( + platform, + registry=registry, + transport="both", + on_startup=(startup,), + on_shutdown=(shutdown,), +) +``` + +In each separately supervised worker process, construct another outbox against +that process's pool and run `await outbox.run_worker()` using the same database +and encryption key. Multiple replicas are safe. Do not launch an +unretained `asyncio.create_task()` immediately before synchronous `serve()`; +that task does not share the server lifecycle. + +The registry commits terminal state and the immutable prepared webhook in one +transaction. The body and callback token are AES-256-GCM encrypted at rest and +bound to the task, account, URL, operation, status, and idempotency key. The +retry horizon starts at the first delivery attempt; the worker replays the same +body/key and retains proof until that exact advertised horizon ends. + +If publication is owned outside this SDK instead, set +`webhook_signing_managed_externally=True`, set +`auto_emit_task_webhooks=False`, and leave SDK webhook sender/supervisor wiring +empty. The external publisher must provide the same atomic state/outbox, +retention, exact-retry, and reconciliation guarantees. Sellers without either +publisher must stop advertising task webhooks and operate polling-only. + ## Brand identity imports The generated `adcp.types.generated_poc.brand.Brand` path was private and is diff --git a/README.md b/README.md index 05c1dbec3..45c53400e 100644 --- a/README.md +++ b/README.md @@ -269,7 +269,7 @@ async with ADCPMultiAgentClient( print(f"✅ Sync completion: {len(result.data.products)} products") if result.status == "submitted": - # Poll or let the seller's external durable publisher send the webhook. + # Poll or let the seller's durable publisher send the webhook. print(f"⏳ Async - webhook registered at: {result.submitted.webhook_url}") # Connections automatically cleaned up here ``` diff --git a/docs/decisioning-capabilities.md b/docs/decisioning-capabilities.md index 172a2ea55..c3bf0df54 100644 --- a/docs/decisioning-capabilities.md +++ b/docs/decisioning-capabilities.md @@ -173,6 +173,15 @@ class MultiTenantSeller(DecisioningPlatform): ) ``` +For SDK-managed terminal task publication, configure `PgTaskWebhookOutbox` on +the same `PgTaskRegistry` and connection pool that own the task lifecycle. +Custom marker-compatible objects are rejected; their publishers use the +external-owner contract instead. The outbox horizon must exactly equal the +request-scoped advertised horizon. Keep +`webhook_signing_managed_externally=False` and +`auto_emit_task_webhooks=True`; server boot validates the outbox's RFC 9421 +sender and atomic durability markers. + Set `webhook_signing_managed_externally=True` only when an external durable outbox signs and publishes outbound webhooks outside the SDK sender stack. Start the handler with `auto_emit_task_webhooks=False` and do not wire an SDK sender or diff --git a/docs/handler-authoring.md b/docs/handler-authoring.md index c28ca2c2c..fc3677776 100644 --- a/docs/handler-authoring.md +++ b/docs/handler-authoring.md @@ -1296,10 +1296,59 @@ and no registry task exists for a webhook `task_id`. `TaskHandoff` and `WorkflowHandoff` are different: a push-configured asynchronous task requires a durable publisher that atomically records the terminal result and -its outbound delivery before acknowledging completion. The SDK does not yet -provide that outbox. With the default `auto_emit_task_webhooks=True`, it therefore -rejects a push-configured handoff before task creation instead of advertising -delivery it cannot guarantee. +its outbound delivery before acknowledging completion. The PostgreSQL extra now +provides that contract through `PgTaskWebhookOutbox` coupled to `PgTaskRegistry`. +With the default `auto_emit_task_webhooks=True`, push-configured handoffs are +admitted only when this atomic registry/outbox pair is wired, or when a conformant +external owner is declared. + +```python +from adcp.decisioning import PgTaskRegistry, PgTaskWebhookOutbox, serve +from adcp.webhook_sender import WebhookSender +from psycopg_pool import AsyncConnectionPool + +pool = AsyncConnectionPool("postgresql://...", open=False) +sender = WebhookSender.from_jwk(webhook_signing_jwk_with_private_d) +outbox = PgTaskWebhookOutbox( + pool=pool, + sender=sender, + encryption_key=task_webhook_encryption_key, # stable 32-byte secret-manager value + delivery_retry_horizon_seconds=86_400, +) +registry = PgTaskRegistry(pool=pool, task_webhook_outbox=outbox) + +async def startup(): + await pool.open() + await registry.create_schema() + await outbox.create_schema() + +async def shutdown(): + await pool.close() + +# Keep webhook_signing_managed_externally=False and the default +# auto_emit_task_webhooks=True. The outbox sender is validated at boot. +serve( + my_platform, + registry=registry, + transport="both", + on_startup=(startup,), + on_shutdown=(shutdown,), +) +``` + +In a separately supervised worker process, construct another outbox against +that process's pool and run `await outbox.run_worker()` with the same database +and encryption key; multiple replicas are safe. Avoid an +unretained `create_task()` beside synchronous `serve()`, because it is not tied +to the server's startup/shutdown lifecycle. + +The task registry captures URL, buyer-supplied `operation_id`, and token as an +encrypted, authenticated registration when the task is issued. Its +`complete()` / `fail()` transaction writes terminal +state and the encrypted, authenticated webhook envelope together. The task-row +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. Production adopters may set `auto_emit_task_webhooks=False` only when an external durable outbox owns publication, retries, immutable body/key retention, and @@ -1408,9 +1457,10 @@ if event_type in subscription.event_types: `WebhookSender` is the transport layer — it constructs and signs one HTTP POST. `InMemoryWebhookDeliverySupervisor` adds best-effort retries and circuit breakers -for local development. Neither it nor the current PostgreSQL supervisor implements -the atomic terminal-state/outbox contract required to back an advertised AdCP 3.2 -retry horizon; the framework never auto-selects them for TaskHandoff publication. +for local development. Neither it nor `PgWebhookDeliverySupervisor` implements +the atomic terminal-state/outbox contract required for TaskHandoff publication; +use `PgTaskWebhookOutbox` with `PgTaskRegistry` for that lifecycle. The older +supervisors remain explicit/manual delivery utilities. ```python import os diff --git a/docs/webhooks/migration-from-fragmented-senders.md b/docs/webhooks/migration-from-fragmented-senders.md index c65d11035..f4e97f7ec 100644 --- a/docs/webhooks/migration-from-fragmented-senders.md +++ b/docs/webhooks/migration-from-fragmented-senders.md @@ -224,11 +224,13 @@ await supervisor.deliver(WebhookDeliveryRequest(url=..., payload=..., ...)) The in-memory supervisor handles retry policy and circuit-breaker state per buyer, but it is process-local and cannot support the AdCP 3.2 retry-horizon capability. -The current `PgWebhookDeliverySupervisor` persists pending attempts but removes +`PgWebhookDeliverySupervisor` persists pending attempts but removes final rows, so it also does not satisfy the beta.5 retention or atomic terminal -state/outbox contract. Use these APIs for explicit best-effort delivery only. A -webhook-emitting production seller must own publication in an external durable -outbox, set `auto_emit_task_webhooks=False`, and advertise the horizon with +state/outbox contract. Use these APIs for explicit best-effort delivery only. +For framework-managed task webhooks, pair `PgTaskWebhookOutbox` with +`PgTaskRegistry` on the same pool, supply a stable 32-byte outbox encryption +key, and run a separately supervised outbox worker; for an adopter-managed publisher, set +`auto_emit_task_webhooks=False` and advertise the horizon with `webhook_signing_managed_externally=True`. ### Pattern: per-call SSRF check diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index 14396ad43..4f9dcbcd8 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -695,6 +695,7 @@ def _resolve_version() -> str: "adcp.webhooks": ( "LegacyHmacFallback", "MemoryBackend", + "PreparedWebhook", "WebhookChallengeError", "WebhookChallengeResult", "WebhookDedupStore", @@ -922,6 +923,7 @@ def get_adcp_version() -> str: "WebhookDedupStore", "MemoryBackend", "LegacyHmacFallback", + "PreparedWebhook", "McpWebhookPayload", # Account operations "AccountAuthorization", @@ -2065,6 +2067,7 @@ def get_adcp_version() -> str: from adcp.webhooks import ( LegacyHmacFallback, MemoryBackend, + PreparedWebhook, WebhookChallengeError, WebhookChallengeResult, WebhookDedupStore, diff --git a/src/adcp/decisioning/__init__.py b/src/adcp/decisioning/__init__.py index f7d6d5e08..438432bad 100644 --- a/src/adcp/decisioning/__init__.py +++ b/src/adcp/decisioning/__init__.py @@ -277,6 +277,7 @@ def create_media_buy( from adcp.decisioning.pg import ( # noqa: F401 PgProposalStore, PgTaskRegistry, + PgTaskWebhookOutbox, PostgresTaskRegistry, ) except ImportError: # pragma: no cover — exercised by the [pg] extra tests @@ -317,6 +318,19 @@ def __init__(self, *args: object, **kwargs: object) -> None: "(Poetry: `poetry add 'adcp[pg]'`)." ) + class PgTaskWebhookOutbox: # type: ignore[no-redef] + """Stub raised when ``adcp[pg]`` isn't installed.""" + + delivery_state_is_durable: _ClassVar[bool] = True + supports_atomic_task_outbox: _ClassVar[bool] = True + + def __init__(self, *args: object, **kwargs: object) -> None: + raise ImportError( + "PgTaskWebhookOutbox requires psycopg3 and psycopg-pool. " + "Install the 'pg' extra: `pip install 'adcp[pg]'` " + "(Poetry: `poetry add 'adcp[pg]'`)." + ) + __all__ = [ "Account", @@ -378,6 +392,7 @@ def __init__(self, *args: object, **kwargs: object) -> None: "PermissionDeniedError", "PgProposalStore", "PgTaskRegistry", + "PgTaskWebhookOutbox", "LazyPlatformRouter", "PlatformRouter", "PostgresTaskRegistry", diff --git a/src/adcp/decisioning/dispatch.py b/src/adcp/decisioning/dispatch.py index 16abde7b7..04ad02674 100644 --- a/src/adcp/decisioning/dispatch.py +++ b/src/adcp/decisioning/dispatch.py @@ -112,6 +112,7 @@ def _is_framework_submitted_projection(value: Any) -> bool: return isinstance(value, _FrameworkSubmittedProjection) + # Strong references for synchronous adopter lifecycles that outlive a # cancelled request. A Python thread cannot be cancelled; its completion hooks # must still settle durable proposal/idempotency state. @@ -1414,12 +1415,13 @@ async def _invoke_platform_method( call in the handler shim. :param webhook_auto_emit: Forwarded to :func:`_project_handoff` as - the async task-webhook delivery gate. SDK-managed publication is - rejected under the beta.5 durability contract. + the legacy post-terminal delivery gate. Registry-owned atomic outbox + publication disables this gate because ``complete`` / ``fail`` enqueue + within their terminal-state transaction. :param webhook_external_owner_ready: Whether static capabilities and - trusted wiring prove that a conformant external outbox owns terminal - task publication. Disabling SDK emission without this proof fails - closed for push-configured handoffs. + trusted wiring prove that a conformant registry-backed or external + outbox owns terminal task publication. The compatibility name is + retained for callers compiled against the earlier beta. :param task_type: Optional canonical AdCP task type when the Python platform method uses an internal compatibility name. Handoff admission, registry rows, and callbacks use this wire identity; @@ -2029,12 +2031,10 @@ async def _project_handoff( ``webhook_auto_emit=False`` and does not invoke this target. :param webhook_auto_emit: Async task-webhook delivery gate. The - production handler defaults this to ``True`` and rejects a - push-configured handoff because the SDK cannot honor beta.5 durable - publication. Callers pass ``False`` only when an external outbox owns - terminal task delivery. + production handler resolves this to ``False`` when a registry-backed + or external outbox owns terminal task delivery. :param webhook_external_owner_ready: Trusted configuration proof that the - external outbox advertised by the platform owns this publication. + configured durable outbox owns this publication. The handoff fn is extracted via the type-identity dispatch in :func:`adcp.decisioning.types.is_task_handoff`. Subclassed @@ -2059,11 +2059,24 @@ async def _project_handoff( # at issue-time means the terminal-state helpers (_fail, # registry.complete) never need to know about request-side # context — keeps the wire-shape boundary in one place. - task_id = await registry.issue( - account_id=ctx.account.id, - task_type=method_name, - request_context=_extract_request_context(request_params), - ) + issue_kwargs: dict[str, Any] = { + "account_id": ctx.account.id, + "task_type": method_name, + "request_context": _extract_request_context(request_params), + } + push_target = _extract_push_notification_url_and_token(request_params) + if ( + method_name in SPEC_WEBHOOK_TASK_TYPES + and push_target is not None + and getattr(registry, "task_webhook_outbox", None) is not None + ): + push_url, push_token = push_target + issue_kwargs.update( + webhook_url=push_url, + webhook_operation_id=_extract_push_operation_id(request_params), + webhook_token=push_token, + ) + task_id = await registry.issue(**issue_kwargs) # Hand off to background. The wire envelope returns immediately; # the fn runs to completion in the background and persists the @@ -2318,11 +2331,24 @@ async def _project_workflow_handoff( # WorkflowHandoff path persists the task and immediately returns # — the adopter's enqueue fn does not write to the registry — so # context capture must happen here at issue-time too. - task_id = await registry.issue( - account_id=ctx.account.id, - task_type=method_name, - request_context=_extract_request_context(request_params), - ) + issue_kwargs: dict[str, Any] = { + "account_id": ctx.account.id, + "task_type": method_name, + "request_context": _extract_request_context(request_params), + } + push_target = _extract_push_notification_url_and_token(request_params) + if ( + method_name in SPEC_WEBHOOK_TASK_TYPES + and push_target is not None + and getattr(registry, "task_webhook_outbox", None) is not None + ): + push_url, push_token = push_target + issue_kwargs.update( + webhook_url=push_url, + webhook_operation_id=_extract_push_operation_id(request_params), + webhook_token=push_token, + ) + task_id = await registry.issue(**issue_kwargs) handoff_ctx = TaskHandoffContext(id=task_id, _registry=registry) try: diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index 69c3d27b5..b86f76917 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -96,8 +96,8 @@ ) from adcp.decisioning.webhook_emit import ( _extract_push_notification_url_and_token, - external_task_webhook_owner_ready, maybe_emit_sync_completion, + task_webhook_owner_ready, ) from adcp.server.base import ADCPHandler, NotImplementedResponse, ToolContext @@ -1681,11 +1681,9 @@ async def _handoff_webhook_kwargs( """Webhook delivery kwargs threaded into :func:`_invoke_platform_method` for the async (handoff) completion path. - SDK-managed TaskHandoff push is unavailable under the beta.5 durable - publication contract. The target is retained only for low-level - compatibility; push-configured handoffs are admitted solely when the - platform declares a conformant external owner and leaves that target - unwired. + A registry-backed atomic outbox owns SDK-managed publication. The + low-level target remains available for manual compatibility paths; + an externally managed publisher instead leaves it unwired. """ capabilities = self._platform.capabilities if _extract_push_notification_url_and_token(params) is not None: @@ -1702,17 +1700,26 @@ async def _handoff_webhook_kwargs( sender=self._webhook_sender, supervisor=self._webhook_supervisor, auto_emit_task_webhooks=self._auto_emit_task_webhooks, + registry=self._registry, ) + owner_ready = task_webhook_owner_ready( + capabilities=capabilities, + sender=self._webhook_sender, + supervisor=self._webhook_supervisor, + auto_emit_task_webhooks=self._auto_emit_task_webhooks, + registry=self._registry, + ) + registry_owns_delivery = ( + owner_ready and getattr(self._registry, "task_webhook_outbox", None) is not None + ) return { "webhook_target": self._webhook_supervisor or self._webhook_sender, - "webhook_auto_emit": self._auto_emit_task_webhooks, - "webhook_external_owner_ready": external_task_webhook_owner_ready( - capabilities=capabilities, - sender=self._webhook_sender, - supervisor=self._webhook_supervisor, - auto_emit_task_webhooks=self._auto_emit_task_webhooks, - ), + # Registry-owned delivery is enqueued by complete()/fail() in the + # same transaction; the post-terminal compatibility emitter must + # remain off or it would create a second notification. + "webhook_auto_emit": self._auto_emit_task_webhooks and not registry_owns_delivery, + "webhook_external_owner_ready": owner_ready, } def _build_ctx( @@ -1825,6 +1832,7 @@ async def get_adcp_capabilities( sender=self._webhook_sender, supervisor=self._webhook_supervisor, auto_emit_task_webhooks=self._auto_emit_task_webhooks, + registry=self._registry, ) # ----- supported_protocols: explicit override > derive from specialisms ----- diff --git a/src/adcp/decisioning/pg/__init__.py b/src/adcp/decisioning/pg/__init__.py index fe3c95e71..e9f9ec9ca 100644 --- a/src/adcp/decisioning/pg/__init__.py +++ b/src/adcp/decisioning/pg/__init__.py @@ -18,6 +18,9 @@ :class:`~adcp.decisioning.InMemoryTaskRegistry` that satisfies the production-mode durability gate. (``PostgresTaskRegistry`` is the pre-4.4 name and remains as a deprecated alias through 4.4.x.) +* :class:`PgTaskWebhookOutbox` — atomic terminal task-webhook publication + coupled to ``PgTaskRegistry``, with lease-based workers and 1–7 day retry + retention. The schema DDL ships alongside the Python code (e.g. ``adcp/decisioning/pg/buyer_agent_registry.sql``, @@ -34,6 +37,7 @@ ) from adcp.decisioning.pg.proposal_store import PgProposalStore from adcp.decisioning.pg.task_registry import PgTaskRegistry, PostgresTaskRegistry +from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox __all__ = [ "DEFAULT_TABLE_NAME", @@ -41,5 +45,6 @@ "PgBuyerAgentRegistry", "PgProposalStore", "PgTaskRegistry", + "PgTaskWebhookOutbox", "PostgresTaskRegistry", ] diff --git a/src/adcp/decisioning/pg/decisioning_tasks.sql b/src/adcp/decisioning/pg/decisioning_tasks.sql index a92460312..76ca24216 100644 --- a/src/adcp/decisioning/pg/decisioning_tasks.sql +++ b/src/adcp/decisioning/pg/decisioning_tasks.sql @@ -20,12 +20,22 @@ CREATE TABLE IF NOT EXISTS decisioning_tasks ( progress JSONB, result JSONB, error JSONB, + request_context JSONB, + -- Immutable callback registration captured when the task is issued. + -- PgTaskRegistry uses these columns to enqueue the terminal webhook in + -- the same transaction as the completed/failed state transition. + webhook_registration BYTEA, + webhook_registration_nonce BYTEA, -- Unix epoch seconds (float), matches TaskRecord.created_at/updated_at -- so Python round-trips the value without lossy TIMESTAMPTZ conversion. created_at DOUBLE PRECISION NOT NULL, updated_at DOUBLE PRECISION NOT NULL ); +ALTER TABLE decisioning_tasks ADD COLUMN IF NOT EXISTS request_context JSONB; +ALTER TABLE decisioning_tasks ADD COLUMN IF NOT EXISTS webhook_registration BYTEA; +ALTER TABLE decisioning_tasks ADD COLUMN IF NOT EXISTS webhook_registration_nonce BYTEA; + -- Supports the cross-tenant get() query: WHERE task_id = $1 AND account_id = $2. -- Without this index, every tasks/get is a full-table scan on account_id. CREATE INDEX IF NOT EXISTS decisioning_tasks_account_idx diff --git a/src/adcp/decisioning/pg/task_registry.py b/src/adcp/decisioning/pg/task_registry.py index cb15b758e..1a7c63d74 100644 --- a/src/adcp/decisioning/pg/task_registry.py +++ b/src/adcp/decisioning/pg/task_registry.py @@ -68,9 +68,13 @@ async def main(): import uuid from typing import TYPE_CHECKING, Any, ClassVar +from adcp.decisioning.account_projection import strip_credentials_from_wire_result + if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + try: from psycopg_pool import AsyncConnectionPool as _AsyncConnectionPool # noqa: F401 @@ -107,6 +111,10 @@ class PgTaskRegistry: Each registry operation acquires a short-lived connection from the pool and returns it immediately after the query. No long-lived transactions, no cross-operation state. + task_webhook_outbox: + Optional :class:`PgTaskWebhookOutbox` sharing this pool. When a task + carries callback registration, ``complete`` / ``fail`` enqueue its + immutable terminal webhook on the same transaction connection. Notes ----- @@ -118,20 +126,33 @@ class PgTaskRegistry: is_durable: ClassVar[bool] = True - def __init__(self, *, pool: AsyncConnectionPool, _table: str = _DEFAULT_TABLE) -> None: + def __init__( + self, + *, + pool: AsyncConnectionPool, + task_webhook_outbox: PgTaskWebhookOutbox | None = None, + _table: str = _DEFAULT_TABLE, + ) -> None: if not PG_AVAILABLE: raise ImportError(_INSTALL_HINT) if not _SAFE_IDENTIFIER_RE.fullmatch(_table): raise ValueError(f"_table must match [a-z_][a-z0-9_]* (ASCII only), got {_table!r}") + if task_webhook_outbox is not None and task_webhook_outbox._pool is not pool: + raise ValueError( + "PgTaskRegistry and PgTaskWebhookOutbox must use the same connection pool" + ) self._pool = pool self._table = _table + self.task_webhook_outbox = task_webhook_outbox + 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. # _table is whitelisted by _SAFE_IDENTIFIER_RE above. self._sql_insert = ( # noqa: S608 — table name is whitelisted f"INSERT INTO {self._table}" - f" (task_id, account_id, state, task_type, created_at, updated_at)" - f" VALUES (%s, %s, 'submitted', %s, %s, %s)" + f" (task_id, account_id, state, task_type, request_context," + f" webhook_registration, webhook_registration_nonce, created_at, updated_at)" + f" VALUES (%s, %s, 'submitted', %s, %s::jsonb, %s, %s, %s, %s)" ) self._sql_update_progress = ( # noqa: S608 f"UPDATE {self._table}" @@ -143,13 +164,19 @@ def __init__(self, *, pool: AsyncConnectionPool, _table: str = _DEFAULT_TABLE) - f"UPDATE {self._table}" f" SET state = 'completed', result = %s::jsonb, updated_at = %s" f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" - f" RETURNING task_id" + f" RETURNING task_id, account_id, task_type, webhook_registration," + f" webhook_registration_nonce" ) self._sql_fail = ( # noqa: S608 f"UPDATE {self._table}" f" SET state = 'failed', error = %s::jsonb, updated_at = %s" f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" - f" RETURNING task_id" + f" RETURNING task_id, account_id, task_type, webhook_registration," + f" webhook_registration_nonce" + ) + self._sql_clear_webhook_registration = ( # noqa: S608 + f"UPDATE {self._table} SET webhook_registration = NULL," + f" webhook_registration_nonce = NULL WHERE task_id = %s" ) # Explicit ``::text`` cast on the optional account-filter # parameter so psycopg's bind-param type inference doesn't @@ -159,12 +186,15 @@ def __init__(self, *, pool: AsyncConnectionPool, _table: str = _DEFAULT_TABLE) - # for the parameter and the query fails at prepare time. self._sql_get = ( # noqa: S608 f"SELECT task_id, account_id, state, task_type," - f" progress, result, error, created_at, updated_at" + f" progress, result, error, request_context, created_at, updated_at" f" FROM {self._table}" f" WHERE task_id = %s AND (%s::text IS NULL OR account_id = %s)" ) self._sql_get_state_result = ( # noqa: S608 - f"SELECT state, result FROM {self._table} WHERE task_id = %s" + f"SELECT state, result, task_type FROM {self._table} WHERE task_id = %s" + ) + self._sql_get_task_type = ( # noqa: S608 + f"SELECT task_type FROM {self._table} WHERE task_id = %s" ) self._sql_get_state_error = ( # noqa: S608 f"SELECT state, error FROM {self._table} WHERE task_id = %s" @@ -179,9 +209,15 @@ def __init__(self, *, pool: AsyncConnectionPool, _table: str = _DEFAULT_TABLE) - f" progress JSONB," f" result JSONB," f" error JSONB," + f" request_context JSONB," + f" webhook_registration BYTEA," + f" webhook_registration_nonce BYTEA," f" created_at DOUBLE PRECISION NOT NULL," f" updated_at DOUBLE PRECISION NOT NULL" f");" + f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS request_context JSONB;" + f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS webhook_registration BYTEA;" + f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS webhook_registration_nonce BYTEA;" f"CREATE INDEX IF NOT EXISTS {self._table}_account_idx" # noqa: S608 f" ON {self._table} (account_id);" ) @@ -207,6 +243,11 @@ async def issue( *, account_id: str, task_type: str, + request_context: dict[str, Any] | None = None, + webhook_url: str | None = None, + webhook_operation_id: str | None = None, + webhook_token: str | None = None, + **_extra: Any, ) -> str: """Allocate a task_id, persist a ``submitted`` row, return the id. @@ -222,10 +263,46 @@ async def issue( "return Account(id=) so cross-tenant cache " "scoping works correctly." ) + if (webhook_url is None) != (webhook_operation_id is None): + raise ValueError("webhook_url and webhook_operation_id must be supplied together") + if webhook_url is not None and not webhook_url: + 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") + outbox = self.task_webhook_outbox + if webhook_url is not None: + if outbox is None: + raise ValueError("webhook registration requires the registry's task_webhook_outbox") + outbox.validate_registration(webhook_url) task_id = f"task_{uuid.uuid4().hex[:16]}" + encrypted_registration: bytes | None = None + registration_nonce: bytes | None = None + if webhook_url is not None: + if webhook_operation_id is None or outbox is None: + raise RuntimeError("validated webhook registration became incomplete") + encrypted_registration, registration_nonce = outbox.protect_registration( + account_id=account_id, + task_id=task_id, + task_type=task_type, + url=webhook_url, + operation_id=webhook_operation_id, + token=webhook_token, + ) now = time.time() async with self._pool.connection() as conn: - await conn.execute(self._sql_insert, (task_id, account_id, task_type, now, now)) + await conn.execute( + self._sql_insert, + ( + task_id, + account_id, + task_type, + json.dumps(request_context) if request_context is not None else None, + encrypted_registration, + registration_nonce, + now, + now, + ), + ) return task_id async def update_progress( @@ -269,8 +346,25 @@ async def complete( race each other into double-completion without detection. """ async with self._pool.connection() as conn: - cur = await conn.execute(self._sql_complete, (json.dumps(result), time.time(), task_id)) - if await cur.fetchone() is not None: + type_cursor = await conn.execute(self._sql_get_task_type, (task_id,)) + type_row = await type_cursor.fetchone() + safe_result = ( + strip_credentials_from_wire_result(type_row[0], result) + if type_row is not None + else result + ) + cur = await conn.execute( + self._sql_complete, + (json.dumps(safe_result), time.time(), task_id), + ) + row = await cur.fetchone() + if row is not None: + await self._enqueue_terminal_if_registered( + conn, + row=row, + status="completed", + payload=safe_result, + ) return # updated successfully # Zero rows in RETURNING — task is unknown or already terminal. @@ -278,9 +372,10 @@ async def complete( row = await cur2.fetchone() if row is None: raise ValueError(f"Task {task_id!r} not found") - state, existing_result = row + state, existing_result, task_type = row + safe_result = strip_credentials_from_wire_result(task_type, result) if state == "completed": - if existing_result == result: + if existing_result == safe_result: return # idempotent raise ValueError(f"Task {task_id!r} already completed with a different result") raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") @@ -297,7 +392,14 @@ async def fail( """ async with self._pool.connection() as conn: cur = await conn.execute(self._sql_fail, (json.dumps(error), time.time(), task_id)) - if await cur.fetchone() is not None: + row = await cur.fetchone() + if row is not None: + await self._enqueue_terminal_if_registered( + conn, + row=row, + status="failed", + payload=error, + ) return # updated successfully # Zero rows in RETURNING — task is unknown or already terminal. @@ -340,10 +442,52 @@ async def get( "progress": row[4], "result": row[5], "error": row[6], - "created_at": row[7], - "updated_at": row[8], + "created_at": row[8], + "updated_at": row[9], + **({"context": row[7]} if row[7] is not None else {}), } + async def _enqueue_terminal_if_registered( + self, + conn: Any, + *, + row: tuple[Any, ...], + status: str, + payload: dict[str, Any], + ) -> None: + """Enqueue on ``conn`` so task state and webhook commit atomically.""" + task_id, account_id, task_type, encrypted_registration, registration_nonce = row + if encrypted_registration is None: + return + if self.task_webhook_outbox is None: + raise RuntimeError( + "Task carries push_notification_config but PgTaskRegistry has no " + "task_webhook_outbox; refusing a non-atomic terminal transition" + ) + 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), + ) + await self.task_webhook_outbox.enqueue_terminal( + conn, + task_id=task_id, + account_id=account_id, + task_type=task_type, + status=status, + result=payload, + url=url, + operation_id=operation_id, + token=token, + ) + # The encrypted outbox envelope now owns the callback registration. + # Clear the task-row copy in this same transaction. + await conn.execute(self._sql_clear_webhook_registration, (task_id,)) + async def discard(self, task_id: str) -> None: """Remove a task_id from the registry — rollback path. diff --git a/src/adcp/decisioning/pg/task_webhook_outbox.py b/src/adcp/decisioning/pg/task_webhook_outbox.py new file mode 100644 index 000000000..c30297f53 --- /dev/null +++ b/src/adcp/decisioning/pg/task_webhook_outbox.py @@ -0,0 +1,641 @@ +"""Crash-durable PostgreSQL outbox for terminal protocol-task webhooks. + +The outbox is intentionally coupled to :class:`PgTaskRegistry`: terminal +task state and the immutable webhook request are written on the same database +connection and commit together. Workers claim rows with expiring leases, +perform HTTP outside the database transaction, then acknowledge or release +the lease. A crash after receiver acceptance can cause an exact retry; the +stable body and idempotency key make that retry safe for conformant receivers. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import random +import re +import time +import uuid +from typing import TYPE_CHECKING, Any, ClassVar + +import httpx +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +from adcp.signing.jwks import SSRFValidationError +from adcp.webhook_sender import PreparedWebhook, WebhookDeliveryResult +from adcp.webhook_supervisor import RetryPolicy + +if TYPE_CHECKING: + from psycopg_pool import AsyncConnectionPool + + from adcp.webhook_sender import WebhookSender + +try: + from psycopg_pool import AsyncConnectionPool as _AsyncConnectionPool # noqa: F401 + + PG_AVAILABLE = True +except ImportError: + PG_AVAILABLE = False + +logger = logging.getLogger(__name__) +_INSTALL_HINT = ( + "PgTaskWebhookOutbox requires psycopg3 and psycopg-pool. " + "Install the 'pg' extra: `pip install 'adcp[pg]'` " + "(Poetry: `poetry add 'adcp[pg]'`)." +) +_SAFE_IDENTIFIER_RE = re.compile(r"^[a-z_][a-z0-9_]{0,44}$") +DEFAULT_TABLE = "adcp_task_webhook_outbox" +MIN_RETRY_HORIZON_SECONDS = 86_400 +MAX_RETRY_HORIZON_SECONDS = 604_800 + + +class PgTaskWebhookOutbox: + """Atomic task-webhook outbox and lease-based delivery worker. + + Construct this with the same pool as :class:`PgTaskRegistry`, then pass it + to ``PgTaskRegistry(..., task_webhook_outbox=outbox)``. Call + :meth:`create_schema` during migration/startup and run at least one + :meth:`run_worker` loop in every deployment. + """ + + delivery_state_is_durable: ClassVar[bool] = True + supports_atomic_task_outbox: ClassVar[bool] = True + + def __init__( + self, + *, + pool: AsyncConnectionPool, + sender: WebhookSender | None, + encryption_key: bytes, + delivery_retry_horizon_seconds: int, + retry: RetryPolicy | None = None, + lease_seconds: int = 60, + table: str = DEFAULT_TABLE, + ) -> None: + if not PG_AVAILABLE: + raise ImportError(_INSTALL_HINT) + if sender is None: + raise ValueError("PgTaskWebhookOutbox requires a non-None WebhookSender") + 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 type(delivery_retry_horizon_seconds) is not int or not ( + MIN_RETRY_HORIZON_SECONDS <= delivery_retry_horizon_seconds <= MAX_RETRY_HORIZON_SECONDS + ): + raise ValueError( + "delivery_retry_horizon_seconds must be an integer from " + f"{MIN_RETRY_HORIZON_SECONDS} through {MAX_RETRY_HORIZON_SECONDS}" + ) + 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: + raise ValueError( + "lease_seconds must exceed the sender HTTP timeout by at least 5 seconds" + ) + resolved_retry = retry or RetryPolicy() + if ( + resolved_retry.base_delay_seconds <= 0 + or resolved_retry.max_delay_seconds <= 0 + or resolved_retry.max_delay_seconds < resolved_retry.base_delay_seconds + ): + raise ValueError( + "retry delays must be positive and max_delay_seconds must be at least " + "base_delay_seconds" + ) + if not _SAFE_IDENTIFIER_RE.fullmatch(table): + raise ValueError( + f"table must match [a-z_][a-z0-9_]{{0,44}} (ASCII only), got {table!r}" + ) + + self._pool = pool + self._sender = sender + self._cipher = AESGCM(encryption_key) + self.delivery_retry_horizon_seconds = delivery_retry_horizon_seconds + self._retry = resolved_retry + self._lease_seconds = lease_seconds + self._table = table + self._worker_started = False + + 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, " + "retry_horizon_seconds" + ") VALUES (%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}" + " WHERE state IN ('pending', 'in_flight') AND retry_until <= now()" + " ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1000)" + f" UPDATE {table} AS outbox SET state = 'expired', lease_token = NULL," + " lease_expires_at = NULL, updated_at = now()" + " FROM expired WHERE outbox.id = expired.id" + ) + self._sql_claim = ( # noqa: S608 + f"WITH candidate AS (" + f" SELECT id FROM {table}" + " WHERE (retry_until IS NULL OR retry_until > now()) AND (" + " (state = 'pending' AND available_at <= now()) OR" + " (state = 'in_flight' AND lease_expires_at <= now())" + " ) ORDER BY available_at, id FOR UPDATE SKIP LOCKED LIMIT 1" + f") UPDATE {table} AS outbox SET" + " state = 'in_flight', lease_token = %s," + " lease_expires_at = now() + (%s * interval '1 second')," + " first_attempt_at = COALESCE(outbox.first_attempt_at, now())," + " retry_until = COALESCE(" + " outbox.retry_until, now() + (outbox.retry_horizon_seconds * interval '1 second')" + " )," + " attempt_count = outbox.attempt_count + 1, updated_at = now()" + " 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.attempt_count" + ) + self._sql_ack = ( # noqa: S608 + f"UPDATE {table} SET state = 'delivered', delivered_at = now()," + " lease_token = NULL, lease_expires_at = NULL," + " last_http_status = %s, last_error = NULL, updated_at = now()" + " WHERE id = %s AND state = 'in_flight' AND lease_token = %s" + ) + self._sql_release = ( # noqa: S608 + f"UPDATE {table} SET" + " state = CASE WHEN retry_until <= now() THEN 'expired' ELSE 'pending' END," + " available_at = CASE WHEN retry_until <= now() THEN available_at" + " ELSE now() + (%s * interval '1 second') END," + " lease_token = NULL, lease_expires_at = NULL," + " last_http_status = %s, last_error = %s, updated_at = now()" + " WHERE id = %s AND state = 'in_flight' AND lease_token = %s" + ) + self._sql_quarantine = ( # noqa: S608 + f"UPDATE {table} SET state = 'invalid', lease_token = NULL," + " lease_expires_at = NULL, last_error = %s, updated_at = now()" + " WHERE id = %s AND state = 'in_flight' AND lease_token = %s" + ) + self._sql_purge = ( # noqa: S608 + f"DELETE FROM {table} WHERE id IN (" + f" SELECT id FROM {table} WHERE retry_until <= now()" + " AND state IN ('delivered', 'expired', 'invalid')" + " ORDER BY id LIMIT 1000" + ")" + ) + + async def create_schema(self) -> None: + """Create the outbox table and work index idempotently.""" + statements = [ + f"""CREATE TABLE IF NOT EXISTS {self._table} ( + id BIGSERIAL PRIMARY KEY, + task_id TEXT COLLATE "C" NOT NULL UNIQUE, + account_id TEXT COLLATE "C" NOT NULL, + task_type TEXT NOT NULL, + terminal_status TEXT NOT NULL, + url TEXT NOT NULL, + operation_id TEXT NOT NULL, + idempotency_key TEXT COLLATE "C" NOT NULL UNIQUE, + encrypted_body BYTEA NOT NULL, + envelope_nonce BYTEA NOT NULL, + state TEXT NOT NULL DEFAULT 'pending', + attempt_count INTEGER NOT NULL DEFAULT 0, + available_at TIMESTAMPTZ NOT NULL DEFAULT now(), + lease_token TEXT COLLATE "C", + lease_expires_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + first_attempt_at TIMESTAMPTZ, + retry_until TIMESTAMPTZ, + retry_horizon_seconds INTEGER NOT NULL, + delivered_at TIMESTAMPTZ, + last_http_status INTEGER, + last_error TEXT, + CHECK (state IN ('pending', 'in_flight', 'delivered', 'expired', 'invalid')), + CHECK (terminal_status IN ('completed', 'failed')), + CHECK (attempt_count >= 0), + CHECK (octet_length(envelope_nonce) = 12), + CHECK (retry_horizon_seconds BETWEEN 86400 AND 604800), + CHECK ((first_attempt_at IS NULL) = (retry_until IS NULL)), + CHECK (retry_until IS NULL OR retry_until > first_attempt_at) + )""", + f"""CREATE INDEX IF NOT EXISTS {self._table}_work_idx + ON {self._table} (available_at, id) + WHERE state IN ('pending', 'in_flight')""", + f"""CREATE INDEX IF NOT EXISTS {self._table}_retry_until_idx + ON {self._table} (retry_until)""", + ] + async with self._pool.connection() as conn: + for statement in statements: + await conn.execute(statement) + + async def enqueue_terminal( + self, + conn: Any, + *, + task_id: str, + account_id: str, + task_type: str, + status: str, + result: dict[str, Any], + url: str, + operation_id: str, + token: str | 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( + url=url, + task_id=task_id, + task_type=task_type, + status=status, + result=result, + operation_id=operation_id, + token=token, + ) + self._validate_callback_url(prepared.url) + nonce = os.urandom(12) + aad = self._envelope_aad( + account_id=account_id, + task_id=task_id, + task_type=task_type, + status=status, + url=prepared.url, + operation_id=operation_id, + idempotency_key=prepared.idempotency_key, + ) + encrypted_body = self._cipher.encrypt(nonce, prepared.body, aad) + cursor = await conn.execute( + self._sql_insert, + ( + task_id, + task_type, + status, + prepared.url, + operation_id, + prepared.idempotency_key, + account_id, + encrypted_body, + nonce, + self.delivery_retry_horizon_seconds, + ), + ) + row = await cursor.fetchone() + if row is None: + raise RuntimeError("task webhook outbox insert returned no id") + return int(row[0]) + + def validate_registration(self, url: str) -> None: + """Validate callback syntax before a task is accepted as Submitted.""" + self._validate_callback_url(url) + + def protect_registration( + self, + *, + account_id: str, + task_id: str, + task_type: str, + url: str, + operation_id: str, + token: str | None, + ) -> tuple[bytes, bytes]: + """Encrypt and authenticate callback registration at task issue time.""" + self._validate_callback_url(url) + nonce = os.urandom(12) + plaintext = json.dumps( + {"url": url, "operation_id": operation_id, "token": token}, + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + return ( + self._cipher.encrypt( + nonce, + plaintext, + self._registration_aad( + account_id=account_id, + task_id=task_id, + task_type=task_type, + ), + ), + nonce, + ) + + def open_registration( + self, + *, + account_id: str, + task_id: str, + task_type: str, + encrypted_registration: bytes, + nonce: bytes, + ) -> tuple[str, str, str | None]: + """Verify and decrypt callback registration at terminal transition.""" + try: + plaintext = self._cipher.decrypt( + nonce, + encrypted_registration, + self._registration_aad( + account_id=account_id, + task_id=task_id, + task_type=task_type, + ), + ) + value = json.loads(plaintext) + except (InvalidTag, json.JSONDecodeError, UnicodeDecodeError) as exc: + raise ValueError("task webhook registration failed authenticated decryption") from exc + if not isinstance(value, dict): + raise ValueError("task webhook registration must decrypt to an object") + url = value.get("url") + operation_id = value.get("operation_id") + token = value.get("token") + 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") + self._validate_callback_url(url) + return url, operation_id, token + + async def run_worker( + self, + *, + poll_interval: float = 1.0, + purge_interval: float = 300.0, + ) -> None: + """Continuously publish eligible rows until the task is cancelled.""" + if poll_interval <= 0: + raise ValueError("poll_interval must be positive") + if purge_interval <= 0: + raise ValueError("purge_interval must be positive") + self._worker_started = True + next_purge = 0.0 + try: + while True: + try: + now = time.monotonic() + if now >= next_purge: + await self.purge_expired() + next_purge = now + purge_interval + processed = await self.process_one() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("[adcp.task_webhook_outbox] worker iteration failed; retrying") + await asyncio.sleep(poll_interval) + continue + if not processed: + await asyncio.sleep(poll_interval) + finally: + self._worker_started = False + + async def process_one(self) -> bool: + """Claim and attempt one delivery; return ``False`` when idle.""" + lease_token = uuid.uuid4().hex + async with self._pool.connection() as conn: + await conn.execute(self._sql_expire) + cursor = await conn.execute( + self._sql_claim, + (lease_token, self._lease_seconds), + ) + row = await cursor.fetchone() + if row is None: + return False + + ( + row_id, + account_id, + task_id, + task_type, + status, + url, + operation_id, + idempotency_key, + encrypted_body, + nonce, + attempt_count, + ) = row + aad = self._envelope_aad( + account_id=str(account_id), + task_id=str(task_id), + task_type=str(task_type), + status=str(status), + url=str(url), + operation_id=str(operation_id), + idempotency_key=str(idempotency_key), + ) + try: + body_bytes = self._cipher.decrypt(bytes(nonce), bytes(encrypted_body), aad) + self._validate_stored_body( + body_bytes, + task_id=str(task_id), + task_type=str(task_type), + status=str(status), + operation_id=str(operation_id), + idempotency_key=str(idempotency_key), + ) + except (InvalidTag, ValueError, json.JSONDecodeError, UnicodeDecodeError): + error_message = ( + "stored webhook envelope failed authenticated binding verification; " + "row quarantined without delivery" + ) + async with self._pool.connection() as conn: + await conn.execute( + self._sql_quarantine, + (error_message, row_id, lease_token), + ) + logger.error( + "[adcp.task_webhook_outbox] integrity failure for task %s; row quarantined", + task_id, + ) + return True + prepared = PreparedWebhook( + url=str(url), + idempotency_key=str(idempotency_key), + body=body_bytes, + ) + delivery: WebhookDeliveryResult | None = None + error: BaseException | None = None + try: + delivery = await asyncio.wait_for( + self._sender.send_prepared(prepared), + timeout=self._lease_seconds - 1, + ) + except SSRFValidationError as exc: + if exc.transient: + error = exc + else: + await self._quarantine_permanent_delivery_error( + row_id=row_id, + lease_token=lease_token, + task_id=str(task_id), + error=exc, + ) + return True + except ValueError as exc: + await self._quarantine_permanent_delivery_error( + row_id=row_id, + lease_token=lease_token, + task_id=str(task_id), + error=exc, + ) + return True + except Exception as exc: + error = exc + + if delivery is not None and delivery.ok: + async with self._pool.connection() as conn: + await conn.execute( + self._sql_ack, + (delivery.status_code, row_id, lease_token), + ) + return True + + if delivery is not None and not self._is_retryable_http_status(delivery.status_code): + error_message = ( + f"permanent HTTP {delivery.status_code}: {delivery.response_body[:200]!r}" + ) + async with self._pool.connection() as conn: + await conn.execute( + self._sql_quarantine, + (error_message[:1000], row_id, lease_token), + ) + logger.error( + "[adcp.task_webhook_outbox] permanent HTTP failure for task %s; " "row quarantined", + task_id, + ) + return True + + delay = self._retry_delay(int(attempt_count)) + http_status = delivery.status_code if delivery is not None else None + if delivery is not None: + error_message = f"HTTP {delivery.status_code}: {delivery.response_body[:200]!r}" + elif error is not None: + error_message = f"{type(error).__name__}: {error}" + else: + error_message = "delivery failed without a result" + async with self._pool.connection() as conn: + await conn.execute( + self._sql_release, + (delay, http_status, error_message, row_id, lease_token), + ) + logger.warning( + "[adcp.task_webhook_outbox] delivery failed for task %s; retry in %.1fs", + task_id, + delay, + ) + return True + + async def _quarantine_permanent_delivery_error( + self, + *, + row_id: int, + lease_token: str, + task_id: str, + error: BaseException, + ) -> None: + error_message = f"permanent delivery validation failure: {type(error).__name__}: {error}" + async with self._pool.connection() as conn: + await conn.execute( + self._sql_quarantine, + (error_message[:1000], row_id, lease_token), + ) + logger.error( + "[adcp.task_webhook_outbox] permanent delivery failure for task %s; " "row quarantined", + task_id, + ) + + @staticmethod + def _is_retryable_http_status(status_code: int) -> bool: + return status_code >= 500 or status_code in {408, 425, 429} + + async def purge_expired(self) -> None: + """Delete delivery proof only after the advertised horizon elapses.""" + async with self._pool.connection() as conn: + await conn.execute(self._sql_expire) + await conn.execute(self._sql_purge) + + def _retry_delay(self, attempt_count: int) -> float: + exponent = max(0, min(attempt_count - 1, 30)) + delay: float = float( + min( + self._retry.base_delay_seconds * (2**exponent), + self._retry.max_delay_seconds, + ) + ) + if self._retry.jitter: + delay *= 0.5 + random.random() * 0.5 + return delay + + @staticmethod + def _envelope_aad( + *, + account_id: str, + task_id: str, + task_type: str, + status: str, + url: str, + operation_id: str, + idempotency_key: str, + ) -> bytes: + """Canonical associated data binding every routing/security field.""" + return json.dumps( + [account_id, task_id, task_type, status, url, operation_id, idempotency_key], + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + + @staticmethod + def _registration_aad(*, account_id: str, task_id: str, task_type: str) -> bytes: + return json.dumps( + ["task-webhook-registration-v1", account_id, task_id, task_type], + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + + @staticmethod + def _validate_callback_url(url: str) -> None: + if len(url) > 2048: + raise ValueError("webhook URL must not exceed 2048 characters") + parsed = httpx.URL(url) + if parsed.scheme != "https" or not parsed.host: + raise ValueError("webhook URL must be an absolute HTTPS URL") + if parsed.username or parsed.password: + raise ValueError("webhook URL must not contain userinfo") + + @staticmethod + def _validate_stored_body( + body: bytes, + *, + task_id: str, + task_type: str, + status: str, + operation_id: str, + idempotency_key: str, + ) -> None: + payload = json.loads(body) + if not isinstance(payload, dict): + raise ValueError("stored webhook body must be a JSON object") + expected = { + "task_id": task_id, + "task_type": task_type, + "status": status, + "operation_id": operation_id, + "idempotency_key": idempotency_key, + } + if any(payload.get(key) != value for key, value in expected.items()): + raise ValueError("stored webhook body does not match its envelope metadata") + + +__all__ = [ + "DEFAULT_TABLE", + "MAX_RETRY_HORIZON_SECONDS", + "MIN_RETRY_HORIZON_SECONDS", + "PG_AVAILABLE", + "PgTaskWebhookOutbox", +] diff --git a/src/adcp/decisioning/pg/task_webhook_outbox.sql b/src/adcp/decisioning/pg/task_webhook_outbox.sql new file mode 100644 index 000000000..511ff6c85 --- /dev/null +++ b/src/adcp/decisioning/pg/task_webhook_outbox.sql @@ -0,0 +1,47 @@ +-- AdCP terminal task-webhook outbox. +-- +-- Terminal task state and an outbox row must be committed in one transaction. +-- PgTaskRegistry performs that write when configured with +-- PgTaskWebhookOutbox. The body is encrypted and authenticated with +-- application-held AES-256-GCM key material; retry_until begins at the first +-- attempt and preserves the immutable binding after successful delivery. + +CREATE TABLE IF NOT EXISTS adcp_task_webhook_outbox ( + id BIGSERIAL PRIMARY KEY, + task_id TEXT COLLATE "C" NOT NULL UNIQUE, + account_id TEXT COLLATE "C" NOT NULL, + task_type TEXT NOT NULL, + terminal_status TEXT NOT NULL, + url TEXT NOT NULL, + operation_id TEXT NOT NULL, + idempotency_key TEXT COLLATE "C" NOT NULL UNIQUE, + encrypted_body BYTEA NOT NULL, + envelope_nonce BYTEA NOT NULL, + state TEXT NOT NULL DEFAULT 'pending', + attempt_count INTEGER NOT NULL DEFAULT 0, + available_at TIMESTAMPTZ NOT NULL DEFAULT now(), + lease_token TEXT COLLATE "C", + lease_expires_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + first_attempt_at TIMESTAMPTZ, + retry_until TIMESTAMPTZ, + retry_horizon_seconds INTEGER NOT NULL, + delivered_at TIMESTAMPTZ, + last_http_status INTEGER, + last_error TEXT, + CHECK (state IN ('pending', 'in_flight', 'delivered', 'expired', 'invalid')), + CHECK (terminal_status IN ('completed', 'failed')), + CHECK (attempt_count >= 0), + CHECK (octet_length(envelope_nonce) = 12), + CHECK (retry_horizon_seconds BETWEEN 86400 AND 604800), + CHECK ((first_attempt_at IS NULL) = (retry_until IS NULL)), + CHECK (retry_until IS NULL OR retry_until > first_attempt_at) +); + +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'); + +CREATE INDEX IF NOT EXISTS adcp_task_webhook_outbox_retry_until_idx + ON adcp_task_webhook_outbox (retry_until); diff --git a/src/adcp/decisioning/serve.py b/src/adcp/decisioning/serve.py index e087a6851..770920b17 100644 --- a/src/adcp/decisioning/serve.py +++ b/src/adcp/decisioning/serve.py @@ -161,8 +161,8 @@ def create_adcp_server_from_platform( :class:`~adcp.webhook_supervisor.WebhookDeliverySupervisor` for explicit/manual retry orchestration. The reference supervisors do not provide the atomic terminal-state/outbox commit required for beta.5 - TaskHandoff push publication. Production publishers own that lifecycle - externally and leave SDK delivery targets unwired. + TaskHandoff push publication. Use ``PgTaskWebhookOutbox`` through the + configured ``PgTaskRegistry`` for SDK-managed publication. :param buyer_agent_registry: BYO :class:`adcp.decisioning.BuyerAgentRegistry` — the v3 commercial identity layer. When wired, the framework calls the registry @@ -186,11 +186,10 @@ def create_adcp_server_from_platform( forbids a task webhook when the initial response is already terminal. Passing ``True`` emits a deprecation warning. :param auto_emit_task_webhooks: Framework ownership of terminal - webhooks for real ``TaskHandoff`` requests. The SDK does not yet - provide the atomic terminal-state/outbox contract required by AdCP - 3.2, so the default ``True`` rejects push-configured handoffs before - task creation. Set ``False`` only when an external durable outbox - owns required task-webhook delivery and capability advertisement. + webhooks for real ``TaskHandoff`` requests. The default ``True`` + admits push-configured handoffs when ``PgTaskRegistry`` carries a + conformant ``PgTaskWebhookOutbox``. Set ``False`` only when an + external durable outbox owns delivery and capability advertisement. :param media_buy_store: Opt-in :class:`adcp.decisioning.MediaBuyStore` wrapper that gates ``targeting_overlay`` echo on the seller's declared specialisms. Typically built via @@ -432,6 +431,7 @@ def create_adcp_server_from_platform( sender=webhook_sender, supervisor=webhook_supervisor, auto_emit_task_webhooks=auto_emit_task_webhooks, + registry=registry, ) # DX #422: boot-time fail-fast on a non-conformant capabilities @@ -529,9 +529,9 @@ def serve( silent on the task-webhook channel. :param auto_emit_task_webhooks: Framework ownership of required terminal webhooks for real ``TaskHandoff`` requests. Defaults to - ``True``. The SDK rejects push-configured handoffs in this mode. Set - ``False`` only when an external durable outbox owns that delivery; - this is independent of the legacy sync-completion flag. + ``True``. Push-configured handoffs require a registry-backed atomic + task outbox. Set ``False`` only when an external durable outbox owns + that delivery; this is independent of the legacy sync-completion flag. :param mock_ad_server: Optional :class:`adcp.decisioning.MockAdServer` whose ``get_traffic()`` is wired into ``GET /_debug/traffic`` when ``enable_debug_endpoints=True``. Default ``None`` — diff --git a/src/adcp/decisioning/task_registry.py b/src/adcp/decisioning/task_registry.py index 3330dd1a1..ca3a16f26 100644 --- a/src/adcp/decisioning/task_registry.py +++ b/src/adcp/decisioning/task_registry.py @@ -216,6 +216,9 @@ async def issue( account_id: str, task_type: str, request_context: dict[str, Any] | None = None, + webhook_url: str | None = None, + webhook_operation_id: str | None = None, + webhook_token: str | None = None, **_extra: Any, ) -> str: """Allocate a fresh task_id, persist a ``submitted`` row, and @@ -238,6 +241,13 @@ async def issue( store and surface this field; older registry impls that ignore it are functionally compatible (no echo on ``tasks/get`` reads, identical to pre-#563 behavior). + :param webhook_url: Buyer-registered terminal task callback URL. + Durable registries with an atomic outbox persist it at issue time + and consume it in ``complete`` / ``fail``. ``None`` means polling. + :param webhook_operation_id: Buyer-generated operation identifier to + 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 _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 e914ca4f9..f840b8fab 100644 --- a/src/adcp/decisioning/webhook_emit.py +++ b/src/adcp/decisioning/webhook_emit.py @@ -32,6 +32,23 @@ logger = logging.getLogger(__name__) +def _sdk_task_outbox_pair_ready(registry: Any, task_outbox: Any) -> bool: + """Accept only the concrete registry/outbox pair whose atomicity we own.""" + if registry is None or task_outbox is None: + return False + try: + from adcp.decisioning.pg.task_registry import PgTaskRegistry + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + except ImportError: + return False + return ( + isinstance(registry, PgTaskRegistry) + and isinstance(task_outbox, PgTaskWebhookOutbox) + and registry.task_webhook_outbox is task_outbox + and registry._pool is task_outbox._pool + ) + + #: Tools eligible for asynchronous task webhooks. Mirrors the closed enum in #: ``schemas/cache/enums/task-type.json`` verbatim. The framework dispatches a #: wider tool surface than this set; the JS side maintains the same set at @@ -331,6 +348,7 @@ def validate_webhook_signing_for_capabilities( sender: WebhookSender | None, supervisor: WebhookDeliverySupervisor | None = None, auto_emit_task_webhooks: bool = True, + registry: Any = None, ) -> None: """Server-boot fail-fast for the #384 capabilities-vs-wiring invariant. @@ -357,11 +375,10 @@ def validate_webhook_signing_for_capabilities( delivery-method axis is a poor gate. ``webhook_signing.supported`` is the self-consistency contract the spec supports directly. - SDK senders and supervisors may still be inspected to provide a precise - configuration error, but none currently supplies the atomic outbox needed - to back the beta.5 horizon. Conformant publishers therefore declare - external ownership, disable SDK automatic emission, and leave the SDK - delivery target unwired. + SDK senders and supervisors are inspected to provide precise diagnostics. + Framework-managed publication additionally requires a + ``PgTaskWebhookOutbox`` attached to the task registry; external publishers + declare external ownership and disable SDK automatic emission. :raises AdcpError: ``code='INVALID_REQUEST'`` when capabilities declare RFC 9421 signing support but no sender (or a non-JWK @@ -370,6 +387,7 @@ def validate_webhook_signing_for_capabilities( boot-time validators (terminal). """ adopter_managed = getattr(capabilities, "webhook_signing_managed_externally", False) + task_outbox = getattr(registry, "task_webhook_outbox", None) from adcp.decisioning.types import AdcpError @@ -422,6 +440,16 @@ def validate_webhook_signing_for_capabilities( ) if adopter_managed is True: + if task_outbox is not None: + raise AdcpError( + "INVALID_REQUEST", + message=( + "webhook_signing_managed_externally=True conflicts with the " + "PgTaskRegistry task_webhook_outbox; choose exactly one owner" + ), + recovery="terminal", + details={"missing": "single_task_webhook_owner"}, + ) if auto_emit_task_webhooks: raise AdcpError( "INVALID_REQUEST", @@ -452,9 +480,66 @@ def validate_webhook_signing_for_capabilities( ) return - resolved_sender: Any = sender + internal_outbox_ready = _sdk_task_outbox_pair_ready(registry, task_outbox) + if task_outbox is not None and not internal_outbox_ready: + raise AdcpError( + "INVALID_REQUEST", + message=( + "SDK-managed task webhook publication requires the concrete " + "PgTaskRegistry/PgTaskWebhookOutbox pair using the same pool; " + "custom publishers must use webhook_signing_managed_externally=True" + ), + recovery="terminal", + details={"missing": "verified_sdk_task_webhook_outbox_pair"}, + ) + if internal_outbox_ready: + if sender is not None or supervisor is not None: + raise AdcpError( + "INVALID_REQUEST", + message=( + "A registry-backed task_webhook_outbox is the sole SDK " + "delivery owner; do not also wire webhook_sender or " + "webhook_supervisor into the handler" + ), + recovery="terminal", + details={"missing": "single_sdk_task_webhook_owner"}, + ) + if not auto_emit_task_webhooks: + raise AdcpError( + "INVALID_REQUEST", + message=( + "A registry-backed task_webhook_outbox requires " + "auto_emit_task_webhooks=True; False declares external ownership" + ), + recovery="terminal", + details={"missing": "sdk_task_webhook_ownership"}, + ) + outbox_horizon = getattr(task_outbox, "delivery_retry_horizon_seconds", None) + if ( + getattr(task_outbox, "supports_atomic_task_outbox", False) is not True + or getattr(task_outbox, "delivery_state_is_durable", False) is not True + or type(outbox_horizon) is not int + or outbox_horizon != retry_horizon + ): + raise AdcpError( + "INVALID_REQUEST", + message=( + "The registry task_webhook_outbox does not prove atomic durable " + "publication for the advertised retry horizon" + ), + recovery="terminal", + details={ + "missing": "atomic_durable_webhook_outbox", + "advertised_horizon_seconds": retry_horizon, + "outbox_horizon_seconds": outbox_horizon, + }, + ) + + resolved_sender: Any = ( + getattr(task_outbox, "_sender", None) if internal_outbox_ready else sender + ) sender_introspectable = True - if resolved_sender is None and supervisor is not 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 # queue-only adopters) may not. Their supervisor must still expose the @@ -547,14 +632,17 @@ def validate_webhook_signing_for_capabilities( }, ) + if internal_outbox_ready: + return + raise AdcpError( "INVALID_REQUEST", message=( - "The Python SDK does not provide an atomic terminal-state/outbox " - "publisher for the AdCP 3.2 beta.5 retry-horizon contract. Set " - "webhook_signing_managed_externally=True, leave the SDK delivery " - "target unwired, and set auto_emit_task_webhooks=False only when " - "adopter infrastructure owns atomic publication and reconciliation." + "No atomic terminal-state/outbox publisher is configured for the " + "AdCP 3.2 retry-horizon contract. Attach PgTaskWebhookOutbox to " + "PgTaskRegistry, or set webhook_signing_managed_externally=True " + "with auto_emit_task_webhooks=False when adopter infrastructure " + "owns atomic publication and reconciliation." ), recovery="terminal", details={"missing": "external_durable_webhook_outbox"}, @@ -588,10 +676,51 @@ def external_task_webhook_owner_ready( ) +def task_webhook_owner_ready( + *, + capabilities: DecisioningCapabilities, + sender: WebhookSender | None, + supervisor: WebhookDeliverySupervisor | None, + auto_emit_task_webhooks: bool, + registry: Any = None, +) -> bool: + """Return whether either the SDK atomic outbox or an external owner is ready.""" + if external_task_webhook_owner_ready( + capabilities=capabilities, + sender=sender, + supervisor=supervisor, + auto_emit_task_webhooks=auto_emit_task_webhooks, + ): + return True + + if getattr(capabilities, "webhook_signing_managed_externally", False) is not False: + return False + webhook_signing = getattr(capabilities, "webhook_signing", None) + advertised_horizon = getattr(webhook_signing, "delivery_retry_horizon_seconds", None) + outbox = getattr(registry, "task_webhook_outbox", None) + outbox_horizon = getattr(outbox, "delivery_retry_horizon_seconds", None) + outbox_sender = getattr(outbox, "_sender", None) + return ( + auto_emit_task_webhooks is True + and sender is None + and supervisor is None + and _sdk_task_outbox_pair_ready(registry, outbox) + and webhook_signing is not None + and getattr(webhook_signing, "supported", False) is True + and type(advertised_horizon) is int + and getattr(outbox, "supports_atomic_task_outbox", False) is True + 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 + ) + + __all__ = [ "SPEC_WEBHOOK_TASK_TYPES", "emit_terminal_completion_webhook", "external_task_webhook_owner_ready", + "task_webhook_owner_ready", "maybe_emit_sync_completion", "validate_webhook_sender_for_platform", "validate_webhook_signing_for_capabilities", diff --git a/src/adcp/signing/jwks.py b/src/adcp/signing/jwks.py index 4f42f4a66..7643079c6 100644 --- a/src/adcp/signing/jwks.py +++ b/src/adcp/signing/jwks.py @@ -126,6 +126,10 @@ class SSRFValidationError(Exception): """Raised when a URL resolves to an IP in a reserved or blocked range.""" + def __init__(self, message: str, *, transient: bool = False) -> None: + super().__init__(message) + self.transient = transient + class JwksFetcher(Protocol): """A callable that fetches and parses a JWKS document from a URL.""" @@ -297,7 +301,7 @@ def resolve_and_validate_host( try: infos = socket.getaddrinfo(host, None) except OSError as exc: - raise SSRFValidationError(f"cannot resolve host {host!r}: {exc}") from exc + raise SSRFValidationError(f"cannot resolve host {host!r}: {exc}", transient=True) from exc accepted_ip: str | None = None last_rejection: str | None = None diff --git a/src/adcp/webhook_sender.py b/src/adcp/webhook_sender.py index c6e6c9e15..47f1ad180 100644 --- a/src/adcp/webhook_sender.py +++ b/src/adcp/webhook_sender.py @@ -91,6 +91,9 @@ # an adversarial payload: json.dumps holds dict + str concurrently, and # .encode() transiently triples memory, so a 1GB body is multiple GB RSS. _MAX_BODY_BYTES = 10 * 1024 * 1024 +# Receiver error pages are untrusted. Webhook delivery only needs a bounded +# diagnostic prefix, never an arbitrarily large buffered response. +_MAX_RESPONSE_BODY_BYTES = 64 * 1024 _legacy_hmac_warned = False @@ -149,6 +152,37 @@ def _enum_value(value: Any) -> str: return str(raw) +async def _post_with_bounded_response( + client: httpx.AsyncClient, + url: str, + *, + body: bytes, + headers: Mapping[str, str], +) -> tuple[int, Mapping[str, str], bytes]: + """POST while retaining at most a small diagnostic response prefix.""" + captured = bytearray() + async with client.stream("POST", url, content=body, headers=headers) as response: + if response.is_stream_consumed: + # Some in-process/custom transports construct a pre-buffered + # response even for stream(). The bytes already exist, so retain + # only the bounded diagnostic prefix. + return ( + response.status_code, + dict(response.headers), + response.content[:_MAX_RESPONSE_BODY_BYTES], + ) + # Read raw wire bytes so httpx cannot inflate a compressed response + # into a large decoded chunk before this cap is applied. + async for chunk in response.aiter_raw(chunk_size=16 * 1024): + remaining = _MAX_RESPONSE_BODY_BYTES - len(captured) + if remaining <= 0: + break + captured.extend(chunk[:remaining]) + if len(chunk) > remaining: + break + return response.status_code, dict(response.headers), bytes(captured) + + @dataclass(frozen=True) class WebhookDeliveryResult: """Outcome of one ``send_*`` call. @@ -178,6 +212,21 @@ def ok(self) -> bool: return 200 <= self.status_code < 300 +@dataclass(frozen=True) +class PreparedWebhook: + """Immutable webhook request prepared for durable outbox storage. + + The serialized body and idempotency key are bound once, before the + outbox transaction commits. Workers may then replay this object under a + fresh RFC 9421 signature without regenerating timestamps or JSON bytes. + """ + + url: str + idempotency_key: str + body: bytes + extra_headers: Mapping[str, str] = field(default_factory=dict) + + class WebhookSender: """Outbound signed-webhook delivery client. @@ -590,6 +639,49 @@ async def send_mcp( back in webhook payload to validate request authenticity"). Cross-language wire-parity with the JS implementation. """ + return await self.send_prepared( + self.prepare_mcp( + url=url, + task_id=task_id, + status=status, + task_type=task_type, + result=result, + timestamp=timestamp, + operation_id=operation_id, + notification_id=notification_id, + message=message, + context_id=context_id, + protocol=protocol, + idempotency_key=idempotency_key, + token=token, + extra_headers=extra_headers, + ) + ) + + def prepare_mcp( + self, + *, + url: str, + task_id: str, + status: GeneratedTaskStatus | str, + task_type: TaskType | str, + result: AdcpAsyncResponseData | dict[str, Any] | None = None, + timestamp: datetime | None = None, + operation_id: str, + notification_id: str | None = None, + message: str | None = None, + context_id: str | None = None, + protocol: AdcpProtocol | str | None = None, + idempotency_key: str | None = None, + token: str | None = None, + extra_headers: Mapping[str, str] | None = None, + ) -> PreparedWebhook: + """Prepare an MCP webhook without performing network I/O. + + Durable publishers call this inside (or immediately before) their + outbox transaction and persist all returned fields verbatim. The + returned body is exactly what :meth:`send_prepared` signs and posts. + """ payload = create_mcp_webhook_payload( task_id=task_id, status=status, @@ -604,11 +696,50 @@ async def send_mcp( idempotency_key=idempotency_key, token=token, ) - return await self.send_raw( + body_dict = { + **to_wire_dict(payload), + "idempotency_key": payload.idempotency_key, + } + body = json.dumps(body_dict).encode("utf-8") + if len(body) > _MAX_BODY_BYTES: + raise ValueError( + f"serialized webhook body is {len(body):,} bytes, over the " + f"{_MAX_BODY_BYTES:,}-byte cap. Split into smaller webhooks " + "or use batch-reporting endpoints." + ) + return PreparedWebhook( url=url, idempotency_key=payload.idempotency_key, - payload=to_wire_dict(payload), - extra_headers=extra_headers, + body=body, + extra_headers=dict(extra_headers) if extra_headers else {}, + ) + + async def send_prepared(self, prepared: PreparedWebhook) -> WebhookDeliveryResult: + """Sign and post a previously prepared immutable webhook request.""" + if not prepared.idempotency_key: + raise ValueError("prepared webhook idempotency_key must be non-empty") + if not prepared.body: + raise ValueError("prepared webhook body must be non-empty") + if len(prepared.body) > _MAX_BODY_BYTES: + raise ValueError( + f"serialized webhook body is {len(prepared.body):,} bytes, over the " + f"{_MAX_BODY_BYTES:,}-byte cap" + ) + try: + payload = json.loads(prepared.body) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise ValueError("prepared webhook body must be a JSON object") from exc + if not isinstance(payload, dict): + raise ValueError("prepared webhook body must be a JSON object") + if payload.get("idempotency_key") != prepared.idempotency_key: + raise ValueError( + "prepared webhook body idempotency_key does not match its immutable binding" + ) + return await self._send_bytes( + url=prepared.url, + body=prepared.body, + idempotency_key=prepared.idempotency_key, + extra_headers=prepared.extra_headers or None, ) async def send_revocation_notification( @@ -989,7 +1120,7 @@ async def _send_bytes( allowed_ports=self._allowed_destination_ports, ) - base_headers = {"Content-Type": "application/json"} + base_headers = {"Content-Type": "application/json", "Accept-Encoding": "identity"} auth_headers = self._auth.build_auth_headers(method="POST", url=effective_url, body=body) headers = merge_extra_headers( base={**base_headers, **auth_headers}, @@ -1011,7 +1142,12 @@ async def _send_bytes( follow_redirects=False, trust_env=False, ) as client: - response = await client.post(effective_url, content=body, headers=headers) + status_code, response_headers, response_body = await _post_with_bounded_response( + client, + effective_url, + body=body, + headers=headers, + ) else: # Operator-supplied client — they own the SSRF guarantees on # their transport (proxy allowlist, mTLS, etc.). Reachable as @@ -1022,14 +1158,19 @@ async def _send_bytes( "WebhookSender's operator-supplied client was already " "closed. Construct a new sender or pass a fresh client." ) - response = await self._client.post(effective_url, content=body, headers=headers) + status_code, response_headers, response_body = await _post_with_bounded_response( + self._client, + effective_url, + body=body, + headers=headers, + ) return WebhookDeliveryResult( - status_code=response.status_code, + status_code=status_code, idempotency_key=idempotency_key, url=effective_url, - response_headers=dict(response.headers), - response_body=response.content, + response_headers=response_headers, + response_body=response_body, sent_body=body, sent_extra_headers=dict(extra_headers) if extra_headers else {}, ) @@ -1047,6 +1188,7 @@ async def _send_bytes( __all__ = [ "DockerLocalhostRewrite", + "PreparedWebhook", "TransportHook", "WebhookDeliveryResult", "WebhookSender", diff --git a/src/adcp/webhooks.py b/src/adcp/webhooks.py index 79895901d..35c8bd24a 100644 --- a/src/adcp/webhooks.py +++ b/src/adcp/webhooks.py @@ -2005,6 +2005,7 @@ def _validate_header_value(name: str, value: Any) -> None: # names before they're bound. This is the canonical Python pattern for breaking # such cycles without a third helper module. from adcp.webhook_sender import ( # noqa: E402 + PreparedWebhook, WebhookDeliveryResult, WebhookSender, ) @@ -2030,6 +2031,7 @@ def _validate_header_value(name: str, value: Any) -> None: "sign_webhook", # Sender — one-call outbound helpers "deliver", + "PreparedWebhook", "WebhookDeliveryResult", "WebhookSender", "WebhookDestinationPolicy", diff --git a/tests/conformance/decisioning/test_pg_task_webhook_outbox.py b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py new file mode 100644 index 000000000..82e424b52 --- /dev/null +++ b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py @@ -0,0 +1,202 @@ +"""Real-PostgreSQL conformance tests for the terminal task webhook outbox. + +Set ``ADCP_PG_TEST_URL`` to enable this module. Each test uses isolated table +names so parallel jobs cannot share leases or terminal task state. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import secrets +import uuid +from collections.abc import AsyncIterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +psycopg = pytest.importorskip("psycopg") +psycopg_pool = pytest.importorskip("psycopg_pool") + +TEST_URL = os.environ.get("ADCP_PG_TEST_URL") +if not TEST_URL: + pytest.skip( + "ADCP_PG_TEST_URL not set — skipping task webhook outbox conformance tests", + allow_module_level=True, + ) + +from adcp.decisioning.pg import PgTaskRegistry, PgTaskWebhookOutbox # noqa: E402 +from adcp.webhook_sender import PreparedWebhook, WebhookDeliveryResult # noqa: E402 + + +def _sender() -> MagicMock: + sender = MagicMock() + sender._owns_client = True + sender._allow_private_destinations = False + sender._timeout = 10.0 + sender.signs_with_rfc9421 = True + + def prepare_mcp(**kwargs: Any) -> PreparedWebhook: + key = f"whk_{uuid.uuid4().hex}" + body = json.dumps( + { + "idempotency_key": key, + "task_id": kwargs["task_id"], + "task_type": kwargs["task_type"], + "status": kwargs["status"], + "operation_id": kwargs["operation_id"], + "result": kwargs["result"], + "token": kwargs["token"], + } + ).encode() + return PreparedWebhook(url=kwargs["url"], idempotency_key=key, body=body) + + sender.prepare_mcp.side_effect = prepare_mcp + sender.send_prepared = AsyncMock( + side_effect=lambda prepared: WebhookDeliveryResult( + status_code=200, + idempotency_key=prepared.idempotency_key, + url=prepared.url, + response_headers={}, + response_body=b"{}", + sent_body=prepared.body, + ) + ) + return sender + + +@pytest.fixture() +async def stack() -> AsyncIterator[tuple[Any, PgTaskRegistry, PgTaskWebhookOutbox, MagicMock]]: + 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() + sender = _sender() + outbox = PgTaskWebhookOutbox( + pool=pool, + sender=sender, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + table=outbox_table, + ) + registry = PgTaskRegistry( + pool=pool, + task_webhook_outbox=outbox, + _table=task_table, + ) + await registry.create_schema() + await outbox.create_schema() + try: + yield pool, registry, outbox, sender + 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_terminal_state_and_encrypted_outbox_commit_together(stack) -> None: + pool, registry, outbox, _sender_mock = stack + task_id = await registry.issue( + account_id="acct_1", + task_type="create_media_buy", + webhook_url="https://buyer.example/webhook", + webhook_operation_id="op_1", + webhook_token="buyer-secret", + ) + await registry.complete(task_id, {"media_buy_id": "mb_1"}) + + async with pool.connection() as conn: + task_row = await ( + await conn.execute( + f"SELECT state, webhook_registration, webhook_registration_nonce " # noqa: S608 + f"FROM {registry._table} WHERE task_id = %s", + (task_id,), + ) + ).fetchone() + outbox_row = await ( + await conn.execute( + f"SELECT encrypted_body, first_attempt_at, retry_until " # noqa: S608 + f"FROM {outbox._table} WHERE task_id = %s", + (task_id,), + ) + ).fetchone() + 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 await outbox.process_one() is True + async with pool.connection() as conn: + delivered = await ( + await conn.execute( + f"SELECT state, first_attempt_at, retry_until " # noqa: S608 + f"FROM {outbox._table} WHERE task_id = %s", + (task_id,), + ) + ).fetchone() + assert delivered is not None + assert delivered[0] == "delivered" + assert delivered[1] is not None + assert delivered[2] is not None + + +@pytest.mark.asyncio +async def test_concurrent_claims_deliver_one_attempt(stack) -> None: + _pool, registry, outbox, sender = stack + task_id = await registry.issue( + account_id="acct_1", + task_type="create_media_buy", + webhook_url="https://buyer.example/webhook", + webhook_operation_id="op_1", + ) + await registry.complete(task_id, {"media_buy_id": "mb_1"}) + + processed = await asyncio.gather(outbox.process_one(), outbox.process_one()) + assert sorted(processed) == [False, True] + sender.send_prepared.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_missing_outbox_table_rolls_back_terminal_transition() -> 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, open=False) as pool: + await pool.open() + outbox = PgTaskWebhookOutbox( + pool=pool, + sender=_sender(), + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + table=outbox_table, + ) + registry = PgTaskRegistry( + pool=pool, + task_webhook_outbox=outbox, + _table=task_table, + ) + await registry.create_schema() + try: + task_id = await registry.issue( + account_id="acct_1", + task_type="create_media_buy", + webhook_url="https://buyer.example/webhook", + webhook_operation_id="op_1", + ) + with pytest.raises(psycopg.errors.UndefinedTable): + await registry.complete(task_id, {"media_buy_id": "mb_1"}) + record = await registry.get(task_id) + assert record is not None + assert record["state"] == "submitted" + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index adb3f79e3..2db60eab3 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -342,6 +342,7 @@ "PreviewRendererMetadata", "PriceGuidance", "PricingCurrency", + "PreparedWebhook", "PricingModel", "PricingOption", "Product", diff --git a/tests/test_decisioning_webhook_emit.py b/tests/test_decisioning_webhook_emit.py index e61a564cd..9cf7941f2 100644 --- a/tests/test_decisioning_webhook_emit.py +++ b/tests/test_decisioning_webhook_emit.py @@ -671,6 +671,57 @@ class _ExternalHandoffPlatform(_HandoffPlatform): sender.send_mcp.assert_not_called() +@pytest.mark.asyncio +async def test_handler_rejects_self_asserted_registry_outbox(executor) -> None: + """Marker attributes cannot impersonate the SDK's atomic PostgreSQL pair.""" + + class _OutboxSender: + signs_with_rfc9421 = True + + class _AtomicOutbox: + delivery_state_is_durable = True + supports_atomic_task_outbox = True + delivery_retry_horizon_seconds = 86_400 + _sender = _OutboxSender() + + class _AtomicRegistry(InMemoryTaskRegistry): + task_webhook_outbox = _AtomicOutbox() + + def __init__(self) -> None: + super().__init__() + self.issue_extra: dict[str, Any] = {} + + async def issue(self, *, account_id, task_type, request_context=None, **extra): + self.issue_extra = extra + return await super().issue( + account_id=account_id, + task_type=task_type, + request_context=request_context, + ) + + class _SdkOutboxPlatform(_HandoffPlatform): + capabilities = DecisioningCapabilities( + specialisms=["sales-non-guaranteed"], + webhook_signing=WebhookSigning( + supported=True, + delivery_retry_horizon_seconds=86_400, + ), + ) + + registry = _AtomicRegistry() + handler = PlatformHandler( + _SdkOutboxPlatform(), + executor=executor, + registry=registry, + ) + + with pytest.raises(AdcpError) as exc_info: + await handler.create_media_buy(_make_request(with_url=True), ToolContext()) + + assert exc_info.value.code == "INVALID_REQUEST" + assert registry.issue_extra == {} + + @pytest.mark.asyncio async def test_scoped_capabilities_can_admit_external_push_owner(executor) -> None: """Push admission uses the tenant-scoped capability set, not static defaults.""" diff --git a/tests/test_decisioning_workflow_handoff.py b/tests/test_decisioning_workflow_handoff.py index 178230811..84cc3078d 100644 --- a/tests/test_decisioning_workflow_handoff.py +++ b/tests/test_decisioning_workflow_handoff.py @@ -418,6 +418,48 @@ def create_media_buy(self, req, request_ctx): assert enqueued is False +@pytest.mark.asyncio +async def test_external_workflow_handoff_does_not_make_registry_own_callback( + executor: ThreadPoolExecutor, +) -> None: + """An external publisher, not a bare registry, owns callback persistence.""" + + class _CapturingRegistry(InMemoryTaskRegistry): + def __init__(self) -> None: + super().__init__() + self.issue_extra = {} + + async def issue(self, *, account_id, task_type, request_context=None, **extra): + self.issue_extra = extra + return await super().issue( + account_id=account_id, + task_type=task_type, + request_context=request_context, + ) + + registry = _CapturingRegistry() + ctx = _build_request_context(ToolContext(), Account(id="acct_a"), None) + envelope = await _project_workflow_handoff( + WorkflowHandoff(lambda task_ctx: None), + ctx, + method_name="create_media_buy", + registry=registry, + executor=executor, + request_params=_PushRequest( + push_notification_config={ + "url": "https://buyer.example.com/hooks", + "operation_id": "op-workflow-123", + "token": "echo-token-1234567890", + } + ), + webhook_auto_emit=False, + webhook_external_owner_ready=True, + ) + + assert envelope["status"] == "submitted" + assert registry.issue_extra == {} + + @pytest.mark.asyncio async def test_workflow_handoff_does_not_run_background_task( executor: ThreadPoolExecutor, diff --git a/tests/test_task_webhook_outbox_pg.py b/tests/test_task_webhook_outbox_pg.py new file mode 100644 index 000000000..d14b1e50c --- /dev/null +++ b/tests/test_task_webhook_outbox_pg.py @@ -0,0 +1,479 @@ +"""Unit coverage for the atomic PostgreSQL task-webhook outbox.""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from adcp.webhook_sender import PreparedWebhook, WebhookDeliveryResult + + +def _cursor(value: Any = None) -> AsyncMock: + cursor = AsyncMock() + cursor.fetchone = AsyncMock(return_value=value) + return cursor + + +def _connection(*values: Any) -> AsyncMock: + conn = AsyncMock() + conn.execute = AsyncMock(side_effect=[_cursor(value) for value in values]) + return conn + + +def _pool(*connections: AsyncMock) -> MagicMock: + contexts = [] + for conn in connections: + context = AsyncMock() + context.__aenter__ = AsyncMock(return_value=conn) + context.__aexit__ = AsyncMock(return_value=False) + contexts.append(context) + pool = MagicMock() + pool.connection = MagicMock(side_effect=contexts) + return pool + + +def _sender() -> MagicMock: + sender = MagicMock() + sender.signs_with_rfc9421 = True + sender._owns_client = True + sender._allow_private_destinations = False + sender._timeout = 10.0 + body = json.dumps( + { + "idempotency_key": "whk_1234567890123456", + "task_id": "task_1", + "task_type": "create_media_buy", + "status": "completed", + "operation_id": "op_1", + } + ).encode() + sender.prepare_mcp = MagicMock( + return_value=PreparedWebhook( + url="https://buyer.example/webhook", + idempotency_key="whk_1234567890123456", + body=body, + ) + ) + sender.send_prepared = AsyncMock( + return_value=WebhookDeliveryResult( + status_code=200, + idempotency_key="whk_1234567890123456", + url="https://buyer.example/webhook", + response_headers={}, + response_body=b"{}", + sent_body=body, + ) + ) + return sender + + +def _outbox(pool: Any, sender: 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=sender, + 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 + + sender = WebhookSender.from_bearer_token("test-token") + prepared = sender.prepare_mcp( + url="https://buyer.example/webhook", + task_id="task_1", + task_type="create_media_buy", + status="completed", + result={"media_buy_id": "mb_1"}, + operation_id="op_1", + idempotency_key="whk_1234567890123456", + ) + + payload = json.loads(prepared.body) + assert prepared.idempotency_key == "whk_1234567890123456" + assert payload["idempotency_key"] == prepared.idempotency_key + assert payload["operation_id"] == "op_1" + assert payload["task_id"] == "task_1" + + +@pytest.mark.asyncio +async def test_send_prepared_rejects_mismatched_body_binding_before_http() -> None: + from adcp.webhook_sender import WebhookSender + + sender = WebhookSender.from_bearer_token("test-token") + with pytest.raises(ValueError, match="immutable binding"): + await sender.send_prepared( + PreparedWebhook( + url="https://buyer.example/webhook", + idempotency_key="whk_expected_123456", + body=b'{"idempotency_key":"whk_other_12345678"}', + ) + ) + + +def test_outbox_rejects_unadvertisable_horizon() -> None: + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + with ( + patch("adcp.decisioning.pg.task_webhook_outbox.PG_AVAILABLE", True), + pytest.raises(ValueError, match="86400 through 604800"), + ): + PgTaskWebhookOutbox( + pool=MagicMock(), + sender=_sender(), + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=3600, + ) + + +def test_outbox_rejects_sender_without_sdk_pinned_transport() -> None: + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + sender = _sender() + sender._owns_client = False + with ( + patch("adcp.decisioning.pg.task_webhook_outbox.PG_AVAILABLE", True), + pytest.raises(ValueError, match="IP-pinned transport"), + ): + PgTaskWebhookOutbox( + pool=MagicMock(), + sender=sender, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + ) + + +def test_registration_is_encrypted_and_bound_at_issue_time() -> None: + outbox = _outbox(MagicMock(), _sender()) + encrypted, nonce = outbox.protect_registration( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + url="https://buyer.example/webhook?route=secret", + operation_id="op_1", + token="buyer-secret", + ) + assert b"buyer-secret" not in encrypted + assert b"route=secret" not in encrypted + 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?route=secret", "op_1", "buyer-secret") + + with pytest.raises(ValueError, match="authenticated decryption"): + outbox.open_registration( + account_id="acct_attacker", + task_id="task_1", + task_type="create_media_buy", + encrypted_registration=encrypted, + nonce=nonce, + ) + + +@pytest.mark.asyncio +async def test_enqueue_persists_prepared_bytes_and_horizon_on_callers_connection() -> None: + conn = _connection((41,)) + sender = _sender() + outbox = _outbox(MagicMock(), sender) + + row_id = await outbox.enqueue_terminal( + conn, + task_id="task_1", + account_id="acct_1", + task_type="create_media_buy", + status="completed", + result={"media_buy_id": "mb_1"}, + url="https://buyer.example/webhook", + operation_id="op_1", + token="buyer-token", + ) + + assert row_id == 41 + sender.prepare_mcp.assert_called_once_with( + url="https://buyer.example/webhook", + task_id="task_1", + task_type="create_media_buy", + status="completed", + result={"media_buy_id": "mb_1"}, + operation_id="op_1", + token="buyer-token", + ) + params = conn.execute.await_args.args[1] + 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[-1] == 86_400 + + +@pytest.mark.asyncio +async def test_worker_recovers_after_transient_iteration_failure() -> None: + outbox = _outbox(MagicMock(), _sender()) + outbox.purge_expired = AsyncMock() + outbox.process_one = AsyncMock( + side_effect=[RuntimeError("database restart"), asyncio.CancelledError()] + ) + + with pytest.raises(asyncio.CancelledError): + await outbox.run_worker(poll_interval=0.001) + + assert outbox.process_one.await_count == 2 + assert outbox._worker_started is False + + +@pytest.mark.asyncio +async def test_worker_claims_outside_http_and_acknowledges_success() -> None: + sender = _sender() + body = sender.prepare_mcp.return_value.body + outbox = _outbox(MagicMock(), sender) + nonce = b"n" * 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", + ), + ) + claim_conn = _connection( + None, + ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + "https://buyer.example/webhook", + "op_1", + "whk_1234567890123456", + encrypted, + nonce, + 1, + ), + ) + ack_conn = _connection(None) + outbox._pool = _pool(claim_conn, ack_conn) + + assert await outbox.process_one() is True + + sender.send_prepared.assert_awaited_once() + prepared = sender.send_prepared.await_args.args[0] + assert prepared.body == body + assert prepared.idempotency_key == "whk_1234567890123456" + assert "state = 'delivered'" in ack_conn.execute.await_args.args[0] + + +@pytest.mark.asyncio +async def test_worker_releases_failed_delivery_for_horizon_retry() -> None: + sender = _sender() + sender.send_prepared.return_value = WebhookDeliveryResult( + status_code=503, + idempotency_key="whk_1234567890123456", + url="https://buyer.example/webhook", + response_headers={}, + response_body=b"try later", + sent_body=sender.prepare_mcp.return_value.body, + ) + outbox = _outbox(MagicMock(), sender) + nonce = b"n" * 12 + encrypted = outbox._cipher.encrypt( + nonce, + sender.prepare_mcp.return_value.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", + ), + ) + claim_conn = _connection( + None, + ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + "https://buyer.example/webhook", + "op_1", + "whk_1234567890123456", + encrypted, + nonce, + 1, + ), + ) + release_conn = _connection(None) + outbox._pool = _pool(claim_conn, release_conn) + + assert await outbox.process_one() is True + + sql, params = release_conn.execute.await_args.args + assert "state = CASE" in sql + assert params[1] == 503 + assert params[3] == 7 + + +@pytest.mark.asyncio +async def test_worker_quarantines_body_that_breaks_authenticated_envelope() -> None: + sender = _sender() + body = sender.prepare_mcp.return_value.body + claim_conn = _connection( + None, + ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + "https://buyer.example/webhook", + "op_1", + "whk_1234567890123456", + body, + b"n" * 12, + 1, + ), + ) + quarantine_conn = _connection(None) + outbox = _outbox(_pool(claim_conn, quarantine_conn), sender) + + assert await outbox.process_one() is True + + sender.send_prepared.assert_not_awaited() + sql, params = quarantine_conn.execute.await_args.args + assert "state = 'invalid'" in sql + assert "authenticated binding" in params[0] + + +@pytest.mark.asyncio +async def test_registry_completion_enqueues_on_same_transaction_connection() -> None: + from adcp.decisioning.pg.task_registry import PgTaskRegistry + + conn = _connection( + ("create_media_buy",), + ( + "task_1", + "acct_1", + "create_media_buy", + b"encrypted-registration", + b"registration-nonce", + ), + None, + ) + outbox = AsyncMock() + outbox.open_registration = MagicMock( + return_value=( + "https://buyer.example/webhook", + "op_1", + "buyer-token", + ) + ) + pool = _pool(conn) + outbox._pool = pool + with patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True): + registry = PgTaskRegistry( + pool=pool, + task_webhook_outbox=outbox, + ) + + await registry.complete("task_1", {"media_buy_id": "mb_1"}) + + outbox.enqueue_terminal.assert_awaited_once_with( + conn, + task_id="task_1", + account_id="acct_1", + task_type="create_media_buy", + status="completed", + result={"media_buy_id": "mb_1"}, + url="https://buyer.example/webhook", + operation_id="op_1", + token="buyer-token", + ) + outbox.open_registration.assert_called_once_with( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + encrypted_registration=b"encrypted-registration", + nonce=b"registration-nonce", + ) + assert "webhook_registration = NULL" in conn.execute.await_args_list[-1].args[0] + + +@pytest.mark.asyncio +async def test_registry_strips_credentials_before_terminal_update() -> None: + from adcp.decisioning.pg.task_registry import PgTaskRegistry + + conn = _connection( + ("sync_accounts",), + ("task_1", "acct_1", "sync_accounts", None, None), + ) + pool = _pool(conn) + with patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True): + registry = PgTaskRegistry(pool=pool) + result = { + "accounts": [ + { + "account_id": "acct_1", + "notification_configs": [ + { + "authentication": { + "schemes": ["Bearer"], + "credentials": "secret-that-must-not-enter-wal", + } + } + ], + } + ] + } + + await registry.complete("task_1", result) + + persisted = json.loads(conn.execute.await_args_list[1].args[1][0]) + assert "secret-that-must-not-enter-wal" not in str(persisted) + + +def test_registry_rejects_outbox_on_different_pool() -> None: + from adcp.decisioning.pg.task_registry import PgTaskRegistry + + outbox = MagicMock() + outbox._pool = MagicMock() + with ( + patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True), + pytest.raises(ValueError, match="same connection pool"), + ): + PgTaskRegistry(pool=MagicMock(), task_webhook_outbox=outbox) + + +def test_delivery_status_classification_treats_conflict_as_permanent() -> None: + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + assert PgTaskWebhookOutbox._is_retryable_http_status(409) is False + assert PgTaskWebhookOutbox._is_retryable_http_status(429) is True + assert PgTaskWebhookOutbox._is_retryable_http_status(503) is True + + +def test_dns_resolution_failure_is_marked_transient() -> None: + from adcp.signing.jwks import SSRFValidationError + + assert SSRFValidationError("blocked private IP").transient is False + assert SSRFValidationError("resolver unavailable", transient=True).transient is True diff --git a/tests/test_webhook_signing_capabilities.py b/tests/test_webhook_signing_capabilities.py index ce52a6fa3..737f6071d 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 +from unittest.mock import MagicMock, patch import httpx import pytest @@ -60,6 +60,11 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: return httpx.Response(200, content=b"{}", request=request) +class _LargeResponseTransport(httpx.AsyncBaseTransport): + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + return httpx.Response(503, content=b"x" * (128 * 1024), request=request) + + # ----- AC4: outbound webhooks carry RFC 9421 headers ----- @@ -111,6 +116,23 @@ async def test_outbound_webhook_carries_rfc9421_signature_headers() -> None: ), f"Signature-Input missing keyid: {signature_input!r}" +@pytest.mark.asyncio +async def test_sender_bounds_untrusted_receiver_response_body() -> None: + client = httpx.AsyncClient(transport=_LargeResponseTransport()) + sender = WebhookSender.from_jwk(_jwk_with_private(), client=client) + async with sender: + result = await sender.send_mcp( + url="https://buyer.example/webhook", + task_id="task_1", + task_type="create_media_buy", + operation_id="op_1", + status="completed", + result={"media_buy_id": "mb_1"}, + ) + assert result.status_code == 503 + assert len(result.response_body) == 64 * 1024 + + @pytest.mark.asyncio async def test_bearer_sender_does_not_emit_rfc9421_headers() -> None: """A bearer-token sender MUST NOT emit RFC 9421 headers — the @@ -169,6 +191,38 @@ def __init__(self, sender: WebhookSender | None, horizon: int = 86400) -> None: self.delivery_retry_horizon_seconds = horizon +class _AtomicOutbox: + delivery_state_is_durable = True + supports_atomic_task_outbox = True + + def __init__(self, sender: WebhookSender, horizon: int = 86400) -> None: + self._sender = sender + self.delivery_retry_horizon_seconds = horizon + + +class _RegistryWithOutbox: + def __init__(self, outbox: _AtomicOutbox) -> None: + self.task_webhook_outbox = outbox + + +def _sdk_registry_with_outbox(sender: WebhookSender, horizon: int = 86400): + from adcp.decisioning.pg.task_registry import PgTaskRegistry + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + pool = MagicMock() + 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=sender, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=horizon, + ) + return PgTaskRegistry(pool=pool, task_webhook_outbox=outbox) + + def test_boot_passes_when_capabilities_omit_webhook_signing() -> None: """No advertisement, no obligation — validator returns silently.""" validate_webhook_signing_for_capabilities( @@ -341,6 +395,39 @@ def test_boot_rejects_bare_jwk_sender_without_durable_delivery_state() -> None: assert exc_info.value.details["missing"] == "external_durable_webhook_outbox" +def test_boot_accepts_registry_backed_atomic_outbox() -> None: + sender = WebhookSender.from_jwk(_jwk_with_private()) + 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_outbox(sender), + ) + + +def test_boot_rejects_atomic_outbox_shorter_than_advertisement() -> None: + sender = WebhookSender.from_jwk(_jwk_with_private()) + with pytest.raises(AdcpError) as exc_info: + validate_webhook_signing_for_capabilities( + capabilities=_Caps( + webhook_signing=WebhookSigning( + supported=True, + delivery_retry_horizon_seconds=172800, + ) + ), + sender=None, + supervisor=None, + registry=_sdk_registry_with_outbox(sender, horizon=86400), + ) + assert exc_info.value.details["missing"] == "atomic_durable_webhook_outbox" + + def test_boot_rejects_inmemory_supervisor_for_advertised_horizon() -> None: sender = WebhookSender.from_jwk(_jwk_with_private()) supervisor = InMemoryWebhookDeliverySupervisor(sender=sender) From d78c219fb6e24a13d85712a8a35ac5a9d6215024 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 23 Aug 2026 06:25:15 +0200 Subject: [PATCH 2/3] fix(webhooks): close outbox atomicity gaps --- MIGRATION_v7_to_v8.md | 16 ++- src/adcp/decisioning/pg/task_registry.py | 115 ++++++++++-------- src/adcp/decisioning/webhook_emit.py | 12 +- .../test_pg_task_webhook_outbox.py | 10 +- tests/test_task_webhook_outbox_pg.py | 51 ++++++++ tests/test_webhook_signing_capabilities.py | 90 ++++++++++++++ 6 files changed, 232 insertions(+), 62 deletions(-) diff --git a/MIGRATION_v7_to_v8.md b/MIGRATION_v7_to_v8.md index 2ba418fd8..ab59ee405 100644 --- a/MIGRATION_v7_to_v8.md +++ b/MIGRATION_v7_to_v8.md @@ -63,10 +63,18 @@ unretained `asyncio.create_task()` immediately before synchronous `serve()`; that task does not share the server lifecycle. The registry commits terminal state and the immutable prepared webhook in one -transaction. The body and callback token are AES-256-GCM encrypted at rest and -bound to the task, account, URL, operation, status, and idempotency key. The -retry horizon starts at the first delivery attempt; the worker replays the same -body/key and retains proof until that exact advertised horizon ends. +explicit transaction, including when the supplied pool uses autocommit. The +body and callback token are AES-256-GCM encrypted at rest and bound to the task, +account, URL, operation, status, and idempotency key. The retry horizon starts +at the first delivery attempt; the worker replays the same body/key and retains +proof until that exact advertised horizon ends. + +SDK-managed publication requires the exact `PgTaskRegistry` and +`PgTaskWebhookOutbox` types. Subclasses are rejected because overriding task +issuance, terminal persistence, enqueue, or claim methods would invalidate the +audited atomicity contract while still satisfying a structural or +`isinstance()` check. Custom registries and publishers must use the explicit +externally managed ownership path. If publication is owned outside this SDK instead, set `webhook_signing_managed_externally=True`, set diff --git a/src/adcp/decisioning/pg/task_registry.py b/src/adcp/decisioning/pg/task_registry.py index 1a7c63d74..246289e43 100644 --- a/src/adcp/decisioning/pg/task_registry.py +++ b/src/adcp/decisioning/pg/task_registry.py @@ -346,39 +346,44 @@ async def complete( race each other into double-completion without detection. """ async with self._pool.connection() as conn: - type_cursor = await conn.execute(self._sql_get_task_type, (task_id,)) - type_row = await type_cursor.fetchone() - safe_result = ( - strip_credentials_from_wire_result(type_row[0], result) - if type_row is not None - else result - ) - cur = await conn.execute( - self._sql_complete, - (json.dumps(safe_result), time.time(), task_id), - ) - row = await cur.fetchone() - if row is not None: - await self._enqueue_terminal_if_registered( - conn, - row=row, - status="completed", - payload=safe_result, + # Do not rely on the pool connection's implicit transaction. + # Explicitly bind the terminal transition, outbox insert, and + # callback-registration clear even when adopters configure the + # pool with autocommit=True. + async with conn.transaction(): + type_cursor = await conn.execute(self._sql_get_task_type, (task_id,)) + type_row = await type_cursor.fetchone() + safe_result = ( + strip_credentials_from_wire_result(type_row[0], result) + if type_row is not None + else result ) - return # updated successfully - - # Zero rows in RETURNING — task is unknown or already terminal. - cur2 = await conn.execute(self._sql_get_state_result, (task_id,)) - row = await cur2.fetchone() - if row is None: - raise ValueError(f"Task {task_id!r} not found") - state, existing_result, task_type = row - safe_result = strip_credentials_from_wire_result(task_type, result) - if state == "completed": - if existing_result == safe_result: - return # idempotent - raise ValueError(f"Task {task_id!r} already completed with a different result") - raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") + cur = await conn.execute( + self._sql_complete, + (json.dumps(safe_result), time.time(), task_id), + ) + row = await cur.fetchone() + if row is not None: + await self._enqueue_terminal_if_registered( + conn, + row=row, + status="completed", + payload=safe_result, + ) + return # updated successfully + + # Zero rows in RETURNING — task is unknown or already terminal. + cur2 = await conn.execute(self._sql_get_state_result, (task_id,)) + row = await cur2.fetchone() + if row is None: + raise ValueError(f"Task {task_id!r} not found") + state, existing_result, task_type = row + safe_result = strip_credentials_from_wire_result(task_type, result) + if state == "completed": + if existing_result == safe_result: + return # idempotent + raise ValueError(f"Task {task_id!r} already completed with a different result") + raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") async def fail( self, @@ -391,28 +396,32 @@ async def fail( :class:`ValueError` on conflicting re-failure. """ async with self._pool.connection() as conn: - cur = await conn.execute(self._sql_fail, (json.dumps(error), time.time(), task_id)) - row = await cur.fetchone() - if row is not None: - await self._enqueue_terminal_if_registered( - conn, - row=row, - status="failed", - payload=error, + async with conn.transaction(): + cur = await conn.execute( + self._sql_fail, + (json.dumps(error), time.time(), task_id), ) - return # updated successfully - - # Zero rows in RETURNING — task is unknown or already terminal. - cur2 = await conn.execute(self._sql_get_state_error, (task_id,)) - row = await cur2.fetchone() - if row is None: - raise ValueError(f"Task {task_id!r} not found") - state, existing_error = row - if state == "failed": - if existing_error == error: - return # idempotent - raise ValueError(f"Task {task_id!r} already failed with a different error") - raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") + row = await cur.fetchone() + if row is not None: + await self._enqueue_terminal_if_registered( + conn, + row=row, + status="failed", + payload=error, + ) + return # updated successfully + + # Zero rows in RETURNING — task is unknown or already terminal. + cur2 = await conn.execute(self._sql_get_state_error, (task_id,)) + row = await cur2.fetchone() + if row is None: + raise ValueError(f"Task {task_id!r} not found") + state, existing_error = row + if state == "failed": + if existing_error == error: + return # idempotent + raise ValueError(f"Task {task_id!r} already failed with a different error") + raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") async def get( self, diff --git a/src/adcp/decisioning/webhook_emit.py b/src/adcp/decisioning/webhook_emit.py index f840b8fab..5856c5b0a 100644 --- a/src/adcp/decisioning/webhook_emit.py +++ b/src/adcp/decisioning/webhook_emit.py @@ -33,7 +33,13 @@ def _sdk_task_outbox_pair_ready(registry: Any, task_outbox: Any) -> bool: - """Accept only the concrete registry/outbox pair whose atomicity we own.""" + """Accept only the exact registry/outbox types whose atomicity we own. + + Subclasses are deliberately rejected. The callback-registration and + terminal-enqueue methods are load-bearing parts of the audited contract; + an override can silently discard webhook arguments or bypass the shared + transaction while still passing an ``isinstance`` check. + """ if registry is None or task_outbox is None: return False try: @@ -42,8 +48,8 @@ def _sdk_task_outbox_pair_ready(registry: Any, task_outbox: Any) -> bool: except ImportError: return False return ( - isinstance(registry, PgTaskRegistry) - and isinstance(task_outbox, PgTaskWebhookOutbox) + type(registry) is PgTaskRegistry + and type(task_outbox) is PgTaskWebhookOutbox and registry.task_webhook_outbox is task_outbox and registry._pool is task_outbox._pool ) diff --git a/tests/conformance/decisioning/test_pg_task_webhook_outbox.py b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py index 82e424b52..0f1ba5c84 100644 --- a/tests/conformance/decisioning/test_pg_task_webhook_outbox.py +++ b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py @@ -166,12 +166,18 @@ async def test_concurrent_claims_deliver_one_attempt(stack) -> None: @pytest.mark.asyncio -async def test_missing_outbox_table_rolls_back_terminal_transition() -> None: +async def test_autocommit_pool_rolls_back_terminal_transition_when_enqueue_fails() -> 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, open=False) as pool: + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + kwargs={"autocommit": True}, + open=False, + ) as pool: await pool.open() + async with pool.connection() as conn: + assert conn.autocommit is True outbox = PgTaskWebhookOutbox( pool=pool, sender=_sender(), diff --git a/tests/test_task_webhook_outbox_pg.py b/tests/test_task_webhook_outbox_pg.py index d14b1e50c..fc4f153d9 100644 --- a/tests/test_task_webhook_outbox_pg.py +++ b/tests/test_task_webhook_outbox_pg.py @@ -21,6 +21,10 @@ def _cursor(value: Any = None) -> AsyncMock: def _connection(*values: Any) -> AsyncMock: conn = AsyncMock() conn.execute = AsyncMock(side_effect=[_cursor(value) for value in values]) + transaction_context = AsyncMock() + transaction_context.__aenter__ = AsyncMock(return_value=None) + transaction_context.__aexit__ = AsyncMock(return_value=False) + conn.transaction = MagicMock(return_value=transaction_context) return conn @@ -416,6 +420,52 @@ async def test_registry_completion_enqueues_on_same_transaction_connection() -> encrypted_registration=b"encrypted-registration", nonce=b"registration-nonce", ) + 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_failure_enqueues_on_same_explicit_transaction() -> None: + from adcp.decisioning.pg.task_registry import PgTaskRegistry + + conn = _connection( + ( + "task_1", + "acct_1", + "create_media_buy", + b"encrypted-registration", + b"registration-nonce", + ), + None, + ) + outbox = AsyncMock() + outbox.open_registration = MagicMock( + return_value=( + "https://buyer.example/webhook", + "op_1", + None, + ) + ) + pool = _pool(conn) + outbox._pool = pool + with patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True): + registry = PgTaskRegistry(pool=pool, task_webhook_outbox=outbox) + + error = {"code": "INTERNAL_ERROR", "message": "failed"} + await registry.fail("task_1", error) + + outbox.enqueue_terminal.assert_awaited_once_with( + conn, + task_id="task_1", + account_id="acct_1", + task_type="create_media_buy", + status="failed", + result=error, + url="https://buyer.example/webhook", + operation_id="op_1", + token=None, + ) + conn.transaction.assert_called_once_with() assert "webhook_registration = NULL" in conn.execute.await_args_list[-1].args[0] @@ -450,6 +500,7 @@ async def test_registry_strips_credentials_before_terminal_update() -> None: persisted = json.loads(conn.execute.await_args_list[1].args[1][0]) assert "secret-that-must-not-enter-wal" not in str(persisted) + conn.transaction.assert_called_once_with() def test_registry_rejects_outbox_on_different_pool() -> None: diff --git a/tests/test_webhook_signing_capabilities.py b/tests/test_webhook_signing_capabilities.py index 737f6071d..25a03cdde 100644 --- a/tests/test_webhook_signing_capabilities.py +++ b/tests/test_webhook_signing_capabilities.py @@ -411,6 +411,96 @@ def test_boot_accepts_registry_backed_atomic_outbox() -> None: ) +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 + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + class _UnsafeRegistry(PgTaskRegistry): + async def issue( + self, + *, + account_id, + task_type, + request_context=None, + **_extra, + ): + # Mirrors the failure mode this gate prevents: callback arguments + # are accepted but silently discarded. + return await super().issue( + account_id=account_id, + task_type=task_type, + request_context=request_context, + ) + + sender = WebhookSender.from_jwk(_jwk_with_private()) + pool = MagicMock() + 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=sender, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + ) + registry = _UnsafeRegistry(pool=pool, task_webhook_outbox=outbox) + + with pytest.raises(AdcpError) as exc_info: + validate_webhook_signing_for_capabilities( + capabilities=_Caps( + webhook_signing=WebhookSigning( + supported=True, + delivery_retry_horizon_seconds=86_400, + ) + ), + sender=None, + supervisor=None, + registry=registry, + ) + + assert exc_info.value.details["missing"] == "verified_sdk_task_webhook_outbox_pair" + + +def test_boot_rejects_pg_outbox_subclass() -> None: + """Outbox subclasses can override enqueue/claim semantics and are unsafe.""" + from adcp.decisioning.pg.task_registry import PgTaskRegistry + from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox + + class _UnsafeOutbox(PgTaskWebhookOutbox): + pass + + sender = WebhookSender.from_jwk(_jwk_with_private()) + pool = MagicMock() + with ( + patch("adcp.decisioning.pg.task_registry.PG_AVAILABLE", True), + patch("adcp.decisioning.pg.task_webhook_outbox.PG_AVAILABLE", True), + ): + outbox = _UnsafeOutbox( + pool=pool, + sender=sender, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + ) + registry = PgTaskRegistry(pool=pool, task_webhook_outbox=outbox) + + with pytest.raises(AdcpError) as exc_info: + validate_webhook_signing_for_capabilities( + capabilities=_Caps( + webhook_signing=WebhookSigning( + supported=True, + delivery_retry_horizon_seconds=86_400, + ) + ), + sender=None, + supervisor=None, + registry=registry, + ) + + assert exc_info.value.details["missing"] == "verified_sdk_task_webhook_outbox_pair" + + def test_boot_rejects_atomic_outbox_shorter_than_advertisement() -> None: sender = WebhookSender.from_jwk(_jwk_with_private()) with pytest.raises(AdcpError) as exc_info: From 9800477e64ac3e81f6f2fda62651bae875765bd4 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 23 Aug 2026 08:59:07 +0200 Subject: [PATCH 3/3] chore(webhooks): clean up optional pg import --- src/adcp/decisioning/pg/task_webhook_outbox.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/adcp/decisioning/pg/task_webhook_outbox.py b/src/adcp/decisioning/pg/task_webhook_outbox.py index c30297f53..421168f79 100644 --- a/src/adcp/decisioning/pg/task_webhook_outbox.py +++ b/src/adcp/decisioning/pg/task_webhook_outbox.py @@ -34,9 +34,9 @@ from adcp.webhook_sender import WebhookSender try: - from psycopg_pool import AsyncConnectionPool as _AsyncConnectionPool # noqa: F401 + import psycopg_pool - PG_AVAILABLE = True + PG_AVAILABLE = bool(psycopg_pool.AsyncConnectionPool) except ImportError: PG_AVAILABLE = False