From fd5165325c665b6b87f632f5f7b192d3818013b7 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sat, 7 Nov 2026 13:43:41 +0000 Subject: [PATCH 1/3] feat(webhooks): sign deliveries under web-bot-auth when a registration asks AdCP 3.3 lets a webhook registration omit authentication and name a signing_profile. WebhookSender could not produce web-bot-auth, so a seller built on the SDK, including its task-webhook outbox, could not honor a buyer registered for it. Sender: - WebhookSender.from_wba(private_key, agent_origin=...) signs each attempt afresh with sign_wba_request (new 64-byte nonce and created, thumbprint keyid, Signature-Agent naming the origin). - with_web_bot_auth() adds web-bot-auth to an adcp/webhook-signing/v1 sender; each delivery is signed under its registration's profile, and resend keeps the original's profile. - signing_profiles / signing_algorithms; Signature-Agent is reserved from extra_headers. - sign_web_bot_auth_directory() signs the origin's key directory. - challenge_webhook_destination signs under the config's signing_profile. Outboxes: - PgTaskWebhookOutbox stores push_notification_config.signing_profile with the encrypted registration and delivery and signs every attempt under it. - PgNotificationOutbox persists required_signing_profile (envelope v2 only for non-default profiles, so earlier pods quarantine instead of mis-signing). - The reporting notification worker signs web-bot-auth subscriptions with ReportingWebBotAuthKey; ReportingProductionSigning(web_bot_auth=True) advertises both profiles. Unset profiles keep fingerprints byte-stable. Validation: - Dispatch refuses, with INVALID_REQUEST before the task is issued, a push registration requiring a profile outside the request's effective webhook_signing.profiles (tenant-scoped overrides included), for every publisher. - Boot validation requires a key for every advertised profile and every sender algorithm to be advertised. Also annotates the pre-existing bandit B608/B311 findings in task_webhook_outbox.py (validated table identifier; retry jitter), which the pre-commit hook reports for any change to that file. Closes #1859 Co-Authored-By: Claude Opus 5.5 --- docs/adcp-3.3-sdk-inventory.md | 4 +- docs/reporting-production.md | 9 +- docs/webhooks/reporting-operation-id-3.2.md | 9 +- docs/webhooks/web-bot-auth.md | 185 +++++ src/adcp/decisioning/dispatch.py | 105 +++ src/adcp/decisioning/handler.py | 6 + src/adcp/decisioning/pg/lazy.py | 18 +- src/adcp/decisioning/pg/task_registry.py | 16 +- .../decisioning/pg/task_webhook_outbox.py | 192 +++-- src/adcp/decisioning/webhook_emit.py | 95 ++- src/adcp/notification_outbox_pg.py | 48 +- src/adcp/reporting/ledger/models.py | 4 + src/adcp/reporting/outbox/__init__.py | 2 + src/adcp/reporting/outbox/routing.py | 51 +- src/adcp/reporting/outbox/worker.py | 13 + .../reporting/production/notifications.py | 49 +- src/adcp/webhook_auth.py | 51 +- src/adcp/webhook_sender.py | 313 +++++++- src/adcp/webhooks.py | 35 +- tests/test_decisioning_webhook_emit.py | 94 +++ tests/test_reporting_webhook_wba.py | 251 +++++++ tests/test_task_webhook_outbox_pg.py | 13 +- tests/test_webhook_challenge.py | 55 ++ tests/test_webhook_sender_wba.py | 688 ++++++++++++++++++ 24 files changed, 2165 insertions(+), 141 deletions(-) create mode 100644 docs/webhooks/web-bot-auth.md create mode 100644 tests/test_reporting_webhook_wba.py create mode 100644 tests/test_webhook_sender_wba.py diff --git a/docs/adcp-3.3-sdk-inventory.md b/docs/adcp-3.3-sdk-inventory.md index da89480a6a..e24a34dcc6 100644 --- a/docs/adcp-3.3-sdk-inventory.md +++ b/docs/adcp-3.3-sdk-inventory.md @@ -88,8 +88,8 @@ it adopted the bundle, and what is left to the codegen/wire-fidelity (2) and | `by_property`, `by_installment_property` | `MediaBuyFeatures.supports_property_breakdown`, `supports_installment_property_breakdown` | Property-grain rollups. | generated; storage package rollups: (3), #1669 | | Reliable Reporting `source.custody` | `core/reporting-report-definition.json` `Source.custody`; enum `reporting-custody` (`system_of_record`, `upstream_derived`) | | generated; `adcp.reporting` ledger: (3), #1669 | | `reporting_webhook.operation_id` | `core/reporting-webhook.json` (`minLength` 1, `maxLength` 255, `^[A-Za-z0-9_.:-]{1,255}$`) | Native in 3.3 (adcp#7895); echoed verbatim; nothing is sent to a registration without it. The 3.2 forward-compat runtime patch in `_forward_compat.py` is retired. | done (1) | -| `reporting_webhook.authentication`, `artifact_webhook.authentication` optional | `core/reporting-webhook.json` `required` is `url`, `reporting_frequency`; `create_media_buy` `ArtifactWebhook` | Required only on requests served below 3.3; buyers pinned to 3.2 keep sending them. The 3.3 → 3.2 downgrade of a request without `authentication` must fail structured (#1621). | generated; downgrade fails structured: done (3), #1667; seller signing when absent: `webhook_signing_profile_for` (also refuses a missing `authentication` below 3.3), and `WebhookSender` refuses before any I/O to send such a registration unsigned or under another profile: done (3), #1644; Web Bot Auth webhook signing: not built | -| `signing_profile` | `ReportingWebhook`, `PushNotificationConfig`, `NotificationConfig`, `AgentNotificationConfig`, `ArtifactWebhook`; enum `webhook-signing-profile` | Per-registration webhook signing profile (Web Bot Auth or RFC 9421). | generated; webhook sender profile selection: (3) | +| `reporting_webhook.authentication`, `artifact_webhook.authentication` optional | `core/reporting-webhook.json` `required` is `url`, `reporting_frequency`; `create_media_buy` `ArtifactWebhook` | Required only on requests served below 3.3; buyers pinned to 3.2 keep sending them. The 3.3 → 3.2 downgrade of a request without `authentication` must fail structured (#1621). | generated; downgrade fails structured: done (3), #1667; seller signing when absent: `webhook_signing_profile_for` (also refuses a missing `authentication` below 3.3), and `WebhookSender` refuses before any I/O to send such a registration unsigned or under another profile: done (3), #1644; Web Bot Auth webhook signing (`WebhookSender.from_wba`, `with_web_bot_auth`): done (3), #1859 | +| `signing_profile` | `ReportingWebhook`, `PushNotificationConfig`, `NotificationConfig`, `AgentNotificationConfig`, `ArtifactWebhook`; enum `webhook-signing-profile` | Per-registration webhook signing profile (Web Bot Auth or RFC 9421). | generated; webhook sender profile selection: done (3), #1859 — `WebhookSender` signs under the registration's profile; `PgTaskWebhookOutbox`, `PgNotificationOutbox` and the reporting notification worker (`ReportingNotificationSubscription.signing_profile`, `ReportingWebBotAuthKey`) carry it to every attempt. `NotificationConfig` / `AgentNotificationConfig` registrations: the caller sets `required_signing_profile` on the prepared notification | | Multi-qualifier metrics omit flat totals | `totals` on buys, windows and webhook per-buy | Receivers ignore `aggregated_totals` in delivery webhooks. | `WebhookReceiver` drops `aggregated_totals` from a `media_buy_delivery` result before parsing; the sender already dropped it: done (3), #1644. The SDK builds no buy or window `totals` (adopters' `get_media_buy_delivery` does), so there is no flat carrier to omit; `adcp.reporting` readers: #1669 | | `sync_audiences` `audience_size`, `targetable_size`, `match_rate_basis`, `platform_id` | `core/audience-size.json`; enums `audience-size-precision`, `audience-match-rate-basis` | Ranges; `matched_count` / `effective_match_rate` may be absent on `ready` audiences. | generated | | `sync_event_sources` `platform_id`, `archived` | request/response `EventSource`; capability `media_buy.event_source_archive` | `platform_id` dual-emitted with deprecated `seller_id`. | generated | diff --git a/docs/reporting-production.md b/docs/reporting-production.md index dcf6f2d099..55c9465dc7 100644 --- a/docs/reporting-production.md +++ b/docs/reporting-production.md @@ -140,7 +140,14 @@ contract. Private key material stays in the resolver and sender. The operator must publish its brand document and corresponding signing keys at the declared identity; the SDK does not assert that a supplied URL proves external ownership. The declaration is RFC 9421 `adcp/webhook-signing/v1`, the exact supplied -algorithm set, no legacy fallback, and an 86,400-second retry horizon. Each +algorithm set, no legacy fallback, and an 86,400-second retry horizon. Pass +`web_bot_auth=True` to also sign subscriptions selecting `web-bot-auth` (AdCP +3.3): the declaration then lists both `profiles`, `ed25519` must be among the +algorithms, and every resolved `ReportingSigningMaterial` must carry +`web_bot_auth=ReportingWebBotAuthKey(key, agent_origin)`. Without it, a +`web-bot-auth` subscription is skipped with a warning, and the account's other +subscribers are still notified. See +[Web Bot Auth webhooks](webhooks/web-bot-auth.md#reporting-notifications). Each queue reserves the first HTTP attempt and its key/body binding before peer I/O, using database time in PostgreSQL. Window insertion, HTTP attempt and ordinal reservation share one transaction; a failed reservation leaves none of them. diff --git a/docs/webhooks/reporting-operation-id-3.2.md b/docs/webhooks/reporting-operation-id-3.2.md index c9442393d4..c2e07feaba 100644 --- a/docs/webhooks/reporting-operation-id-3.2.md +++ b/docs/webhooks/reporting-operation-id-3.2.md @@ -151,10 +151,11 @@ raises `WebhookSigningProfileUnavailableError` before any network I/O unless the sender signs under that profile. A bearer or HMAC sender (`from_bearer_token`, `from_adcp_legacy_hmac`, `from_standard_webhooks_secret`) would otherwise send a credential the buyer never registered. Use -`WebhookSender.from_jwk` / `from_pem` for these registrations. The SDK does not -yet sign webhooks under `web-bot-auth`, so a registration selecting it also -raises; do not list `web-bot-auth` in `webhook_signing.profiles` and reject such -registrations with `INVALID_REQUEST`. For artifact webhooks pass the stored +`WebhookSender.from_jwk` / `from_pem` for `adcp/webhook-signing/v1`, and +`WebhookSender.from_wba` (or `with_web_bot_auth` for both profiles) for +`web-bot-auth`; see [Web Bot Auth webhooks](web-bot-auth.md). Reject a +registration selecting a profile you do not list in `webhook_signing.profiles` +with `INVALID_REQUEST`. For artifact webhooks pass the stored `artifact_webhook` as `registration=` to `prepare_artifact_webhook` / `send_artifact_webhook`, or call `sender.check_registration(registration)`. diff --git a/docs/webhooks/web-bot-auth.md b/docs/webhooks/web-bot-auth.md new file mode 100644 index 0000000000..6e51cdfcfc --- /dev/null +++ b/docs/webhooks/web-bot-auth.md @@ -0,0 +1,185 @@ +# Web Bot Auth webhooks (AdCP 3.3) + +At AdCP 3.3 a webhook registration (`push_notification_config`, +`reporting_webhook`, `artifact_webhook`, `accounts[].notification_configs[]`, +or an agent-level notification configuration) may omit `authentication` and +name a `signing_profile`. The seller signs every delivery under that profile: + +- `adcp/webhook-signing/v1`: the AdCP RFC 9421 webhook profile, also used + when the registration names no profile. Sign with + `WebhookSender.from_jwk` / `from_pem`. +- `web-bot-auth`: the Web Bot Auth profile, experimental in 3.3 and the only + HTTP signing profile in AdCP 4.0. A webhook is an ordinary Web Bot Auth + request from the seller's origin. Sign with `WebhookSender.from_wba`. + +## Senders + +A seller whose buyers all use Web Bot Auth: + +```python +from adcp.signing import load_private_key_pem +from adcp.webhooks import WebhookSender + +wba_key = load_private_key_pem(Path("wba-ed25519.pem").read_bytes()) +sender = WebhookSender.from_wba(wba_key, agent_origin="https://seller.example") +``` + +During 3.3 a seller usually has buyers on both profiles. Extend the existing +RFC 9421 sender, which stays the default: + +```python +sender = WebhookSender.from_jwk(webhook_jwk).with_web_bot_auth( + wba_key, agent_origin="https://seller.example" +) +assert sender.signing_profiles == {"adcp/webhook-signing/v1", "web-bot-auth"} +``` + +The sender signs a delivery under the profile its registration selected, +which the prepared request carries as `required_signing_profile` +(`prepare_media_buy_delivery`, `prepare_artifact_webhook(registration=...)`, +and the task outbox below set it). A delivery whose profile the sender holds +no key for raises `WebhookSigningProfileUnavailableError` before any network +I/O. `sender.check_registration(registration)` runs the same check when the +buyer registers. Prove control of a notification destination under its +profile too: `challenge_webhook_destination(..., sender=sender, +signing_profile=config.signing_profile)` signs the challenge under it +(`adcp/webhook-signing/v1` when omitted). + +Each `web-bot-auth` attempt is signed afresh by +`adcp.signing.wba.sign_wba_request`: a new 64-byte nonce and `created`, +`Signature-Agent` naming the origin, the key's RFC 7638 thumbprint as +`keyid`, and `@method`, `@target-uri`, `@authority`, `content-type`, +`content-digest` and `signature-agent` covered. Receivers record nonces, +including on an attempt they answered 503, so `resend`, `send_prepared` and +both outboxes never resend a stored signature. `resend` signs under the +profile the original was signed under. `Signature-Agent` is reserved like the +other signature fields: `extra_headers` cannot set it. + +The key must be Ed25519, the profile's one algorithm. An ES256 key stays on +`adcp/webhook-signing/v1`. + +## Publish the key directory first + +Receivers verify a `web-bot-auth` webhook against the key directory at +`https://seller.example/.well-known/http-message-signatures-directory`. The +sender's key is verifiable only once that directory is live, so publish it +before the first delivery: + +```python +from adcp.signing.wba import DIRECTORY_PATH + +directory = sender.sign_web_bot_auth_directory( + other_keys=[request_signing_key], # every other Web Bot Auth key of the origin +) + +@app.get(DIRECTORY_PATH) +def key_directory() -> Response: + return Response(directory.body, status_code=200, headers=dict(directory.headers)) +``` + +The directory belongs to the origin, not to the sender. List every Ed25519 key +the origin signs with under Web Bot Auth, including request-signing keys and +keys mid-rotation, in `other_keys`; a key left out stops verifying once +receivers refresh their cached copy (up to five minutes). The signatures +expire after `expires_in_seconds` (one day by default), so re-sign and +republish well before then. When rotating, publish the new key, wait for +caches to expire, switch the sender, and keep the old key in the directory +while redeliveries signed with it may still arrive. +`adcp.signing.wba.sign_key_directory` is the underlying helper; see +[agent identity](../agent-identity-3-3-migration.md) for the directory rules. + +## Capabilities + +Advertise the profiles you can sign under: + +```python +WebhookSigning( + supported=True, + profiles=["adcp/webhook-signing/v1", "web-bot-auth"], + algorithms=["ed25519"], + delivery_retry_horizon_seconds=86400, +) +``` + +At boot, `validate_webhook_signing_for_capabilities` refuses a wired sender +that holds no key for an advertised profile (omitted `profiles` means +`adcp/webhook-signing/v1`, so a `from_wba`-only sender must list +`web-bot-auth`), or that signs with an algorithm outside `algorithms` (a dual +sender is checked for both of its keys). + +## Outboxes + +`PgTaskWebhookOutbox` reads `push_notification_config.signing_profile` when +the task is issued and stores it in the encrypted registration and, at the +terminal transition, in the encrypted delivery row. Every attempt is signed +under it, `adcp/webhook-signing/v1` when the registration named none and +carried no `authentication`. Before the task is issued, a registration is +refused with `INVALID_REQUEST` (field `push_notification_config.signing_profile`) +when the profile it requires is not in this request's advertised +`webhook_signing.profiles` (`adcp/webhook-signing/v1` when omitted), checked +against the effective capabilities, a tenant's +`get_adcp_capabilities_for_request` override included. That applies to every +publisher: a fixed-sender outbox, a `sender_resolver` outbox, or an externally +managed one. A fixed sender is also checked for a key under that profile. A +`sender_resolver` outbox's resolved sender is checked again at each attempt, +and a row it cannot sign is quarantined. A resolved sender's every algorithm +must be in the scope's `advertised_algorithms`, including its Web Bot Auth +key's `ed25519`. Registrations naming `adcp/webhook-signing/v1` are stored +exactly like ones naming no profile. + +`PgNotificationOutbox` stores the prepared request's +`required_signing_profile` with the body and signs each attempt under it. For +notification registrations (`accounts[].notification_configs[]`, agent-level +configurations), set it from the stored registration before enqueueing: + +```python +from dataclasses import replace +from adcp.webhooks import webhook_signing_profile_for + +prepared = WebhookSender.prepare_account_status_changed(url=str(config.url), payload=event) +prepared = replace(prepared, required_signing_profile=webhook_signing_profile_for(config)) +await outbox.enqueue_prepared( + conn, prepared, notification_type="account.status_changed", caller_scope_id=scope +) +``` + +Rolling deploys: a delivery row stored with a profile by an upgraded pod is +quarantined, not sent under the wrong profile, by a pod running an earlier +SDK, in both outboxes. Rows without a profile are unchanged. One gap remains: +a task registered on an upgraded pod and completed on an earlier one is +enqueued and signed by the earlier pod under its sender's default profile, +because that pod ignores the registration's profile. Finish the rollout +before listing `web-bot-auth` in `webhook_signing.profiles`. + +## Reporting notifications + +`ReportingNotificationSubscription` takes the registration's +`signing_profile` (only with a `signing_scope_id`, never with legacy +`authentication`). A subscription without one is stored and fingerprinted +exactly as before, so existing subscriptions and queued deliveries are +unaffected. For a `web-bot-auth` subscription, the +`ReportingSigningMaterial` your `ReportingSigningResolver` returns must carry +the origin's key: + +```python +from adcp.reporting.outbox import ReportingSigningMaterial, ReportingWebBotAuthKey + +return ReportingSigningMaterial( + webhook_key, webhook_kid, "ed25519", frozenset({"ed25519"}), + web_bot_auth=ReportingWebBotAuthKey(wba_key, "https://seller.example"), +) +``` + +The worker builds a `from_wba` sender for each attempt of a `web-bot-auth` +subscription, and keeps the `adcp/webhook-signing/v1` key for every other +one. A `web-bot-auth` subscription whose material carries no Web Bot Auth key +is quarantined (`permanent_scope`). `ed25519` must be among +`advertised_algorithms` whenever `web_bot_auth` is set. Production reporting +opts in with `ReportingProductionSigning(..., web_bot_auth=True)`; see +[reporting production](../reporting-production.md). + +Rolling deploys: a pod on an earlier SDK quarantines a delivery stored for a +subscription with `signing_profile` (`integrity_failure`) rather than sign it +under the wrong profile, and its resolver cannot return such a subscription. +Deploy every reporting worker before your subscription resolver starts +setting the field. diff --git a/src/adcp/decisioning/dispatch.py b/src/adcp/decisioning/dispatch.py index f7c7fb80ed..409e8be74c 100644 --- a/src/adcp/decisioning/dispatch.py +++ b/src/adcp/decisioning/dispatch.py @@ -78,6 +78,7 @@ _extract_push_authentication, _extract_push_notification_url_and_token, _extract_push_operation_id, + _extract_push_signing_profile, emit_terminal_completion_webhook, ) from adcp.exceptions import ADCPTaskError @@ -1542,6 +1543,7 @@ async def _invoke_platform_method( webhook_target: WebhookDeliveryTarget | None = None, webhook_auto_emit: bool = True, webhook_external_owner_ready: bool = False, + webhook_signing_profiles: frozenset[str] | None = None, task_type: str | None = None, pre_handoff_reject: Callable[[], None] | None = None, sync_admission: SyncExecutorAdmission | None = None, @@ -1611,6 +1613,11 @@ async def _invoke_platform_method( 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 webhook_signing_profiles: The ``webhook_signing.profiles`` this + request's effective capabilities advertise (``adcp/webhook-signing/v1`` + when omitted). A push registration requiring another profile is + refused with ``INVALID_REQUEST`` before the task is issued. ``None`` + skips the check. :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; @@ -1869,6 +1876,7 @@ async def _invoke_platform_method( webhook_target=webhook_target, webhook_auto_emit=webhook_auto_emit, webhook_external_owner_ready=webhook_external_owner_ready, + webhook_signing_profiles=webhook_signing_profiles, allow_workflow_handoff=allow_workflow_handoff, ) @@ -1888,6 +1896,7 @@ async def _project_invocation_result( webhook_auto_emit: bool, webhook_external_owner_ready: bool, allow_workflow_handoff: bool, + webhook_signing_profiles: frozenset[str] | None = None, ) -> Any: """Project a raw adopter result and settle its framework lifecycle hooks.""" if is_task_handoff(result) or is_workflow_handoff(result): @@ -1923,6 +1932,7 @@ async def _project_invocation_result( webhook_target=webhook_target, webhook_auto_emit=webhook_auto_emit, webhook_external_owner_ready=webhook_external_owner_ready, + webhook_signing_profiles=webhook_signing_profiles, ) if is_workflow_handoff(result): return await _project_workflow_handoff( @@ -1934,6 +1944,7 @@ async def _project_invocation_result( request_params=request_params, webhook_auto_emit=webhook_auto_emit, webhook_external_owner_ready=webhook_external_owner_ready, + webhook_signing_profiles=webhook_signing_profiles, on_failure=on_failure, ) @@ -2111,6 +2122,70 @@ async def _safe_on_failure_call( ) +async def _push_signing_profile( + request_params: BaseModel | Any | None, + registry: Any, + *, + method_name: str, + on_failure: Callable[[BaseException], Awaitable[None]] | None, + advertised_profiles: frozenset[str] | None = None, +) -> str | None: + """The task push registration's ``signing_profile``, checked before issue. + + The profile it requires is the one it names, or ``adcp/webhook-signing/v1`` + when it names none and carries no legacy ``authentication``. Refuses with + ``INVALID_REQUEST`` before the task is issued when the registration is + malformed, when that profile is outside ``advertised_profiles`` (this + request's effective ``webhook_signing.profiles``, for every outbox), or + when a fixed-sender outbox holds no key for it. A ``sender_resolver`` + outbox's key is checked again at each delivery attempt. + """ + from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP + + rejection: AdcpError | None = None + profile: str | None = None + try: + profile = _extract_push_signing_profile(request_params) + except ValueError as exc: + rejection = AdcpError( + "INVALID_REQUEST", + message=str(exc), + recovery="correctable", + field="push_notification_config.signing_profile", + ) + else: + outbox = getattr(registry, "task_webhook_outbox", None) + supports = getattr(outbox, "supports_signing_profile", None) + required = ( + profile + if profile is not None + else ( + WEBHOOK_SIGNING_PROFILE_ADCP + if _extract_push_authentication(request_params) is None + else None + ) + ) + unadvertised = advertised_profiles is not None and required not in advertised_profiles + if required is not None and ( + unadvertised or (callable(supports) and supports(required) is False) + ): + rejection = AdcpError( + "INVALID_REQUEST", + message=(f"This seller does not sign task webhooks under {required!r}"), + recovery="correctable", + field="push_notification_config.signing_profile", + suggestion=( + "Select a signing_profile listed in the seller's " + "webhook_signing.profiles capability" + ), + ) + if rejection is None: + return profile + if on_failure is not None: + await _safe_on_failure_call(on_failure, rejection, method_name) + raise rejection + + async def _validate_handoff_push_ownership( *, method_name: str, @@ -2178,6 +2253,7 @@ async def _project_handoff( webhook_target: WebhookDeliveryTarget | None = None, webhook_auto_emit: bool = True, webhook_external_owner_ready: bool = False, + webhook_signing_profiles: frozenset[str] | None = None, ) -> dict[str, Any]: """Promote a TaskHandoff to a background task. @@ -2252,6 +2328,8 @@ async def _project_handoff( or external outbox owns terminal task delivery. :param webhook_external_owner_ready: Trusted configuration proof that the configured durable outbox owns this publication. + :param webhook_signing_profiles: Advertised webhook-signing profiles; see + :func:`_invoke_platform_method`. The handoff fn is extracted via the type-identity dispatch in :func:`adcp.decisioning.types.is_task_handoff`. Subclassed @@ -2281,6 +2359,17 @@ async def _project_handoff( "request_context": _extract_request_context(request_params), } push_target = _extract_push_notification_url_and_token(request_params) + push_signing_profile: str | None = None + if method_name in SPEC_WEBHOOK_TASK_TYPES and push_target is not None: + # Every publisher, SDK outbox or external, signs under the profile the + # registration selects, so an unadvertised one is refused here. + push_signing_profile = await _push_signing_profile( + request_params, + registry, + method_name=method_name, + on_failure=on_failure, + advertised_profiles=webhook_signing_profiles, + ) if ( method_name in SPEC_WEBHOOK_TASK_TYPES and push_target is not None @@ -2299,6 +2388,8 @@ async def _project_handoff( webhook_authentication=push_authentication, webhook_signing_scope_id=signing_scope_id, ) + if push_signing_profile is not None: + issue_kwargs["webhook_signing_profile"] = push_signing_profile task_id = await registry.issue(**issue_kwargs) # Hand off to background. The wire envelope returns immediately; @@ -2505,6 +2596,7 @@ async def _project_workflow_handoff( request_params: BaseModel | Any | None = None, webhook_auto_emit: bool = True, webhook_external_owner_ready: bool = False, + webhook_signing_profiles: frozenset[str] | None = None, on_failure: Callable[[BaseException], Awaitable[None]] | None = None, ) -> dict[str, Any]: """Project a :class:`WorkflowHandoff` to the wire Submitted envelope. @@ -2556,6 +2648,17 @@ async def _project_workflow_handoff( "request_context": _extract_request_context(request_params), } push_target = _extract_push_notification_url_and_token(request_params) + push_signing_profile: str | None = None + if method_name in SPEC_WEBHOOK_TASK_TYPES and push_target is not None: + # Every publisher, SDK outbox or external, signs under the profile the + # registration selects, so an unadvertised one is refused here. + push_signing_profile = await _push_signing_profile( + request_params, + registry, + method_name=method_name, + on_failure=on_failure, + advertised_profiles=webhook_signing_profiles, + ) if ( method_name in SPEC_WEBHOOK_TASK_TYPES and push_target is not None @@ -2574,6 +2677,8 @@ async def _project_workflow_handoff( webhook_authentication=push_authentication, webhook_signing_scope_id=signing_scope_id, ) + if push_signing_profile is not None: + issue_kwargs["webhook_signing_profile"] = push_signing_profile task_id = await registry.issue(**issue_kwargs) handoff_ctx = TaskHandoffContext(id=task_id, _registry=registry) diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index a97d78e8c9..7af6b43dd7 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -120,6 +120,7 @@ ) from adcp.decisioning.webhook_emit import ( _extract_push_notification_url_and_token, + advertised_webhook_signing_profiles, maybe_emit_sync_completion, task_webhook_owner_ready, ) @@ -2072,6 +2073,10 @@ async def _handoff_webhook_kwargs( 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. + + ``webhook_signing_profiles`` comes from this request's effective + capabilities (a tenant-scoped override included): dispatch refuses a + push registration selecting a profile outside it. """ capabilities = self._platform.capabilities if _extract_push_notification_url_and_token(params) is not None: @@ -2108,6 +2113,7 @@ async def _handoff_webhook_kwargs( # 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, + "webhook_signing_profiles": advertised_webhook_signing_profiles(capabilities), } def _build_ctx( diff --git a/src/adcp/decisioning/pg/lazy.py b/src/adcp/decisioning/pg/lazy.py index 542a3b1c17..bfbf1e2ac3 100644 --- a/src/adcp/decisioning/pg/lazy.py +++ b/src/adcp/decisioning/pg/lazy.py @@ -396,6 +396,7 @@ async def enqueue_terminal( token: str | None, authentication: TaskWebhookAuthentication | None = None, signing_scope_id: str | None = None, + signing_profile: str | None = None, ) -> int: return await (await self.resolve()).enqueue_terminal( conn, @@ -409,12 +410,23 @@ async def enqueue_terminal( token=token, authentication=authentication, signing_scope_id=signing_scope_id, + signing_profile=signing_profile, ) def validate_registration( - self, url: str, authentication: TaskWebhookAuthentication | None = None + self, + url: str, + authentication: TaskWebhookAuthentication | None = None, + *, + signing_profile: str | None = None, ) -> None: - (self._require_resolved()).validate_registration(url, authentication) + (self._require_resolved()).validate_registration( + url, authentication, signing_profile=signing_profile + ) + + def supports_signing_profile(self, profile: str) -> bool | None: + store = self.resolved + return store.supports_signing_profile(profile) if store is not None else None def protect_registration( self, @@ -427,6 +439,7 @@ def protect_registration( token: str | None, authentication: TaskWebhookAuthentication | None = None, signing_scope_id: str | None = None, + signing_profile: str | None = None, ) -> tuple[bytes, bytes]: return (self._require_resolved()).protect_registration( account_id=account_id, @@ -437,6 +450,7 @@ def protect_registration( token=token, authentication=authentication, signing_scope_id=signing_scope_id, + signing_profile=signing_profile, ) def open_registration( diff --git a/src/adcp/decisioning/pg/task_registry.py b/src/adcp/decisioning/pg/task_registry.py index d0a6621ba0..4b64608bf0 100644 --- a/src/adcp/decisioning/pg/task_registry.py +++ b/src/adcp/decisioning/pg/task_registry.py @@ -270,10 +270,16 @@ async def issue( webhook_token: str | None = None, webhook_authentication: TaskWebhookAuthentication | None = None, webhook_signing_scope_id: str | None = None, + webhook_signing_profile: str | None = None, **_extra: Any, ) -> str: """Allocate a task_id, persist a ``submitted`` row, return the id. + ``webhook_signing_profile`` is the registration's ``signing_profile`` + (AdCP 3.3); the terminal webhook is signed under it, or under + ``adcp/webhook-signing/v1`` when it is ``None`` and no legacy + ``webhook_authentication`` was registered. + Mirrors :meth:`~adcp.decisioning.InMemoryTaskRegistry.issue` including the account_id validation guard — empty or sentinel account_ids would allow cross-tenant task-id probing via the ``WHERE account_id = %s`` @@ -300,11 +306,15 @@ async def issue( raise ValueError( "legacy webhook_authentication must not carry an RFC 9421 signing scope" ) + if webhook_url is None and webhook_signing_profile is not None: + raise ValueError("webhook_signing_profile requires webhook_url") 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, webhook_authentication) + outbox.validate_registration( + webhook_url, webhook_authentication, signing_profile=webhook_signing_profile + ) task_id = f"task_{uuid.uuid4().hex[:16]}" encrypted_registration: bytes | None = None registration_nonce: bytes | None = None @@ -320,6 +330,7 @@ async def issue( token=webhook_token, authentication=webhook_authentication, signing_scope_id=webhook_signing_scope_id, + signing_profile=webhook_signing_profile, ) now = time.time() async with self._pool.connection() as conn: @@ -616,7 +627,7 @@ async def _enqueue_terminal_if_registered( ) if registration_nonce is None: raise RuntimeError(f"Task {task_id!r} has incomplete webhook registration") - url, operation_id, token, authentication, signing_scope_id = ( + url, operation_id, token, authentication, signing_scope_id, signing_profile = ( self.task_webhook_outbox._open_registration_with_scope( account_id=account_id, task_id=task_id, @@ -637,6 +648,7 @@ async def _enqueue_terminal_if_registered( token=token, authentication=authentication, signing_scope_id=signing_scope_id, + signing_profile=signing_profile, ) # The encrypted outbox envelope now owns the callback registration. # Clear the task-row copy in this same transaction. diff --git a/src/adcp/decisioning/pg/task_webhook_outbox.py b/src/adcp/decisioning/pg/task_webhook_outbox.py index bd5f4c74d4..2d51dc3d88 100644 --- a/src/adcp/decisioning/pg/task_webhook_outbox.py +++ b/src/adcp/decisioning/pg/task_webhook_outbox.py @@ -39,6 +39,7 @@ WebhookSenderResolver, ) from adcp.webhook_supervisor import RetryPolicy +from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILES if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool @@ -164,14 +165,14 @@ def __init__( self._worker_started = False self._sql_insert = ( # noqa: S608 - f"INSERT INTO {table} (" + f"INSERT INTO {table} (" # nosec B608 "task_id, task_type, terminal_status, url, operation_id, " "idempotency_key, signing_scope_id, account_id, encrypted_body, envelope_nonce, " "retry_horizon_seconds" ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id" ) self._sql_expire = ( # noqa: S608 - f"WITH expired AS (SELECT id FROM {table}" + f"WITH expired AS (SELECT id FROM {table}" # nosec B608 " 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," @@ -179,7 +180,7 @@ def __init__( " FROM expired WHERE outbox.id = expired.id" ) self._sql_claim = ( # noqa: S608 - f"WITH candidate AS (" + f"WITH candidate AS (" # nosec B608 f" SELECT id FROM {table}" " WHERE (retry_until IS NULL OR retry_until > now()) AND (" " (state = 'pending' AND available_at <= now()) OR" @@ -201,13 +202,13 @@ def __init__( " outbox.attempt_count" ) self._sql_ack = ( # noqa: S608 - f"UPDATE {table} SET state = 'delivered', delivered_at = now()," + f"UPDATE {table} SET state = 'delivered', delivered_at = now()," # nosec B608 " 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" + f"UPDATE {table} SET" # nosec B608 " 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," @@ -216,12 +217,12 @@ def __init__( " 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," + f"UPDATE {table} SET state = 'invalid', lease_token = NULL," # nosec B608 " 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"DELETE FROM {table} WHERE id IN (" # nosec B608 f" SELECT id FROM {table} WHERE retry_until <= now()" " AND state IN ('delivered', 'expired', 'invalid')" " ORDER BY id LIMIT 1000" @@ -290,11 +291,18 @@ async def enqueue_terminal( token: str | None, authentication: TaskWebhookAuthentication | None = None, signing_scope_id: str | None = None, + signing_profile: str | None = None, ) -> int: - """Insert a terminal webhook using the caller's open transaction.""" + """Insert a terminal webhook using the caller's open transaction. + + ``signing_profile`` is the registration's ``signing_profile``; every + attempt is signed under it (``adcp/webhook-signing/v1`` when ``None`` + and ``authentication`` is absent). + """ if status not in {"completed", "failed"}: raise ValueError(f"terminal webhook status must be completed or failed, got {status!r}") self._validate_authentication(authentication) + self._validate_signing_profile(signing_profile, authentication) self._validate_scope_for_mode( signing_scope_id, authentication=authentication, @@ -322,7 +330,7 @@ async def enqueue_terminal( idempotency_key=prepared.idempotency_key, signing_scope_id=signing_scope_id, ) - protected_body = self._protect_delivery_body(prepared.body, authentication) + protected_body = self._protect_delivery_body(prepared.body, authentication, signing_profile) encrypted_body = self._cipher.encrypt(nonce, protected_body, aad) cursor = await conn.execute( self._sql_insert, @@ -349,10 +357,23 @@ def validate_registration( self, url: str, authentication: TaskWebhookAuthentication | None = None, + *, + signing_profile: str | None = None, ) -> None: """Validate callback syntax before a task is accepted as Submitted.""" self._validate_callback_url(url) self._validate_authentication(authentication) + self._validate_signing_profile(signing_profile, authentication) + + def supports_signing_profile(self, profile: str) -> bool | None: + """Whether deliveries can be signed under ``profile``. + + ``None`` for a ``sender_resolver`` outbox: its sender is resolved per + signing scope at each attempt, so the check happens there. + """ + if self._sender is None: + return None + return profile in self._sender.signing_profiles def protect_registration( self, @@ -365,31 +386,38 @@ def protect_registration( token: str | None, authentication: TaskWebhookAuthentication | None = None, signing_scope_id: str | None = None, + signing_profile: str | None = None, ) -> tuple[bytes, bytes]: """Encrypt and authenticate callback registration at task issue time.""" self._validate_callback_url(url) self._validate_authentication(authentication) + self._validate_signing_profile(signing_profile, authentication) self._validate_scope_for_mode( signing_scope_id, authentication=authentication, require_resolver_scope=True, ) nonce = os.urandom(12) + registration: dict[str, Any] = { + "url": url, + "operation_id": operation_id, + "token": token, + "authentication": ( + { + "scheme": authentication.scheme, + "credentials": authentication.credentials, + } + if authentication is not None + else None + ), + "signing_scope_id": signing_scope_id, + } + if signing_profile not in (None, WEBHOOK_SIGNING_PROFILE_ADCP): + # The default is implied by its absence, keeping such + # registrations identical to the ones stored before profiles. + registration["signing_profile"] = signing_profile plaintext = json.dumps( - { - "url": url, - "operation_id": operation_id, - "token": token, - "authentication": ( - { - "scheme": authentication.scheme, - "credentials": authentication.credentials, - } - if authentication is not None - else None - ), - "signing_scope_id": signing_scope_id, - }, + registration, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") @@ -420,7 +448,7 @@ def open_registration( The three-item return shape is retained for compatibility. Durable registry dispatch uses the private scope-aware decoder below. """ - url, operation_id, token, _authentication, _signing_scope_id = ( + url, operation_id, token, _authentication, _signing_scope_id, _signing_profile = ( self._open_registration_with_scope( account_id=account_id, task_id=task_id, @@ -445,8 +473,9 @@ def _open_registration_with_scope( str | None, TaskWebhookAuthentication | None, str | None, + str | None, ]: - """Verify and decrypt callback registration with its trusted scope.""" + """Verify and decrypt callback registration with its scope and signing profile.""" try: plaintext = self._cipher.decrypt( nonce, @@ -467,6 +496,7 @@ def _open_registration_with_scope( token = value.get("token") authentication_value = value.get("authentication") signing_scope_id = value.get("signing_scope_id") + signing_profile = value.get("signing_profile") 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): @@ -495,7 +525,8 @@ def _open_registration_with_scope( authentication=authentication, require_resolver_scope=False, ) - return url, operation_id, token, authentication, signing_scope_id + self._validate_signing_profile(signing_profile, authentication) + return url, operation_id, token, authentication, signing_scope_id, signing_profile async def run_worker( self, @@ -568,7 +599,7 @@ async def process_one(self) -> bool: ) try: protected_body = self._cipher.decrypt(bytes(nonce), bytes(encrypted_body), aad) - body_bytes, authentication = self._open_delivery_body(protected_body) + body_bytes, authentication, signing_profile = self._open_delivery_body(protected_body) self._validate_stored_body( body_bytes, task_id=str(task_id), @@ -596,6 +627,15 @@ async def process_one(self) -> bool: url=str(url), idempotency_key=str(idempotency_key), body=body_bytes, + # A registration without legacy authentication requires its + # signing_profile, adcp/webhook-signing/v1 when it named none + # (every row stored before profiles were recorded). The stored + # bytes carry no signature: each attempt is signed afresh. + required_signing_profile=( + (signing_profile or WEBHOOK_SIGNING_PROFILE_ADCP) + if authentication is None + else None + ), ) delivery: WebhookDeliveryResult | None = None error: BaseException | None = None @@ -724,7 +764,7 @@ def _retry_delay(self, attempt_count: int) -> float: ) ) if self._retry.jitter: - delay *= 0.5 + random.random() * 0.5 + delay *= 0.5 + random.random() * 0.5 # nosec B311 - retry jitter, not security return delay @staticmethod @@ -799,8 +839,8 @@ async def _resolve_delivery_sender(self, signing_scope_id: str | None) -> Webhoo raise ValueError("resolver returned an invalid sender resolution") sender = resolution.sender self._validate_delivery_sender(sender) - sender_algorithm = getattr(getattr(sender, "_auth", None), "alg", None) - if sender_algorithm not in resolution.advertised_algorithms: + sender_algorithms = sender.signing_algorithms + if not sender_algorithms or not sender_algorithms <= resolution.advertised_algorithms: raise ValueError("resolved sender algorithm was not advertised for its scope") sender_timeout = float(getattr(sender, "_timeout", 0.0)) if self._lease_seconds < sender_timeout + 5: @@ -874,58 +914,82 @@ def _validate_authentication( def _protect_delivery_body( body: bytes, authentication: TaskWebhookAuthentication | None, + signing_profile: str | None = None, ) -> bytes: - """Keep old RFC rows byte-compatible; wrap encrypted legacy secrets.""" - if authentication is None: + """Keep default-profile RFC rows byte-compatible; wrap legacy secrets or a profile.""" + if signing_profile == WEBHOOK_SIGNING_PROFILE_ADCP: + # A bare body already means adcp/webhook-signing/v1, and an + # earlier SDK can still deliver it. + signing_profile = None + if authentication is None and signing_profile is None: return body - return json.dumps( - { - "task_webhook_delivery_version": _ENCRYPTED_DELIVERY_VERSION, - "body": base64.b64encode(body).decode("ascii"), - "authentication": { - "scheme": authentication.scheme, - "credentials": authentication.credentials, - }, - }, - ensure_ascii=False, - separators=(",", ":"), - ).encode("utf-8") + envelope: dict[str, Any] = { + "task_webhook_delivery_version": _ENCRYPTED_DELIVERY_VERSION, + "body": base64.b64encode(body).decode("ascii"), + } + if authentication is not None: + envelope["authentication"] = { + "scheme": authentication.scheme, + "credentials": authentication.credentials, + } + else: + envelope["signing_profile"] = signing_profile + return json.dumps(envelope, ensure_ascii=False, separators=(",", ":")).encode("utf-8") def _open_delivery_body( self, protected_body: bytes, - ) -> tuple[bytes, TaskWebhookAuthentication | None]: - """Decode a legacy-auth envelope or accept a pre-feature RFC body.""" + ) -> tuple[bytes, TaskWebhookAuthentication | None, str | None]: + """Decode a legacy-auth or signing-profile envelope, or accept a bare RFC body.""" try: value = json.loads(protected_body) except (json.JSONDecodeError, UnicodeDecodeError): - return protected_body, None + return protected_body, None, None if not isinstance(value, dict) or "task_webhook_delivery_version" not in value: - return protected_body, None + return protected_body, None, None if value.get("task_webhook_delivery_version") != _ENCRYPTED_DELIVERY_VERSION: raise ValueError("unsupported encrypted task webhook delivery version") - authentication_value = value.get("authentication") - if not isinstance(authentication_value, dict): - raise ValueError("encrypted task webhook authentication is missing") - scheme = authentication_value.get("scheme") - credentials = authentication_value.get("credentials") encoded_body = value.get("body") - if ( - not isinstance(scheme, str) - or not isinstance(credentials, str) - or not isinstance(encoded_body, str) - ): + if not isinstance(encoded_body, str): raise ValueError("encrypted task webhook delivery envelope is invalid") + authentication_value = value.get("authentication") + signing_profile = value.get("signing_profile") + if (authentication_value is None) == (signing_profile is None): + raise ValueError("encrypted task webhook authentication is missing") + authentication: TaskWebhookAuthentication | None = None + if authentication_value is not None: + if not isinstance(authentication_value, dict): + raise ValueError("encrypted task webhook delivery envelope is invalid") + scheme = authentication_value.get("scheme") + credentials = authentication_value.get("credentials") + if not isinstance(scheme, str) or not isinstance(credentials, str): + raise ValueError("encrypted task webhook delivery envelope is invalid") + try: + authentication = TaskWebhookAuthentication( + scheme=scheme, + credentials=credentials, + ) + except (TypeError, ValueError) as exc: + raise ValueError("encrypted task webhook delivery envelope is invalid") from exc try: - authentication = TaskWebhookAuthentication( - scheme=scheme, - credentials=credentials, - ) body = base64.b64decode(encoded_body, validate=True) - except (TypeError, ValueError) as exc: + except ValueError as exc: raise ValueError("encrypted task webhook delivery envelope is invalid") from exc self._validate_authentication(authentication) - return body, authentication + self._validate_signing_profile(signing_profile, authentication) + return body, authentication, signing_profile + + @staticmethod + def _validate_signing_profile( + signing_profile: object, + authentication: TaskWebhookAuthentication | None, + ) -> None: + if signing_profile is None: + return + if signing_profile not in WEBHOOK_SIGNING_PROFILES: + raise ValueError(f"unknown webhook signing_profile {signing_profile!r}") + if authentication is not None: + raise ValueError("legacy webhook authentication must not carry a signing_profile") @staticmethod def _envelope_aad( diff --git a/src/adcp/decisioning/webhook_emit.py b/src/adcp/decisioning/webhook_emit.py index 660a1bad93..417b361448 100644 --- a/src/adcp/decisioning/webhook_emit.py +++ b/src/adcp/decisioning/webhook_emit.py @@ -205,6 +205,39 @@ def _extract_push_authentication(params: Any) -> TaskWebhookAuthentication | Non return TaskWebhookAuthentication(scheme=raw_scheme, credentials=credentials) +def _extract_push_signing_profile(params: Any) -> str | None: + """Extract the webhook-signing profile a task push registration names (AdCP 3.3). + + ``None`` when ``push_notification_config`` names no ``signing_profile``; + a registration that also omits legacy ``authentication`` then requires + ``adcp/webhook-signing/v1``. Raises :class:`ValueError` for an unknown + profile, or one combined with ``authentication``. + """ + from adcp.webhooks import WEBHOOK_SIGNING_PROFILES + + config = getattr(params, "push_notification_config", None) + if config is None and isinstance(params, dict): + config = params.get("push_notification_config") + if config is None: + return None + if isinstance(config, dict): + profile = config.get("signing_profile") + authentication = config.get("authentication") + else: + profile = getattr(config, "signing_profile", None) + authentication = getattr(config, "authentication", None) + if profile is None: + return None + profile = getattr(profile, "value", profile) + if authentication is not None: + raise ValueError( + "push_notification_config must not combine authentication with signing_profile" + ) + if profile not in WEBHOOK_SIGNING_PROFILES: + raise ValueError(f"unknown push_notification_config.signing_profile {profile!r}") + return str(profile) + + def maybe_emit_sync_completion( *, sender: WebhookSender | None, @@ -688,9 +721,11 @@ def validate_webhook_signing_for_capabilities( # skip the cross-check when omitted (no advertisement to violate). advertised_algorithms = getattr(webhook_signing, "algorithms", None) if advertised_algorithms and sender_introspectable: - sender_alg = getattr(getattr(resolved_sender, "_auth", None), "alg", None) + sender_algs = _sender_algorithms(resolved_sender) advertised_alg_values = [getattr(a, "value", a) for a in advertised_algorithms] - if sender_alg not in advertised_alg_values: + unadvertised = sorted(alg for alg in sender_algs if alg not in advertised_alg_values) + if not sender_algs or unadvertised: + sender_alg = unadvertised[0] if unadvertised else None raise AdcpError( "INVALID_REQUEST", message=( @@ -711,6 +746,37 @@ def validate_webhook_signing_for_capabilities( }, ) + # Every advertised profile is one a registration may select, and the + # sender must hold a key for each. Omission means adcp/webhook-signing/v1, + # so a web-bot-auth-only sender must advertise its profile. + from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP + + advertised_profiles = getattr(webhook_signing, "profiles", None) or [ + WEBHOOK_SIGNING_PROFILE_ADCP + ] + sender_profiles = getattr(resolved_sender, "signing_profiles", None) + if sender_introspectable and isinstance(sender_profiles, frozenset): + advertised_profile_values = [getattr(p, "value", p) for p in advertised_profiles] + missing_profiles = sorted(p for p in advertised_profile_values if p not in sender_profiles) + if missing_profiles: + raise AdcpError( + "INVALID_REQUEST", + message=( + "capabilities.webhook_signing.profiles advertises " + f"{advertised_profile_values!r} but the wired WebhookSender " + f"holds no key for {missing_profiles!r}. A registration " + "selecting that profile could never be delivered. Add the " + "key (WebhookSender.from_wba, or with_web_bot_auth on a " + "from_jwk / from_pem sender), or stop advertising the profile." + ), + recovery="terminal", + details={ + "missing": "webhook_signing_profile_key", + "advertised_profiles": advertised_profile_values, + "sender_profiles": sorted(sender_profiles), + }, + ) + if internal_outbox_ready: return @@ -728,6 +794,31 @@ def validate_webhook_signing_for_capabilities( ) +def advertised_webhook_signing_profiles(capabilities: Any) -> frozenset[str] | None: + """The webhook-signing profiles a seller's capabilities advertise. + + ``webhook_signing.profiles``, or ``adcp/webhook-signing/v1`` when it is + omitted. ``None`` when the capabilities declare no webhook signing, so + there is no advertisement to check a registration against. + """ + from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP + + webhook_signing = getattr(capabilities, "webhook_signing", None) + if webhook_signing is None or getattr(webhook_signing, "supported", False) is not True: + return None + profiles = getattr(webhook_signing, "profiles", None) or [WEBHOOK_SIGNING_PROFILE_ADCP] + return frozenset(str(getattr(profile, "value", profile)) for profile in profiles) + + +def _sender_algorithms(sender: Any) -> frozenset[str]: + """Every ``alg`` a sender's signatures may carry; duck-typed for supervisors.""" + algorithms = getattr(sender, "signing_algorithms", None) + if isinstance(algorithms, frozenset): + return algorithms + alg = getattr(getattr(sender, "_auth", None), "alg", None) + return frozenset({alg}) if isinstance(alg, str) else frozenset() + + def external_task_webhook_owner_ready( *, capabilities: DecisioningCapabilities, diff --git a/src/adcp/notification_outbox_pg.py b/src/adcp/notification_outbox_pg.py index a040315310..412e963fa0 100644 --- a/src/adcp/notification_outbox_pg.py +++ b/src/adcp/notification_outbox_pg.py @@ -41,6 +41,7 @@ WebhookSenderResolver, ) from adcp.webhook_supervisor import RetryPolicy +from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILES if TYPE_CHECKING: from psycopg import AsyncConnection, Connection @@ -415,6 +416,7 @@ def _protected(prepared: PreparedWebhook) -> bytes: "signature", "signature-input", "content-digest", + "signature-agent", } ), ) @@ -423,14 +425,25 @@ def _protected(prepared: PreparedWebhook) -> bytes: c in value for c in ("\r", "\n", "\x00") ): raise ValueError("invalid notification HTTP headers") + profile = prepared.required_signing_profile + if profile is not None and profile not in WEBHOOK_SIGNING_PROFILES: + raise ValueError(f"unknown webhook signing profile {profile!r}") # Credentials/headers are authenticated and encrypted along with the body. - return _canonical( - { - "version": 1, - "body": base64.b64encode(prepared.body).decode("ascii"), - "extra_headers": headers, - } - ) + envelope: dict[str, Any] = { + "version": 1, + "body": base64.b64encode(prepared.body).decode("ascii"), + "extra_headers": headers, + } + if profile is not None: + # The registration's profile binds every attempt; it is part of + # the immutable delivery, so a replayed enqueue must match it. + envelope["required_signing_profile"] = profile + if profile != WEBHOOK_SIGNING_PROFILE_ADCP: + # Version 2: an earlier SDK, whose only profile is the default, + # refuses (quarantines) the row rather than ignore the profile. + # An adcp/webhook-signing/v1 row stays readable to it. + envelope["version"] = 2 + return _canonical(envelope) def _prepare_insert( self, @@ -613,7 +626,7 @@ def _open(self, row: tuple[Any, ...]) -> PreparedWebhook: if cipher is None: raise ValueError("notification encryption key is unavailable") value = json.loads(cipher.decrypt(bytes(nonce), bytes(ciphertext), aad)) - if not isinstance(value, dict) or value.get("version") != 1: + if not isinstance(value, dict) or value.get("version") not in (1, 2): raise ValueError("invalid encrypted notification envelope") body = base64.b64decode(value["body"], validate=True) headers = value.get("extra_headers") @@ -621,6 +634,11 @@ def _open(self, row: tuple[Any, ...]) -> PreparedWebhook: not isinstance(k, str) or not isinstance(v, str) for k, v in headers.items() ): raise ValueError("invalid stored notification headers") + profile = value.get("required_signing_profile") + if (value["version"] == 2) != ( + profile is not None and profile != WEBHOOK_SIGNING_PROFILE_ADCP + ) or (profile is not None and profile not in WEBHOOK_SIGNING_PROFILES): + raise ValueError("invalid stored notification signing profile") validate_notification_body( body, idempotency_key=key, @@ -628,7 +646,13 @@ def _open(self, row: tuple[Any, ...]) -> PreparedWebhook: account_id=account, experimental_agent_identity=self._experimental_agent_identity, ) - return PreparedWebhook(url=url, idempotency_key=key, body=body, extra_headers=headers) + return PreparedWebhook( + url=url, + idempotency_key=key, + body=body, + extra_headers=headers, + required_signing_profile=profile, + ) def _validate_sender(self, sender: WebhookSender) -> None: if ( @@ -661,10 +685,8 @@ async def _resolve_sender(self, scope: str | None) -> WebhookSender: if not isinstance(resolution, WebhookSenderResolution): raise ValueError("invalid sender resolution") self._validate_sender(resolution.sender) - if ( - getattr(getattr(resolution.sender, "_auth", None), "alg", None) - not in resolution.advertised_algorithms - ): + algorithms = resolution.sender.signing_algorithms + if not algorithms or not algorithms <= resolution.advertised_algorithms: raise ValueError("resolved signing algorithm is not advertised") except Exception: raise ScopePermanentlyUnknown from None diff --git a/src/adcp/reporting/ledger/models.py b/src/adcp/reporting/ledger/models.py index b522f48945..a0437e7a0e 100644 --- a/src/adcp/reporting/ledger/models.py +++ b/src/adcp/reporting/ledger/models.py @@ -354,6 +354,10 @@ def _ledger_dict(pairs: list[tuple[str, Any]]) -> dict[str, Any]: # A subscription binding's path_template; account bindings never set # it, so their fingerprints and stored payloads are unchanged. del value["path_template"] + if "proof_of_control_ref" in value and value.get("signing_profile", "") is None: + # A notification subscription that selected no signing profile: left + # out so subscriptions stored before the field keep their fingerprint. + del value["signing_profile"] return value diff --git a/src/adcp/reporting/outbox/__init__.py b/src/adcp/reporting/outbox/__init__.py index 1987055e69..61be5ca99c 100644 --- a/src/adcp/reporting/outbox/__init__.py +++ b/src/adcp/reporting/outbox/__init__.py @@ -45,6 +45,7 @@ ReportingSigningMaterial, ReportingSigningResolver, ReportingSubscriptionResolver, + ReportingWebBotAuthKey, ) from adcp.reporting.outbox.status import ( ReportingStatusProjector, @@ -75,6 +76,7 @@ "ReportingDurableNotificationLedger", "ReportingDurableStatusStore", "ReportingStatusSupport", + "ReportingWebBotAuthKey", "StatusChanged", "StatusBoundary", "StatusCheckpoint", diff --git a/src/adcp/reporting/outbox/routing.py b/src/adcp/reporting/outbox/routing.py index 26eb91f4db..3430291a2f 100644 --- a/src/adcp/reporting/outbox/routing.py +++ b/src/adcp/reporting/outbox/routing.py @@ -29,6 +29,7 @@ from adcp.reporting.outbox.models import DeliveryBinding, StoredDelivery from adcp.signing.crypto import ALG_ED25519, ALG_ES256, ALLOWED_ALGS, PrivateKey from adcp.webhook_sender import PreparedWebhook +from adcp.webhooks import WEBHOOK_SIGNING_PROFILES @dataclass(frozen=True) @@ -67,6 +68,11 @@ class ReportingNotificationSubscription: proof_of_control_ref: str signing_scope_id: str | None = None authentication: ReportingLegacyAuthentication | None = field(default=None, repr=False) + #: The registration's ``signing_profile`` (AdCP 3.3), for a subscription + #: without legacy ``authentication``. ``None`` means + #: ``adcp/webhook-signing/v1``; it is left out of the fingerprint, so + #: subscriptions and deliveries stored before this field are unchanged. + signing_profile: str | None = None active: bool = field(kw_only=True) authorized: bool = field(kw_only=True) proof_valid: bool = field(kw_only=True) @@ -102,6 +108,11 @@ def __post_init__(self) -> None: and type(self.authentication) is not ReportingLegacyAuthentication ): raise ValueError + if self.signing_profile is not None and ( + self.signing_profile not in WEBHOOK_SIGNING_PROFILES + or self.authentication is not None + ): + raise ValueError if type(self.url) is not str or not self.url.isprintable() or len(self.url) > 8192: raise ValueError parsed = httpx.URL(self.url) @@ -163,17 +174,55 @@ async def get_active( ) -> ReportingNotificationSubscription | None: ... +@dataclass(frozen=True) +class ReportingWebBotAuthKey: + """The seller origin's Web Bot Auth key, for ``web-bot-auth`` subscriptions. + + Receivers verify it against the key directory at + ``{agent_origin}/.well-known/http-message-signatures-directory``; see + ``docs/webhooks/web-bot-auth.md``. + """ + + private_key: ed25519.Ed25519PrivateKey = field(repr=False) + agent_origin: str + + def __post_init__(self) -> None: + from adcp.signing.wba import _https_origin + + try: + if not isinstance(self.private_key, ed25519.Ed25519PrivateKey): + raise ValueError + object.__setattr__(self, "agent_origin", _https_origin(self.agent_origin)) + except (ValueError, TypeError, AttributeError): + pass + else: + return + raise ReportingNotificationError("permanent_scope") + + @dataclass(frozen=True) class ReportingSigningMaterial: - """Trusted current key material, not an adopter-supplied HTTP sender.""" + """Trusted current key material, not an adopter-supplied HTTP sender. + + ``private_key`` signs under ``adcp/webhook-signing/v1``, which every 3.3 + seller signing webhooks supports. ``web_bot_auth`` adds the origin's Web + Bot Auth key for subscriptions selecting ``web-bot-auth``; it signs with + ``ed25519``, so that algorithm must be advertised too. + """ private_key: PrivateKey = field(repr=False) key_id: str algorithm: str advertised_algorithms: frozenset[str] + web_bot_auth: ReportingWebBotAuthKey | None = field(default=None, kw_only=True, repr=False) def __post_init__(self) -> None: object.__setattr__(self, "advertised_algorithms", frozenset(self.advertised_algorithms)) + if self.web_bot_auth is not None and ( + type(self.web_bot_auth) is not ReportingWebBotAuthKey + or ALG_ED25519 not in self.advertised_algorithms + ): + raise ReportingNotificationError("permanent_scope") key_matches = ( self.algorithm == ALG_ED25519 and isinstance(self.private_key, ed25519.Ed25519PrivateKey) diff --git a/src/adcp/reporting/outbox/worker.py b/src/adcp/reporting/outbox/worker.py index 6e1ebcb7a7..1a0696f810 100644 --- a/src/adcp/reporting/outbox/worker.py +++ b/src/adcp/reporting/outbox/worker.py @@ -44,6 +44,7 @@ ScopeTransientlyUnavailable, WebhookSender, ) +from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH if TYPE_CHECKING: from adcp.reporting.ledger.delivery_models import ReportingDeliveryScope @@ -434,6 +435,18 @@ async def _sender(self, subscription: ReportingNotificationSubscription) -> Webh raise ScopePermanentlyUnknown from None try: ReportingSigningMaterial.__post_init__(material) + if subscription.signing_profile == WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH: + # Signed afresh per attempt (new nonce); a missing key is a + # configuration error retries cannot repair. + web_bot_auth = material.web_bot_auth + if web_bot_auth is None: + raise ValueError("no web-bot-auth key for this signing scope") + return WebhookSender.from_wba( + web_bot_auth.private_key, + agent_origin=web_bot_auth.agent_origin, + timeout_seconds=timeout, + allowed_destination_ports=frozenset({443}), + ) return WebhookSender( private_key=material.private_key, key_id=material.key_id, diff --git a/src/adcp/reporting/production/notifications.py b/src/adcp/reporting/production/notifications.py index 768cd4a1e3..93a772d5b6 100644 --- a/src/adcp/reporting/production/notifications.py +++ b/src/adcp/reporting/production/notifications.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from dataclasses import dataclass, field from datetime import datetime from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable @@ -25,7 +26,10 @@ ReportingProductionProjection, ReportingProductionStore, ) -from adcp.signing.crypto import ALLOWED_ALGS +from adcp.signing.crypto import ALG_ED25519, ALLOWED_ALGS +from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH + +logger = logging.getLogger(__name__) if TYPE_CHECKING: from adcp.reporting.production.service import ReportingProductionSupport @@ -43,6 +47,10 @@ class ReportingProductionSigning: resolver: ReportingSigningResolver = field(repr=False) algorithms: tuple[str, ...] brand_json_url: str = field(kw_only=True) + #: Also sign subscriptions that select ``web-bot-auth`` (AdCP 3.3). Every + #: resolved material must then carry ``web_bot_auth``, ``ed25519`` must be + #: advertised, and the capability lists both profiles. + web_bot_auth: bool = field(default=False, kw_only=True) def __post_init__(self) -> None: object.__setattr__(self, "algorithms", tuple(self.algorithms)) @@ -51,6 +59,8 @@ def __post_init__(self) -> None: or len(set(self.algorithms)) != len(self.algorithms) or not set(self.algorithms) <= ALLOWED_ALGS or not callable(getattr(self.resolver, "resolve", None)) + or type(self.web_bot_auth) is not bool + or (self.web_bot_auth and ALG_ED25519 not in self.algorithms) ): raise ReportingNotificationError("notification_signing_unready") try: @@ -82,22 +92,28 @@ async def resolve( self.algorithms ): raise ReportingNotificationError("notification_signing_unready") + if self.web_bot_auth and material.web_bot_auth is None: + raise ReportingNotificationError("notification_signing_unready") ReportingSigningMaterial.__post_init__(material) return material def wire(self) -> dict[str, Any]: - return { + wire: dict[str, Any] = { "supported": True, - "profile": "adcp/webhook-signing/v1", + "profile": WEBHOOK_SIGNING_PROFILE_ADCP, "algorithms": list(self.algorithms), "legacy_hmac_fallback": False, "delivery_retry_horizon_seconds": RETRY_HORIZON_SECONDS, } + if self.web_bot_auth: + wire["profiles"] = [WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH] + return wire @dataclass(frozen=True) class _SignedSubscriptions: resolver: ReportingSubscriptionResolver = field(repr=False) + web_bot_auth: bool = False @staticmethod def _check(value: ReportingNotificationSubscription) -> ReportingNotificationSubscription: @@ -105,13 +121,29 @@ def _check(value: ReportingNotificationSubscription) -> ReportingNotificationSub raise ReportingNotificationError("notification_signing_unready") return value + def _signable(self, value: ReportingNotificationSubscription) -> bool: + """Whether this seller signs the profile the subscription selected. + + A ``web-bot-auth`` subscription without ``web_bot_auth`` signing is + skipped, not raised: one such subscriber must not stop deliveries to + every other subscriber of the account. The registration service + should refuse it, since the capability does not list the profile. + """ + if value.signing_profile == WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH and not self.web_bot_auth: + logger.warning( + "[adcp.reporting] skipping a web-bot-auth subscription: production " + "signing is not configured with web_bot_auth=True" + ) + return False + return True + async def list_active( self, *, account_id: str, notification_type: str ) -> tuple[ReportingNotificationSubscription, ...]: values = await self.resolver.list_active( account_id=account_id, notification_type=notification_type ) - return tuple(self._check(v) for v in values) + return tuple(v for v in map(self._check, values) if self._signable(v)) async def get_active( self, *, account_id: str, subscriber_id: str, notification_type: str @@ -119,7 +151,10 @@ async def get_active( value = await self.resolver.get_active( account_id=account_id, subscriber_id=subscriber_id, notification_type=notification_type ) - return self._check(value) if value is not None else None + if value is None: + return None + checked = self._check(value) + return checked if self._signable(checked) else None def production_notification_workers( @@ -144,7 +179,7 @@ def production_notification_workers( ): raise ReportingNotificationError("notification_chain_unready") ReportingProductionSigning.__post_init__(signing) - signed_subscriptions = _SignedSubscriptions(subscriptions) + signed_subscriptions = _SignedSubscriptions(subscriptions, web_bot_auth=signing.web_bot_auth) # The store builds its own exact queue classes; a subclass is refused there. outboxes = store.production_outboxes(projection) return tuple( @@ -219,7 +254,7 @@ def check_workers(support: ReportingProductionSupport) -> None: ReportingProductionSigning.__post_init__(signing) subscriptions = workers[0].subscriptions assert isinstance(subscriptions, _SignedSubscriptions) - if not all( + if subscriptions.web_bot_auth is not signing.web_bot_auth or not all( callable(getattr(subscriptions.resolver, method, None)) for method in ("list_active", "get_active") ): diff --git a/src/adcp/webhook_auth.py b/src/adcp/webhook_auth.py index 3e4e443199..022c9b0f89 100644 --- a/src/adcp/webhook_auth.py +++ b/src/adcp/webhook_auth.py @@ -1,7 +1,7 @@ """Auth-mode strategies for :class:`adcp.webhook_sender.WebhookSender`. -Each strategy owns one authentication mode (RFC 9421 JWK signing, -Standard-Webhooks HMAC, AdCP-legacy HMAC, bearer token) and produces +Each strategy owns one authentication mode (RFC 9421 JWK signing, Web Bot +Auth, Standard-Webhooks HMAC, AdCP-legacy HMAC, bearer token) and produces the auth-bearing headers for one outgoing webhook delivery. The sender calls :meth:`WebhookAuthStrategy.build_auth_headers` per request and merges the returned headers into the wire request. @@ -14,7 +14,7 @@ collapses that into one method call. Strategies are constructed by the public :class:`WebhookSender` -classmethods (:meth:`from_jwk`, :meth:`from_pem`, +classmethods (:meth:`from_jwk`, :meth:`from_pem`, :meth:`from_wba`, :meth:`from_bearer_token`, :meth:`from_adcp_legacy_hmac`, :meth:`from_standard_webhooks_secret`). They are not part of the public API directly — call the matching ``WebhookSender.from_*``. @@ -30,6 +30,8 @@ from dataclasses import dataclass, field from typing import Protocol +from cryptography.hazmat.primitives.asymmetric import ed25519 + from adcp.signing.crypto import PrivateKey from adcp.signing.standard_webhooks import ( HEADER_ID as _SW_HEADER_ID, @@ -50,6 +52,11 @@ # nor can Content-Type (changing it would break body framing on the # receiver). _BASE_RESERVED: frozenset[str] = frozenset({"content-length", "content-type", "host"}) +# Both RFC 9421 profiles reserve the same fields, so a sender holding both +# refuses the same extra_headers whichever profile a delivery uses. +_RFC9421_RESERVED: frozenset[str] = frozenset( + {"signature", "signature-input", "content-digest", "signature-agent"} +) class WebhookAuthStrategy(Protocol): @@ -94,7 +101,42 @@ def build_auth_headers(self, *, method: str, url: str, body: bytes) -> dict[str, return signed.as_dict() def reserved_headers(self) -> frozenset[str]: - return _BASE_RESERVED | {"signature", "signature-input", "content-digest"} + return _BASE_RESERVED | _RFC9421_RESERVED + + +@dataclass(frozen=True) +class WbaSignerStrategy: + """Web Bot Auth webhook-signing profile (``web-bot-auth``, AdCP 3.3). + + A webhook under this profile is an ordinary Web Bot Auth request from the + seller's origin: :func:`adcp.signing.wba.sign_wba_request` covers + ``@method``, ``@target-uri``, ``@authority``, ``content-type``, + ``content-digest`` and ``signature-agent``, with a fresh 64-byte nonce and + ``created`` on every call. Receivers record nonces, so every attempt, + retries included, must be signed afresh — which the sender does by + re-entering :meth:`build_auth_headers`. + """ + + private_key: ed25519.Ed25519PrivateKey = field(repr=False) + agent_origin: str + #: RFC 7638 thumbprint of the public key, the ``keyid`` every signature names. + key_id: str + alg: str = "ed25519" + + def build_auth_headers(self, *, method: str, url: str, body: bytes) -> dict[str, str]: + from adcp.signing.wba import sign_wba_request + + return sign_wba_request( + method=method, + url=url, + headers={"Content-Type": "application/json"}, + body=body, + private_key=self.private_key, + agent_origin=self.agent_origin, + ) + + def reserved_headers(self) -> frozenset[str]: + return _BASE_RESERVED | _RFC9421_RESERVED @dataclass(frozen=True) @@ -218,6 +260,7 @@ def merge_extra_headers( "BearerTokenStrategy", "JwkSignerStrategy", "StandardWebhooksHmacStrategy", + "WbaSignerStrategy", "WebhookAuthStrategy", "merge_extra_headers", ] diff --git a/src/adcp/webhook_sender.py b/src/adcp/webhook_sender.py index 1cbf3e8865..cdb1fa7e48 100644 --- a/src/adcp/webhook_sender.py +++ b/src/adcp/webhook_sender.py @@ -28,13 +28,14 @@ from __future__ import annotations +import copy import json import warnings from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass, field, replace from datetime import datetime, timezone from pathlib import Path -from typing import Any, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable import httpx from cryptography.hazmat.primitives.asymmetric import ec, ed25519 @@ -70,6 +71,7 @@ BearerTokenStrategy, JwkSignerStrategy, StandardWebhooksHmacStrategy, + WbaSignerStrategy, WebhookAuthStrategy, merge_extra_headers, ) @@ -79,6 +81,9 @@ apply_hooks, ) +if TYPE_CHECKING: + from adcp.signing.wba import DirectoryResponse + # WholesaleFeedEvent is a union, so validation goes through an adapter. @@ -147,6 +152,39 @@ def _validate_hooks(hooks: tuple[TransportHook, ...], allow_private_destinations validate(allow_private_destinations=allow_private_destinations) +def _profile_strategies_for(auth: WebhookAuthStrategy) -> dict[str, WebhookAuthStrategy]: + """The webhook-signing profiles ``auth`` produces, keyed by profile name. + + Bearer and HMAC strategies produce none: they serve only registrations + carrying legacy ``authentication``. + """ + if isinstance(auth, JwkSignerStrategy): + return {WEBHOOK_SIGNING_PROFILE_ADCP: auth} + if isinstance(auth, WbaSignerStrategy): + return {WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH: auth} + return {} + + +def _wba_strategy(private_key: Any, agent_origin: str) -> WbaSignerStrategy: + """Validate a Web Bot Auth key and origin up front, at construction.""" + from adcp.signing.wba import _https_origin, public_directory_key + + if not isinstance(private_key, ed25519.Ed25519PrivateKey): + raise ValueError( + f"Web Bot Auth signs with Ed25519 keys only (got {type(private_key).__name__}); " + "an ES256 key stays on adcp/webhook-signing/v1 via WebhookSender.from_jwk" + ) + if not isinstance(agent_origin, str): + raise ValueError("agent_origin must be the seller's https origin") + return WbaSignerStrategy( + private_key=private_key, + # sign_wba_request sends the canonical form; store it so the + # directory helper publishes for the same origin the signatures name. + agent_origin=_https_origin(agent_origin), + key_id=public_directory_key(private_key)["kid"], + ) + + def _entity_type_for_wholesale_notification(notification_type: str) -> str: if notification_type.startswith("product."): return "product" @@ -217,6 +255,10 @@ class WebhookDeliveryResult: response_body: bytes sent_body: bytes = b"" sent_extra_headers: Mapping[str, str] = field(default_factory=dict) + #: The registration's required signing profile the delivery was signed + #: under, or ``None`` when it imposed none (the sender's default mode). + #: :meth:`WebhookSender.resend` signs the replay under the same profile. + signing_profile: str | None = None @property def ok(self) -> bool: @@ -248,21 +290,35 @@ class PreparedWebhook: class WebhookSigningProfileUnavailableError(ValueError): """The sender cannot sign under the profile a webhook registration requires. - A ``reporting_webhook`` / ``artifact_webhook`` registration without - ``authentication`` (AdCP 3.3) must be signed under its ``signing_profile`` - (``adcp/webhook-signing/v1`` when omitted). Raised before any network I/O - when the sender would instead send a bearer token or legacy HMAC the buyer - never registered, or when it cannot produce the selected profile (this SDK - does not yet sign webhooks under ``web-bot-auth``). Configure an RFC 9421 - sender (:meth:`WebhookSender.from_jwk` / :meth:`WebhookSender.from_pem`), - or reject registrations naming a profile the seller does not list in + A ``reporting_webhook`` / ``artifact_webhook`` / ``push_notification_config`` + registration without ``authentication`` (AdCP 3.3) must be signed under its + ``signing_profile`` (``adcp/webhook-signing/v1`` when omitted). Raised + before any network I/O when the sender would instead send a bearer token or + legacy HMAC the buyer never registered, or when it holds no key for the + selected profile. Configure a sender for that profile + (:meth:`WebhookSender.from_jwk` / :meth:`WebhookSender.from_pem` for + ``adcp/webhook-signing/v1``, :meth:`WebhookSender.from_wba` for + ``web-bot-auth``, :meth:`WebhookSender.with_web_bot_auth` for both), or + reject registrations naming a profile the seller does not list in ``webhook_signing.profiles`` with ``INVALID_REQUEST``. """ - def __init__(self, *, required_profile: str, sender_profile: str | None) -> None: + def __init__( + self, + *, + required_profile: str, + sender_profile: str | None, + sender_profiles: frozenset[str] = frozenset(), + ) -> None: self.required_profile = required_profile self.sender_profile = sender_profile - mode = sender_profile if sender_profile is not None else "legacy bearer/HMAC auth" + self.sender_profiles = sender_profiles + if sender_profiles: + mode = " and ".join(sorted(sender_profiles)) + elif sender_profile is not None: + mode = sender_profile + else: + mode = "legacy bearer/HMAC auth" super().__init__( f"webhook registration has no authentication and requires signing under " f"{required_profile!r}, but this WebhookSender signs with {mode}; " @@ -366,6 +422,7 @@ def __init__( self._auth: WebhookAuthStrategy = JwkSignerStrategy( private_key=private_key, key_id=key_id, alg=alg ) + self._profile_strategies = _profile_strategies_for(self._auth) self._key_id = key_id self._timeout = timeout_seconds self._client = client @@ -395,6 +452,7 @@ def _from_strategy( """ sender = cls.__new__(cls) sender._auth = auth + sender._profile_strategies = _profile_strategies_for(auth) sender._key_id = key_id sender._timeout = timeout_seconds sender._client = client @@ -543,6 +601,135 @@ def from_pem( transport_hooks=transport_hooks, ) + @classmethod + def from_wba( + cls, + private_key: ed25519.Ed25519PrivateKey, + *, + agent_origin: str, + client: httpx.AsyncClient | None = None, + timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS, + allow_private_destinations: bool = False, + allowed_destination_ports: frozenset[int] | None = None, + transport_hooks: tuple[TransportHook, ...] = (), + ) -> WebhookSender: + """Build a sender that signs under the ``web-bot-auth`` profile (AdCP 3.3). + + Every attempt, retries and outbox redeliveries included, is signed + afresh by :func:`adcp.signing.wba.sign_wba_request`: a new 64-byte + nonce and ``created``, ``Signature-Agent`` naming ``agent_origin``, + the key's RFC 7638 thumbprint as ``keyid``, and ``@method``, + ``@target-uri``, ``@authority``, ``content-type``, ``content-digest`` + and ``signature-agent`` covered. Receivers record nonces, including on + an attempt they answered 503, so a stored signature is never resent. + + Receivers verify against the key directory at + ``{agent_origin}/.well-known/http-message-signatures-directory``; + serve :meth:`sign_web_bot_auth_directory` there before sending. + + A seller with buyers on both profiles during 3.3 chains + :meth:`with_web_bot_auth` onto its ``from_jwk`` / ``from_pem`` sender + instead. + + :param private_key: The origin's Ed25519 key (the profile's one + algorithm). Load a PEM with + :func:`adcp.signing.load_private_key_pem`. + :param agent_origin: The seller's https origin, e.g. + ``https://seller.example``; sent in canonical form. + :raises ValueError: the key is not Ed25519, or ``agent_origin`` is not + an https origin without userinfo. + """ + strategy = _wba_strategy(private_key, agent_origin) + return cls._from_strategy( + strategy, + key_id=strategy.key_id, + client=client, + timeout_seconds=timeout_seconds, + allow_private_destinations=allow_private_destinations, + allowed_destination_ports=allowed_destination_ports, + transport_hooks=transport_hooks, + ) + + def with_web_bot_auth( + self, + private_key: ed25519.Ed25519PrivateKey, + *, + agent_origin: str, + ) -> WebhookSender: + """Return a sender that also signs under ``web-bot-auth``. + + For AdCP 3.3, when a seller has buyers registered on both profiles. + The returned sender signs each delivery under the profile its + registration selected (``required_signing_profile``); a delivery that + selects none (a registration carrying legacy ``authentication`` is + handled by a separate sender, so this means a direct ``send_*`` call) + keeps this sender's ``adcp/webhook-signing/v1`` key. Transport + settings are shared; this sender is left unchanged. + + :: + + sender = WebhookSender.from_jwk(webhook_jwk).with_web_bot_auth( + wba_private_key, agent_origin="https://seller.example" + ) + + :raises ValueError: this sender does not sign under + ``adcp/webhook-signing/v1``, or already signs under + ``web-bot-auth``, or the key or origin is invalid (see + :meth:`from_wba`). + """ + if WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH in self._profile_strategies: + raise ValueError("this WebhookSender already signs under web-bot-auth") + if self.signing_profile != WEBHOOK_SIGNING_PROFILE_ADCP: + raise ValueError( + "with_web_bot_auth extends an adcp/webhook-signing/v1 sender " + "(from_jwk / from_pem); use WebhookSender.from_wba for a " + "web-bot-auth-only sender" + ) + strategy = _wba_strategy(private_key, agent_origin) + combined = copy.copy(self) + combined._profile_strategies = { + **self._profile_strategies, + WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH: strategy, + } + return combined + + def sign_web_bot_auth_directory( + self, + *, + other_keys: Sequence[ed25519.Ed25519PrivateKey] = (), + expires_in_seconds: int = 86400, + ) -> DirectoryResponse: + """Sign the key directory that makes this sender's signatures verifiable. + + Serve the returned body and headers, unchanged, at + ``{agent_origin}/.well-known/http-message-signatures-directory`` + (:data:`adcp.signing.wba.DIRECTORY_PATH`) with status 200, and + republish before ``expires_in_seconds`` elapses. Receivers resolve the + ``web-bot-auth`` key there, so publish the directory before the first + delivery, and keep the outgoing key in it while redeliveries may still + carry it. + + The directory is the origin's, not the sender's: list every Ed25519 + key the origin signs with under Web Bot Auth, including request + signing keys and keys mid-rotation, in ``other_keys``. A key left out + stops verifying once receivers refresh their cached copy. + + :raises ValueError: this sender holds no ``web-bot-auth`` key. + """ + from adcp.signing.wba import sign_key_directory + + strategy = self._profile_strategies.get(WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH) + if not isinstance(strategy, WbaSignerStrategy): + raise ValueError( + "this WebhookSender holds no web-bot-auth key; build it with " + "from_wba or with_web_bot_auth" + ) + return sign_key_directory( + agent_origin=strategy.agent_origin, + private_keys=[strategy.private_key, *other_keys], + expires_in_seconds=expires_in_seconds, + ) + @classmethod def from_bearer_token( cls, @@ -672,17 +859,50 @@ def __repr__(self) -> str: # Explicit repr so no future debug helper or error traceback auto- # renders self.__dict__ and pulls the private key (or HMAC secret / # bearer token) into logs. - return f"WebhookSender(auth={type(self._auth).__name__}, key_id={self._key_id!r})" + profiles = sorted(self._profile_strategies) + return ( + f"WebhookSender(auth={type(self._auth).__name__}, key_id={self._key_id!r}, " + f"profiles={profiles!r})" + ) @property def signing_profile(self) -> str | None: - """The webhook-signing profile this sender produces, or ``None``. + """The webhook-signing profile this sender signs under by default, or ``None``. + + ``adcp/webhook-signing/v1`` for :meth:`from_jwk` / :meth:`from_pem` + (with or without :meth:`with_web_bot_auth`); ``web-bot-auth`` for + :meth:`from_wba`; ``None`` for the bearer and HMAC senders, which + serve only registrations carrying legacy ``authentication``. A + delivery whose registration selects a profile is signed under that + one instead; see :attr:`signing_profiles`. + """ + for profile, strategy in self._profile_strategies.items(): + if strategy is self._auth: + return profile + return None + + @property + def signing_profiles(self) -> frozenset[str]: + """Every webhook-signing profile this sender can sign a delivery under. - ``adcp/webhook-signing/v1`` for an RFC 9421 sender (:meth:`from_jwk`, - :meth:`from_pem`); ``None`` for the bearer and HMAC senders, which - serve only registrations carrying legacy ``authentication``. + Compare with ``webhook_signing.profiles`` in the seller's + capabilities: a registration naming a profile outside this set is + refused with :class:`WebhookSigningProfileUnavailableError`. """ - return WEBHOOK_SIGNING_PROFILE_ADCP if self.signs_with_rfc9421 else None + return frozenset(self._profile_strategies) + + @property + def signing_algorithms(self) -> frozenset[str]: + """The RFC 9421 ``alg`` values this sender's signatures may carry. + + Empty for the bearer and HMAC senders. Compare with + ``webhook_signing.algorithms`` in the seller's capabilities. + """ + return frozenset( + alg + for strategy in self._profile_strategies.values() + if isinstance(alg := getattr(strategy, "alg", None), str) + ) def check_registration(self, registration: BaseModel | Mapping[str, Any]) -> str | None: """Check this sender can deliver to a webhook registration; no network I/O. @@ -703,23 +923,32 @@ def check_registration(self, registration: BaseModel | Mapping[str, Any]) -> str return required def _require_profile(self, required: str | None) -> None: - if required is not None and required != self.signing_profile: + if required is not None and required not in self._profile_strategies: raise WebhookSigningProfileUnavailableError( - required_profile=required, sender_profile=self.signing_profile + required_profile=required, + sender_profile=self.signing_profile, + sender_profiles=self.signing_profiles, ) + def _strategy_for(self, required: str | None) -> WebhookAuthStrategy: + """The strategy for a delivery's required profile; the default when none.""" + self._require_profile(required) + return self._auth if required is None else self._profile_strategies[required] + @property def signs_with_rfc9421(self) -> bool: - """``True`` iff this sender uses the RFC 9421 webhook-signing profile. + """``True`` iff this sender signs with an RFC 9421 webhook-signing profile. Boot-time validators read this to enforce the ``webhook_signing.supported=true`` capability invariant: capabilities advertise RFC 9421 → wired sender must produce - ``Signature`` / ``Signature-Input`` headers. ``from_bearer_token``, - ``from_adcp_legacy_hmac``, and ``from_standard_webhooks_secret`` - senders return ``False``. + ``Signature`` / ``Signature-Input`` headers. ``from_jwk``, + ``from_pem`` and ``from_wba`` senders (``adcp/webhook-signing/v1`` + and ``web-bot-auth`` are both RFC 9421 profiles) return ``True``; + ``from_bearer_token``, ``from_adcp_legacy_hmac``, and + ``from_standard_webhooks_secret`` senders return ``False``. """ - return isinstance(self._auth, JwkSignerStrategy) + return isinstance(self._auth, (JwkSignerStrategy, WbaSignerStrategy)) async def aclose(self) -> None: """Close the internal httpx client if we own it.""" @@ -983,6 +1212,7 @@ async def send_prepared( idempotency_key=prepared.idempotency_key, extra_headers=prepared.extra_headers or None, before_attempt=before_attempt, + signing_profile=prepared.required_signing_profile, ) @staticmethod @@ -1962,9 +2192,13 @@ async def resend( The bytes are identical (same ``idempotency_key``, same payload fields, same serialization) — only the Signature / Signature-Input / - Content-Digest headers are regenerated. The receiver dedupes via - ``idempotency_key``, so the replayed event is a spec-correct retry - that won't cause double-processing. + Content-Digest headers are regenerated, under the same signing + profile as the original (``result.signing_profile``). The receiver + dedupes via ``idempotency_key``, so the replayed event is a + spec-correct retry that won't cause double-processing. + + :raises WebhookSigningProfileUnavailableError: ``result`` was signed + under a registration's profile this sender does not hold. """ if not result.sent_body: raise ValueError( @@ -1985,6 +2219,7 @@ async def resend( body=result.sent_body, idempotency_key=result.idempotency_key, extra_headers=result.sent_extra_headers or None, + signing_profile=result.signing_profile, ) async def _send_bytes( @@ -1995,9 +2230,13 @@ async def _send_bytes( idempotency_key: str, extra_headers: Mapping[str, str] | None, before_attempt: Callable[[], Awaitable[bool]] | None = None, + signing_profile: str | None = None, ) -> WebhookDeliveryResult: """Sign + POST a pre-serialized body through an SSRF-validated transport. + ``signing_profile`` is the delivery's required profile; ``None`` signs + with the sender's default mode. It is checked before any network I/O. + When the sender owns its httpx client (the default — ``client=None`` was passed to ``__init__``), every delivery builds a per-request :class:`adcp.signing.ip_pinned_transport.AsyncIpPinnedTransport` @@ -2028,6 +2267,7 @@ async def _send_bytes( fail. The hook output is bounded (hostname-only rewrite, scheme and port preserved), so this can't widen the destination space. """ + auth = self._strategy_for(signing_profile) effective_url = apply_hooks(url, self._transport_hooks) # Build the pinned transport up-front for the owned-client path. @@ -2045,12 +2285,17 @@ async def _send_bytes( ) 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}, - extra=extra_headers, - reserved=self._auth.reserved_headers(), - ) + try: + auth_headers = auth.build_auth_headers(method="POST", url=effective_url, body=body) + headers = merge_extra_headers( + base={**base_headers, **auth_headers}, + extra=extra_headers, + reserved=auth.reserved_headers(), + ) + except BaseException: + if transport is not None: + await transport.aclose() + raise if before_attempt is not None: try: @@ -2106,6 +2351,7 @@ async def _send_bytes( response_body=response_body, sent_body=body, sent_extra_headers=dict(extra_headers) if extra_headers else {}, + signing_profile=signing_profile, ) @@ -2114,6 +2360,7 @@ async def _send_bytes( # cycle — see the note next to the removed top-level import. from adcp.webhooks import ( # noqa: E402 WEBHOOK_SIGNING_PROFILE_ADCP, + WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH, create_mcp_webhook_payload, create_media_buy_delivery_webhook_payload, create_webhook_challenge_payload, diff --git a/src/adcp/webhooks.py b/src/adcp/webhooks.py index 86db15bd63..d6d2705d80 100644 --- a/src/adcp/webhooks.py +++ b/src/adcp/webhooks.py @@ -343,7 +343,8 @@ def __init__(self, *, media_buy_ids: tuple[str, ...] = ()) -> None: WEBHOOK_SIGNING_PROFILE_ADCP = "adcp/webhook-signing/v1" #: The Web Bot Auth webhook-signing profile (experimental, AdCP 3.3). WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH = "web-bot-auth" -_WEBHOOK_SIGNING_PROFILES = frozenset( +#: Every webhook-signing profile a registration may select. +WEBHOOK_SIGNING_PROFILES: frozenset[str] = frozenset( {WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH} ) @@ -432,10 +433,10 @@ def webhook_signing_profile_for( ) if profile is None: return WEBHOOK_SIGNING_PROFILE_ADCP - if profile not in _WEBHOOK_SIGNING_PROFILES: + if profile not in WEBHOOK_SIGNING_PROFILES: raise WebhookRegistrationError( f"unknown signing_profile {profile!r}; expected one of " - f"{sorted(_WEBHOOK_SIGNING_PROFILES)}", + f"{sorted(WEBHOOK_SIGNING_PROFILES)}", field="signing_profile", ) return cast(str, profile) @@ -1547,6 +1548,7 @@ async def _send_sender_webhook_challenge( timeout_seconds: float | None, policy: WebhookDestinationPolicy, extra_headers: Mapping[str, str] | None, + signing_profile: str = WEBHOOK_SIGNING_PROFILE_ADCP, ) -> httpx.Response: body_bytes = json.dumps(payload, separators=(",", ":")).encode("utf-8") if len(body_bytes) > _MAX_BODY_BYTES: @@ -1555,7 +1557,8 @@ async def _send_sender_webhook_challenge( f"over the {_MAX_BODY_BYTES:,}-byte cap." ) - auth = getattr(cast(Any, sender), "_auth") + # Sign under the profile the config selected, never the sender's default. + auth = sender._strategy_for(signing_profile) auth_headers = auth.build_auth_headers(method="POST", url=url, body=body_bytes) from adcp.webhook_auth import merge_extra_headers @@ -1610,6 +1613,7 @@ async def challenge_webhook_destination( policy: WebhookDestinationPolicy | None = None, field: str | None = None, extra_headers: Mapping[str, str] | None = None, + signing_profile: str | None = None, ) -> WebhookChallengeResult: """Validate and prove control of a durable webhook destination. @@ -1621,6 +1625,8 @@ async def challenge_webhook_destination( when present, the challenge is sent with Bearer or HMAC-SHA256. When omitted, pass an RFC 9421 :class:`WebhookSender`; the helper uses that sender's webhook-signing key and the SDK-managed pinned transport. + ``signing_profile`` is the config's ``signing_profile`` (AdCP 3.3); the + challenge is signed under it, ``adcp/webhook-signing/v1`` when omitted. """ error_url = str(url) if isinstance(url, (str, AnyUrl)) else None @@ -1658,6 +1664,25 @@ async def challenge_webhook_destination( "or pass config.authentication for legacy Bearer/HMAC configs." ), ) + required_profile = signing_profile or WEBHOOK_SIGNING_PROFILE_ADCP + if sender is not None and required_profile not in sender.signing_profiles: + raise WebhookChallengeError( + f"this WebhookSender cannot sign under the config's {required_profile!r} profile", + reason="signing_profile_unavailable", + field=field, + url=error_url, + suggestion=( + "Use WebhookSender.from_wba(...) or with_web_bot_auth(...) for web-bot-auth " + "configs, and from_jwk(...) / from_pem(...) for adcp/webhook-signing/v1." + ), + ) + if authentication is not None and signing_profile is not None: + raise WebhookChallengeError( + "a config with authentication cannot also select a signing_profile", + reason="ambiguous_auth_mode", + field=field, + url=error_url, + ) if sender is None and authentication is None: raise WebhookChallengeError( "webhook challenge requires sender= when authentication is omitted", @@ -1706,6 +1731,7 @@ async def challenge_webhook_destination( timeout_seconds=effective_timeout, policy=destination.policy, extra_headers=extra_headers, + signing_profile=required_profile, ) status_code = response.status_code response_headers = dict(response.headers) @@ -2282,6 +2308,7 @@ def _validate_header_value(name: str, value: Any) -> None: "REPORTING_WEBHOOK_OPERATION_ID_MISSING", "ReportingWebhookOperationIdMissingError", "WEBHOOK_SIGNING_PROFILE_ADCP", + "WEBHOOK_SIGNING_PROFILES", "WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH", "WebhookRegistrationError", "WebhookSigningProfileUnavailableError", diff --git a/tests/test_decisioning_webhook_emit.py b/tests/test_decisioning_webhook_emit.py index bfbaa1ba38..69ca2cd1a3 100644 --- a/tests/test_decisioning_webhook_emit.py +++ b/tests/test_decisioning_webhook_emit.py @@ -1043,3 +1043,97 @@ class _Params: result={"media_buy_id": "mb_1"}, ) assert caplog.records == [] + + +def _signed_request(profile: str | None, *, idem_suffix: str) -> CreateMediaBuyRequest: + request = _make_request(with_url=True, idem_suffix=idem_suffix) + if profile is None: + return request + payload = request.model_dump(mode="json", exclude_none=True) + payload["push_notification_config"]["signing_profile"] = profile + return CreateMediaBuyRequest(**payload) + + +class _TenantProfilesPlatform(_HandoffPlatform): + """Static capabilities sign adcp/webhook-signing/v1 only; one tenant adds web-bot-auth.""" + + capabilities = DecisioningCapabilities( + specialisms=["sales-non-guaranteed"], + webhook_signing=WebhookSigning(supported=True, delivery_retry_horizon_seconds=86400), + webhook_signing_managed_externally=True, + ) + + def get_adcp_capabilities_for_request(self, params=None, context=None): + if context is None or context.tenant_id != "wba-tenant": + return None + return DecisioningCapabilities( + specialisms=["sales-non-guaranteed"], + webhook_signing=WebhookSigning( + supported=True, + delivery_retry_horizon_seconds=86400, + profiles=["adcp/webhook-signing/v1", "web-bot-auth"], + ), + webhook_signing_managed_externally=True, + ) + + +@pytest.mark.asyncio +async def test_scoped_capabilities_admit_a_push_profile_the_static_set_lacks(executor) -> None: + """The tenant's advertised profiles decide, not the static capabilities.""" + registry = InMemoryTaskRegistry() + handler = PlatformHandler( + _TenantProfilesPlatform(), + executor=executor, + registry=registry, + auto_emit_task_webhooks=False, + ) + + result = await handler.create_media_buy( + _signed_request("web-bot-auth", idem_suffix="wba"), + ToolContext(tenant_id="wba-tenant"), + ) + + assert result["status"] == "submitted" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tenant", [None, "other-tenant"]) +async def test_unadvertised_push_profile_is_refused_before_the_task_is_issued( + executor, tenant: str | None +) -> None: + registry = InMemoryTaskRegistry() + handler = PlatformHandler( + _TenantProfilesPlatform(), + executor=executor, + registry=registry, + auto_emit_task_webhooks=False, + ) + + with pytest.raises(AdcpError) as exc_info: + await handler.create_media_buy( + _signed_request("web-bot-auth", idem_suffix="no_wba"), + ToolContext(tenant_id=tenant), + ) + + assert exc_info.value.code == "INVALID_REQUEST" + assert exc_info.value.field == "push_notification_config.signing_profile" + assert registry._records == {} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("profile", [None, "adcp/webhook-signing/v1"]) +async def test_default_push_profile_stays_admitted(executor, profile: str | None) -> None: + """Omitted webhook_signing.profiles advertises adcp/webhook-signing/v1.""" + handler = PlatformHandler( + _TenantProfilesPlatform(), + executor=executor, + registry=InMemoryTaskRegistry(), + auto_emit_task_webhooks=False, + ) + + result = await handler.create_media_buy( + _signed_request(profile, idem_suffix="adcp"), + ToolContext(tenant_id="other-tenant"), + ) + + assert result["status"] == "submitted" diff --git a/tests/test_reporting_webhook_wba.py b/tests/test_reporting_webhook_wba.py new file mode 100644 index 0000000000..e5065652b6 --- /dev/null +++ b/tests/test_reporting_webhook_wba.py @@ -0,0 +1,251 @@ +"""Reporting notifications signed under ``web-bot-auth`` when a subscription asks (#1859). + +A reporting subscription without legacy ``authentication`` records the +registration's ``signing_profile``. The worker signs each attempt with a fresh +``from_wba`` sender for ``web-bot-auth`` subscriptions, using the Web Bot Auth +key the trusted signing resolver returns, and leaves every subscription and +delivery stored before the field byte-identical. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict +from types import SimpleNamespace +from typing import Any + +import pytest +from cryptography.hazmat.primitives.asymmetric import ec, ed25519 + +from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.ledger.models import ledger_asdict +from adcp.reporting.ledger.notification_models import ReportingNotificationError +from adcp.reporting.outbox import ( + ReportingLegacyAuthentication, + ReportingNotificationSubscription, + ReportingNotificationWorker, + ReportingSigningMaterial, + ReportingWebBotAuthKey, +) +from adcp.reporting.outbox.routing import _decode_subscription +from adcp.reporting.production.notifications import ( + ReportingProductionSigning, + _SignedSubscriptions, +) +from adcp.webhook_sender import ScopePermanentlyUnknown +from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH + +ORIGIN = "https://seller.example" +WBA = WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH +ADCP = WEBHOOK_SIGNING_PROFILE_ADCP + + +def _subscription(**changes: Any) -> ReportingNotificationSubscription: + values: dict[str, Any] = dict( + account_id="acct_a", + subscriber_id="buyer", + principal_id="buyer", + url="https://receiver.example.test/reporting", + event_types=("reporting.ledger_changed",), + configuration_revision="registration-1", + authorization_ref="authorized-principal-1", + proof_of_control_ref="account-challenge-1", + signing_scope_id="seller-signing-scope-1", + active=True, + authorized=True, + proof_valid=True, + ) + values.update(changes) + return ReportingNotificationSubscription(**values) + + +def _material(*, web_bot_auth: bool = True, algorithms: frozenset[str] = frozenset({"ed25519"})): + return ReportingSigningMaterial( + ed25519.Ed25519PrivateKey.generate(), + "https://seller.example/keys#webhook-1", + "ed25519", + algorithms, + web_bot_auth=( + ReportingWebBotAuthKey(ed25519.Ed25519PrivateKey.generate(), ORIGIN) + if web_bot_auth + else None + ), + ) + + +# -- Subscriptions ------------------------------------------------------------- + + +def test_a_subscription_without_a_profile_keeps_its_stored_fingerprint() -> None: + subscription = _subscription() + pre_field = {k: v for k, v in asdict(subscription).items() if k != "signing_profile"} + + assert "signing_profile" not in ledger_asdict(subscription) + assert subscription.fingerprint == hashlib.sha256(canonical_json_utf8_v1(pre_field)).hexdigest() + # Envelopes stored before the field decode exactly as before. + assert _decode_subscription(json.loads(canonical_json_utf8_v1(pre_field))) == subscription + + +def test_a_web_bot_auth_subscription_binds_its_profile() -> None: + subscription = _subscription(signing_profile=WBA) + stored = json.loads(canonical_json_utf8_v1(ledger_asdict(subscription))) + + assert stored["signing_profile"] == WBA + assert _decode_subscription(stored) == subscription + assert subscription.fingerprint != _subscription().fingerprint + assert subscription.auth_mode == "rfc9421" + + +@pytest.mark.parametrize( + "changes", + [ + {"signing_profile": "custom"}, + { + "signing_profile": WBA, + "signing_scope_id": None, + "authentication": ReportingLegacyAuthentication("Bearer", "x" * 32), + }, + ], +) +def test_an_invalid_subscription_profile_is_refused(changes: dict[str, Any]) -> None: + with pytest.raises(ReportingNotificationError, match="invalid_configuration"): + _subscription(**changes) + + +# -- Signing material ---------------------------------------------------------- + + +@pytest.mark.parametrize( + "key,origin", + [ + (ec.generate_private_key(ec.SECP256R1()), ORIGIN), + (ed25519.Ed25519PrivateKey.generate(), "http://seller.example"), + ], +) +def test_web_bot_auth_key_must_be_ed25519_on_an_https_origin(key: Any, origin: str) -> None: + with pytest.raises(ReportingNotificationError, match="permanent_scope"): + ReportingWebBotAuthKey(key, origin) + + +def test_web_bot_auth_material_requires_ed25519_to_be_advertised() -> None: + p256 = ec.generate_private_key(ec.SECP256R1()) + with pytest.raises(ReportingNotificationError, match="permanent_scope"): + ReportingSigningMaterial( + p256, + "kid", + "ecdsa-p256-sha256", + frozenset({"ecdsa-p256-sha256"}), + web_bot_auth=ReportingWebBotAuthKey(ed25519.Ed25519PrivateKey.generate(), ORIGIN), + ) + + +# -- The worker's sender ------------------------------------------------------- + + +async def _sender(subscription: ReportingNotificationSubscription, material: Any) -> Any: + class Signing: + async def resolve(self, **_: Any) -> Any: + return material + + worker = SimpleNamespace(signing=Signing(), lease_seconds=30) + return await ReportingNotificationWorker._sender(worker, subscription) # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_worker_signs_a_web_bot_auth_subscription_with_the_origin_key() -> None: + material = _material() + sender = await _sender(_subscription(signing_profile=WBA), material) + + assert sender.signing_profile == WBA + assert sender.signing_profiles == frozenset({WBA}) + assert sender._allowed_destination_ports == frozenset({443}) + directory = sender.sign_web_bot_auth_directory() + assert directory.headers["Content-Type"] == "application/http-message-signatures-directory+json" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("profile", [None, ADCP]) +async def test_worker_keeps_adcp_subscriptions_on_the_adcp_key(profile: str | None) -> None: + sender = await _sender(_subscription(signing_profile=profile), _material()) + assert sender.signing_profiles == frozenset({ADCP}) + + +@pytest.mark.asyncio +async def test_worker_quarantines_a_web_bot_auth_subscription_without_a_key() -> None: + with pytest.raises(ScopePermanentlyUnknown): + await _sender(_subscription(signing_profile=WBA), _material(web_bot_auth=False)) + + +# -- Production signing contract ----------------------------------------------- + + +class _Resolver: + def __init__(self, material: ReportingSigningMaterial) -> None: + self.material = material + + async def resolve(self, **_: Any) -> ReportingSigningMaterial: + return self.material + + +def _production(material: ReportingSigningMaterial, **kwargs: Any) -> ReportingProductionSigning: + return ReportingProductionSigning( + _Resolver(material), + ("ed25519",), + brand_json_url="https://seller.example/brand.json", + **kwargs, + ) + + +def test_production_signing_advertises_web_bot_auth_only_when_enabled() -> None: + assert "profiles" not in _production(_material()).wire() + assert _production(_material(), web_bot_auth=True).wire()["profiles"] == [ADCP, WBA] + + +def test_production_web_bot_auth_requires_ed25519() -> None: + p256 = ec.generate_private_key(ec.SECP256R1()) + material = ReportingSigningMaterial( + p256, "kid", "ecdsa-p256-sha256", frozenset({"ecdsa-p256-sha256"}) + ) + with pytest.raises(ReportingNotificationError, match="notification_signing_unready"): + ReportingProductionSigning( + _Resolver(material), + ("ecdsa-p256-sha256",), + brand_json_url="https://seller.example/brand.json", + web_bot_auth=True, + ) + + +@pytest.mark.asyncio +async def test_production_web_bot_auth_requires_the_key_in_every_material() -> None: + signing = _production(_material(web_bot_auth=False), web_bot_auth=True) + with pytest.raises(ReportingNotificationError, match="notification_signing_unready"): + await signing.resolve(account_id="acct_a", principal_id="buyer", signing_scope_id="s") + assert ( + await _production(_material(), web_bot_auth=True).resolve( + account_id="acct_a", principal_id="buyer", signing_scope_id="s" + ) + ).web_bot_auth is not None + + +@pytest.mark.asyncio +async def test_production_skips_only_the_web_bot_auth_subscriptions_it_cannot_sign() -> None: + wba = _subscription(subscriber_id="wba_buyer", signing_profile=WBA) + adcp = _subscription(subscriber_id="adcp_buyer") + + class Subscriptions: + async def list_active(self, **_: Any) -> tuple[ReportingNotificationSubscription, ...]: + return (wba, adcp) + + async def get_active(self, **_: Any) -> ReportingNotificationSubscription: + return wba + + query = {"account_id": "acct_a", "notification_type": "reporting.ledger_changed"} + without = _SignedSubscriptions(Subscriptions()) + # One unsignable subscriber does not block the account's other subscribers. + assert await without.list_active(**query) == (adcp,) + assert await without.get_active(subscriber_id="wba_buyer", **query) is None + + enabled = _SignedSubscriptions(Subscriptions(), web_bot_auth=True) + assert await enabled.list_active(**query) == (wba, adcp) + assert await enabled.get_active(subscriber_id="wba_buyer", **query) == wba diff --git a/tests/test_task_webhook_outbox_pg.py b/tests/test_task_webhook_outbox_pg.py index 2600f7a034..04e9852a39 100644 --- a/tests/test_task_webhook_outbox_pg.py +++ b/tests/test_task_webhook_outbox_pg.py @@ -58,6 +58,8 @@ def _sender() -> MagicMock: sender._transport_hooks = () sender._timeout = 10.0 sender._auth.alg = "ed25519" + sender.signing_algorithms = frozenset({"ed25519"}) + sender.signing_profiles = frozenset({"adcp/webhook-signing/v1"}) body = json.dumps( { "idempotency_key": "whk_1234567890123456", @@ -277,7 +279,7 @@ def test_scoped_registration_is_encrypted_and_mode_bound() -> None: task_type="create_media_buy", encrypted_registration=encrypted, nonce=nonce, - ) == ("https://buyer.example/webhook", "op_1", None, None, "tenant-key-scope-a") + ) == ("https://buyer.example/webhook", "op_1", None, None, "tenant-key-scope-a", None) assert outbox.open_registration( account_id="acct_1", task_id="task_1", @@ -394,6 +396,7 @@ def test_registration_round_trips_every_authentication_mode(authentication) -> N "buyer-token", authentication, None, + None, ) @@ -437,6 +440,7 @@ def test_pre_authentication_registration_remains_readable() -> None: "buyer-token", None, None, + None, ) @@ -567,8 +571,9 @@ async def test_enqueue_encrypts_legacy_credentials_for_success_and_failure( idempotency_key="whk_1234567890123456", ), ) - body, opened_authentication = outbox._open_delivery_body(protected) + body, opened_authentication, opened_profile = outbox._open_delivery_body(protected) assert opened_authentication == authentication + assert opened_profile is None assert body == outbox._sender.prepare_mcp.return_value.body @@ -1093,6 +1098,7 @@ async def test_registry_completion_enqueues_on_same_transaction_connection() -> "buyer-token", None, None, + None, ) ) pool = _pool(conn) @@ -1117,6 +1123,7 @@ async def test_registry_completion_enqueues_on_same_transaction_connection() -> token="buyer-token", authentication=None, signing_scope_id=None, + signing_profile=None, ) outbox._open_registration_with_scope.assert_called_once_with( account_id="acct_1", @@ -1152,6 +1159,7 @@ async def test_registry_failure_enqueues_on_same_explicit_transaction() -> None: None, None, None, + None, ) ) pool = _pool(conn) @@ -1174,6 +1182,7 @@ async def test_registry_failure_enqueues_on_same_explicit_transaction() -> None: token=None, authentication=None, signing_scope_id=None, + signing_profile=None, ) conn.transaction.assert_called_once_with() assert "webhook_registration = NULL" in conn.execute.await_args_list[-1].args[0] diff --git a/tests/test_webhook_challenge.py b/tests/test_webhook_challenge.py index 92e6c7c394..d74ff78f04 100644 --- a/tests/test_webhook_challenge.py +++ b/tests/test_webhook_challenge.py @@ -454,3 +454,58 @@ async def test_challenge_webhook_destination_sender_uses_sender_timeout( ) assert async_client.call_args.kwargs["timeout"] == 42.0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "profile,tag", + [ + (None, "adcp/webhook-signing/v1"), + ("adcp/webhook-signing/v1", "adcp/webhook-signing/v1"), + ("web-bot-auth", "web-bot-auth"), + ], +) +async def test_challenge_is_signed_under_the_configs_signing_profile( + monkeypatch: pytest.MonkeyPatch, profile: str | None, tag: str +) -> None: + from cryptography.hazmat.primitives.asymmetric import ed25519 + + _public_dns(monkeypatch) + captured: list[httpx.Request] = [] + sender = _rfc9421_sender().with_web_bot_auth( + ed25519.Ed25519PrivateKey.generate(), agent_origin="https://seller.example" + ) + + with patch( + "adcp.signing.ip_pinned_transport.build_async_ip_pinned_transport", + return_value=_challenge_transport(captured), + ): + result = await challenge_webhook_destination( + url="https://buyer.example/webhooks/catalog", + account_id="acct_1", + subscriber_id="buyer-primary", + sender=sender, + signing_profile=profile, + ) + + assert result.ok + assert f';tag="{tag}"' in captured[0].headers["signature-input"] + assert ("signature-agent" in captured[0].headers) is (tag == "web-bot-auth") + + +@pytest.mark.asyncio +async def test_challenge_refuses_a_profile_the_sender_cannot_sign() -> None: + from cryptography.hazmat.primitives.asymmetric import ed25519 + + sender = WebhookSender.from_wba( + ed25519.Ed25519PrivateKey.generate(), agent_origin="https://seller.example" + ) + # An omitted signing_profile means adcp/webhook-signing/v1. + with pytest.raises(WebhookChallengeError) as exc: + await challenge_webhook_destination( + url="https://buyer.example/webhooks/catalog", + account_id="acct_1", + subscriber_id="buyer-primary", + sender=sender, + ) + assert exc.value.reason == "signing_profile_unavailable" diff --git a/tests/test_webhook_sender_wba.py b/tests/test_webhook_sender_wba.py new file mode 100644 index 0000000000..b7687d4f76 --- /dev/null +++ b/tests/test_webhook_sender_wba.py @@ -0,0 +1,688 @@ +"""WebhookSender signs deliveries under ``web-bot-auth`` when a registration asks (#1859). + +AdCP 3.3 lets a webhook registration omit ``authentication`` and name a +``signing_profile``; the seller signs under it. These tests cover the +``web-bot-auth`` sender (``from_wba``), a sender holding both profiles +(``with_web_bot_auth``), the key directory that makes its signatures +verifiable, and the outboxes carrying each registration's profile to every +freshly signed attempt. +""" + +from __future__ import annotations + +import base64 +import copy +import json +import time +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from cryptography.hazmat.primitives.asymmetric import ec, ed25519 + +from adcp.decisioning.capabilities import WebhookSigning +from adcp.decisioning.task_registry import TaskWebhookAuthentication +from adcp.decisioning.types import AdcpError +from adcp.decisioning.webhook_emit import ( + _extract_push_signing_profile, + validate_webhook_signing_for_capabilities, +) +from adcp.signing import InMemoryReplayStore +from adcp.signing.wba import ( + DIRECTORY_PATH, + DirectoryResponse, + DiscoveredSignerKeys, + public_directory_key, + verify_directory_response, + verify_wba_request, +) +from adcp.webhook_sender import ( + PreparedWebhook, + WebhookDeliveryResult, + WebhookSender, + WebhookSigningProfileUnavailableError, +) +from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH + +ORIGIN = "https://seller.example" +HOOK = "https://buyer.example/webhooks/adcp" +WBA = WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH +ADCP = WEBHOOK_SIGNING_PROFILE_ADCP + +_VECTORS = Path(__file__).parent / "conformance" / "vectors" / "request-signing" +_REQUEST_ED25519 = next( + k + for k in json.loads((_VECTORS / "keys.json").read_text())["keys"] + if k["kid"] == "test-ed25519-2026" +) +_WEBHOOK_JWK = { + **copy.deepcopy(_REQUEST_ED25519), + "kid": "test-webhook-ed25519-2026", + "adcp_use": "webhook-signing", + "d": _REQUEST_ED25519["_private_d_for_test_only"], +} + + +class _Capture(httpx.AsyncBaseTransport): + """Record every request; answer with the queued statuses, then 200.""" + + def __init__(self, *statuses: int) -> None: + self.requests: list[httpx.Request] = [] + self._statuses = list(statuses) + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + await request.aread() + self.requests.append(request) + status = self._statuses.pop(0) if self._statuses else 200 + return httpx.Response(status, content=b"{}", request=request) + + +def _key() -> ed25519.Ed25519PrivateKey: + return ed25519.Ed25519PrivateKey.generate() + + +def _prepared(profile: str | None = None) -> PreparedWebhook: + prepared = WebhookSender.prepare_mcp( + url=HOOK, + task_id="task_1", + task_type="create_media_buy", + status="completed", + result={"media_buy_id": "mb_1"}, + operation_id="op_1", + ) + return replace(prepared, required_signing_profile=profile) + + +def _tag(request: httpx.Request) -> str: + return request.headers["signature-input"].split(';tag="', 1)[1].split('"', 1)[0] + + +async def _verify(request: httpx.Request, directory: DirectoryResponse, replay: Any) -> Any: + async def fetch(url: str, *, allow_private: bool = False) -> DirectoryResponse: + assert url == ORIGIN + DIRECTORY_PATH + return directory + + return await verify_wba_request( + method=request.method, + url=str(request.url), + headers=dict(request.headers), + body=request.content, + signer_keys=DiscoveredSignerKeys(fetcher=fetch), + replay_store=replay, + ) + + +# -- The web-bot-auth sender -------------------------------------------------- + + +def test_from_wba_reports_its_profile_algorithm_and_redacts_the_key() -> None: + key = _key() + sender = WebhookSender.from_wba(key, agent_origin="https://Seller.Example:443/") + + assert sender.signing_profile == WBA + assert sender.signing_profiles == frozenset({WBA}) + assert sender.signing_algorithms == frozenset({"ed25519"}) + assert sender.signs_with_rfc9421 is True + thumbprint = public_directory_key(key)["kid"] + assert repr(sender) == ( + f"WebhookSender(auth=WbaSignerStrategy, key_id={thumbprint!r}, profiles=['web-bot-auth'])" + ) + + +@pytest.mark.parametrize( + "key,origin,match", + [ + (ec.generate_private_key(ec.SECP256R1()), ORIGIN, "Ed25519 keys only"), + (_key(), "http://seller.example", "HTTPS"), + (_key(), "https://user@seller.example", "HTTPS"), + ], +) +def test_from_wba_rejects_keys_and_origins_the_profile_cannot_use( + key: Any, origin: str, match: str +) -> None: + with pytest.raises(ValueError, match=match): + WebhookSender.from_wba(key, agent_origin=origin) + + +@pytest.mark.asyncio +async def test_each_attempt_is_signed_afresh_and_verifies_against_the_directory() -> None: + """A receiver records nonces, including on an attempt it answered 503.""" + capture = _Capture(503) + sender = WebhookSender.from_wba( + _key(), agent_origin=ORIGIN, client=httpx.AsyncClient(transport=capture) + ) + directory = sender.sign_web_bot_auth_directory() + replay = InMemoryReplayStore() + + first = await sender.send_prepared(_prepared(WBA)) + retry = await sender.resend(first) + + assert (first.status_code, retry.status_code) == (503, 200) + assert retry.signing_profile == WBA + one, two = capture.requests + assert one.content == two.content + assert one.headers["signature-agent"] == f'sig1="{ORIGIN}"' + assert _tag(one) == WBA + assert one.headers["signature-input"].startswith( + 'sig1=("@method" "@target-uri" "@authority" "content-type" ' + '"content-digest" "signature-agent";key="sig1");created=' + ) + assert one.headers["signature"] != two.headers["signature"] + # Both verify against one replay store: the retry is not a replayed nonce. + for request in (one, two): + (signer,) = await _verify(request, directory, replay) + assert signer.agent_url == ORIGIN + + +@pytest.mark.asyncio +async def test_web_bot_auth_sender_refuses_registrations_for_the_adcp_profile() -> None: + capture = _Capture() + sender = WebhookSender.from_wba( + _key(), agent_origin=ORIGIN, client=httpx.AsyncClient(transport=capture) + ) + with pytest.raises(WebhookSigningProfileUnavailableError, match="adcp/webhook-signing/v1"): + await sender.send_prepared(_prepared(ADCP)) + with pytest.raises(WebhookSigningProfileUnavailableError): + sender.check_registration({"url": HOOK, "reporting_frequency": "daily"}) + assert ( + sender.check_registration( + {"url": HOOK, "reporting_frequency": "daily", "signing_profile": WBA} + ) + == WBA + ) + assert capture.requests == [] + + +@pytest.mark.asyncio +async def test_extra_headers_cannot_name_a_signature_agent() -> None: + capture = _Capture() + for sender in ( + WebhookSender.from_wba( + _key(), agent_origin=ORIGIN, client=httpx.AsyncClient(transport=capture) + ), + WebhookSender.from_jwk(_WEBHOOK_JWK, client=httpx.AsyncClient(transport=capture)), + ): + prepared = replace(_prepared(), extra_headers={"Signature-Agent": 'sig1="https://x"'}) + with pytest.raises(ValueError, match="extra_headers may not override"): + await sender.send_prepared(prepared) + assert capture.requests == [] + + +# -- A sender holding both profiles ------------------------------------------- + + +@pytest.mark.asyncio +async def test_dual_sender_signs_each_delivery_under_its_registrations_profile() -> None: + capture = _Capture() + adcp_only = WebhookSender.from_jwk(_WEBHOOK_JWK, client=httpx.AsyncClient(transport=capture)) + sender = adcp_only.with_web_bot_auth(_key(), agent_origin=ORIGIN) + + assert sender.signing_profile == ADCP + assert sender.signing_profiles == frozenset({ADCP, WBA}) + assert sender.signing_algorithms == frozenset({"ed25519"}) + assert adcp_only.signing_profiles == frozenset({ADCP}) + + await sender.send_prepared(_prepared(WBA)) + await sender.send_prepared(_prepared(ADCP)) + default = await sender.send_prepared(_prepared(None)) + await sender.resend(default) + + wba, adcp, unbound, replayed = capture.requests + assert _tag(wba) == WBA and "signature-agent" in wba.headers + for request in (adcp, unbound, replayed): + assert _tag(request) == ADCP + assert "signature-agent" not in request.headers + + +def test_with_web_bot_auth_extends_only_an_adcp_sender_once() -> None: + dual = WebhookSender.from_jwk(_WEBHOOK_JWK).with_web_bot_auth(_key(), agent_origin=ORIGIN) + with pytest.raises(ValueError, match="already signs under web-bot-auth"): + dual.with_web_bot_auth(_key(), agent_origin=ORIGIN) + with pytest.raises(ValueError, match="already signs under web-bot-auth"): + WebhookSender.from_wba(_key(), agent_origin=ORIGIN).with_web_bot_auth( + _key(), agent_origin=ORIGIN + ) + with pytest.raises(ValueError, match="extends an adcp/webhook-signing/v1 sender"): + WebhookSender.from_bearer_token("token").with_web_bot_auth(_key(), agent_origin=ORIGIN) + + +@pytest.mark.asyncio +async def test_resend_keeps_the_profile_the_original_was_signed_under() -> None: + capture = _Capture() + sender = WebhookSender.from_jwk(_WEBHOOK_JWK, client=httpx.AsyncClient(transport=capture)) + signed_wba = WebhookDeliveryResult( + status_code=503, + idempotency_key="whk_1234567890123456", + url=HOOK, + response_headers={}, + response_body=b"", + sent_body=_prepared().body, + signing_profile=WBA, + ) + with pytest.raises(WebhookSigningProfileUnavailableError): + await sender.resend(signed_wba) + assert capture.requests == [] + + +# -- Key directory ------------------------------------------------------------ + + +def test_directory_lists_the_sender_key_and_the_origins_other_keys() -> None: + key, request_key = _key(), _key() + sender = WebhookSender.from_jwk(_WEBHOOK_JWK).with_web_bot_auth(key, agent_origin=ORIGIN) + + directory = sender.sign_web_bot_auth_directory(other_keys=[request_key]) + + keys = verify_directory_response(ORIGIN, directory, now=time.time()) + assert set(keys) == { + public_directory_key(key)["kid"], + public_directory_key(request_key)["kid"], + } + + +def test_directory_needs_a_web_bot_auth_key() -> None: + with pytest.raises(ValueError, match="no web-bot-auth key"): + WebhookSender.from_jwk(_WEBHOOK_JWK).sign_web_bot_auth_directory() + + +# -- Boot validation against the advertised capability ------------------------ + + +class _Caps: + def __init__(self, webhook_signing: WebhookSigning) -> None: + self.webhook_signing = webhook_signing + self.webhook_signing_managed_externally = False + + +def _boot(sender: WebhookSender, **signing: Any) -> AdcpError: + with pytest.raises(AdcpError) as caught: + validate_webhook_signing_for_capabilities( + capabilities=_Caps( + WebhookSigning(supported=True, delivery_retry_horizon_seconds=86400, **signing) + ), + sender=sender, + supervisor=None, + ) + return caught.value + + +def test_boot_refuses_an_advertised_profile_the_sender_has_no_key_for() -> None: + error = _boot(WebhookSender.from_jwk(_WEBHOOK_JWK), profiles=[ADCP, WBA]) + assert error.details["missing"] == "webhook_signing_profile_key" + assert error.details["sender_profiles"] == [ADCP] + + +def test_boot_refuses_a_web_bot_auth_only_sender_when_profiles_are_omitted() -> None: + # Omitted profiles mean adcp/webhook-signing/v1, which this sender cannot sign. + error = _boot(WebhookSender.from_wba(_key(), agent_origin=ORIGIN), algorithms=["ed25519"]) + assert error.details["missing"] == "webhook_signing_profile_key" + assert error.details["advertised_profiles"] == [ADCP] + + +def test_boot_accepts_a_web_bot_auth_sender_for_an_advertised_profile() -> None: + sender = WebhookSender.from_jwk(_WEBHOOK_JWK).with_web_bot_auth(_key(), agent_origin=ORIGIN) + error = _boot(sender, profiles=[ADCP, WBA], algorithms=["ed25519"]) + # Past every sender check: what remains is the missing durable outbox. + assert error.details["missing"] == "external_durable_webhook_outbox" + + +def test_boot_checks_every_algorithm_a_dual_sender_signs_with() -> None: + es256_jwk = { + "kty": "EC", + "crv": "P-256", + "kid": "es256-webhook", + "alg": "ES256", + "adcp_use": "webhook-signing", + } + private = ec.generate_private_key(ec.SECP256R1()) + numbers = private.private_numbers() + + def b64(value: int) -> str: + return base64.urlsafe_b64encode(value.to_bytes(32, "big")).rstrip(b"=").decode() + + es256_jwk |= { + "x": b64(numbers.public_numbers.x), + "y": b64(numbers.public_numbers.y), + "d": b64(numbers.private_value), + } + sender = WebhookSender.from_jwk(es256_jwk).with_web_bot_auth(_key(), agent_origin=ORIGIN) + assert sender.signing_algorithms == frozenset({"ecdsa-p256-sha256", "ed25519"}) + error = _boot(sender, algorithms=["ecdsa-p256-sha256"]) + assert error.details["missing"] == "webhook_signing_algorithm_alignment" + assert error.details["sender_alg"] == "ed25519" + + +# -- Task push registrations --------------------------------------------------- + + +@pytest.mark.parametrize( + "config,expected", + [ + ({"url": HOOK}, None), + ({"url": HOOK, "signing_profile": WBA}, WBA), + ({"url": HOOK, "signing_profile": ADCP}, ADCP), + ], +) +def test_push_signing_profile_is_read_from_the_registration( + config: dict[str, Any], expected: str | None +) -> None: + assert _extract_push_signing_profile({"push_notification_config": config}) == expected + + +@pytest.mark.parametrize( + "config,match", + [ + ({"url": HOOK, "signing_profile": "custom"}, "unknown"), + ( + { + "url": HOOK, + "signing_profile": WBA, + "authentication": {"schemes": ["Bearer"], "credentials": "x" * 32}, + }, + "must not combine", + ), + ], +) +def test_malformed_push_signing_profile_is_refused(config: dict[str, Any], match: str) -> None: + with pytest.raises(ValueError, match=match): + _extract_push_signing_profile({"push_notification_config": config}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "config,refused", + [ + ({"url": HOOK, "signing_profile": WBA}, True), + ({"url": HOOK}, False), + ({"url": HOOK, "signing_profile": "custom"}, True), + ], +) +async def test_dispatch_refuses_a_profile_the_outbox_cannot_sign( + config: dict[str, Any], refused: bool +) -> None: + from adcp.decisioning.dispatch import _push_signing_profile + + registry = MagicMock() + registry.task_webhook_outbox.supports_signing_profile = lambda profile: profile == ADCP + on_failure = AsyncMock() + params = {"push_notification_config": config} + if refused: + with pytest.raises(AdcpError) as caught: + await _push_signing_profile( + params, registry, method_name="create_media_buy", on_failure=on_failure + ) + assert caught.value.code == "INVALID_REQUEST" + assert caught.value.field == "push_notification_config.signing_profile" + on_failure.assert_awaited_once() + else: + assert ( + await _push_signing_profile( + params, registry, method_name="create_media_buy", on_failure=on_failure + ) + is None + ) + on_failure.assert_not_awaited() + + +# -- Task webhook outbox ------------------------------------------------------- + + +def _task_outbox(sender: 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=MagicMock(), + sender=sender, + encryption_key=b"e" * 32, + delivery_retry_horizon_seconds=86_400, + ) + + +def test_task_outbox_accepts_a_web_bot_auth_sender_and_reports_its_profiles() -> None: + outbox = _task_outbox(WebhookSender.from_wba(_key(), agent_origin=ORIGIN)) + assert outbox.supports_signing_profile(WBA) is True + assert outbox.supports_signing_profile(ADCP) is False + + +def test_task_registration_round_trips_its_signing_profile() -> None: + outbox = _task_outbox(WebhookSender.from_jwk(_WEBHOOK_JWK)) + encrypted, nonce = outbox.protect_registration( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + url=HOOK, + operation_id="op_1", + token=None, + signing_profile=WBA, + ) + opened = outbox._open_registration_with_scope( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + encrypted_registration=encrypted, + nonce=nonce, + ) + assert opened == (HOOK, "op_1", None, None, None, WBA) + + +@pytest.mark.parametrize( + "authentication,profile,match", + [ + (TaskWebhookAuthentication(scheme="Bearer", credentials="x" * 32), WBA, "must not carry"), + (None, "custom", "unknown webhook signing_profile"), + ], +) +def test_task_registration_refuses_an_invalid_signing_profile( + authentication: Any, profile: str, match: str +) -> None: + outbox = _task_outbox(WebhookSender.from_jwk(_WEBHOOK_JWK)) + with pytest.raises(ValueError, match=match): + outbox.validate_registration(HOOK, authentication, signing_profile=profile) + + +@pytest.mark.parametrize( + "stored_profile,required", + [(WBA, WBA), (ADCP, ADCP), (None, ADCP)], +) +def test_task_delivery_envelope_carries_the_profile_without_signatures( + stored_profile: str | None, required: str +) -> None: + outbox = _task_outbox(WebhookSender.from_jwk(_WEBHOOK_JWK)) + body = _prepared().body + protected = outbox._protect_delivery_body(body, None, stored_profile) + if stored_profile != WBA: + # The default, named or not, stays a bare body an earlier SDK can send. + assert protected == body + opened_body, authentication, opened_profile = outbox._open_delivery_body(protected) + assert (opened_body, authentication) == (body, None) + assert opened_profile == (WBA if stored_profile == WBA else None) + assert b"signature" not in protected.lower() + assert (opened_profile or ADCP) == required + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stored_profile,expected", [(WBA, WBA), (None, ADCP)]) +async def test_task_outbox_sends_each_attempt_under_the_stored_profile( + stored_profile: str | None, expected: str +) -> None: + sender = MagicMock(spec=WebhookSender) + sender.signs_with_rfc9421 = True + sender._owns_client = True + sender._allow_private_destinations = False + sender._timeout = 10.0 + sender.signing_algorithms = frozenset({"ed25519"}) + sender.send_prepared = AsyncMock( + return_value=WebhookDeliveryResult( + status_code=200, + idempotency_key="k", + url=HOOK, + response_headers={}, + response_body=b"", + ) + ) + outbox = _task_outbox(sender) + prepared = _prepared() + nonce = b"n" * 12 + encrypted = outbox._cipher.encrypt( + nonce, + outbox._protect_delivery_body(prepared.body, None, stored_profile), + outbox._envelope_aad( + account_id="acct_1", + task_id="task_1", + task_type="create_media_buy", + status="completed", + url=HOOK, + operation_id="op_1", + idempotency_key=prepared.idempotency_key, + ), + ) + row = ( + 7, + "acct_1", + "task_1", + "create_media_buy", + "completed", + HOOK, + "op_1", + prepared.idempotency_key, + None, + encrypted, + nonce, + 1, + ) + + def connection(fetch: Any = None) -> Any: + conn = MagicMock() + cursor = MagicMock() + cursor.fetchone = AsyncMock(return_value=fetch) + conn.execute = AsyncMock(return_value=cursor) + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=conn) + ctx.__aexit__ = AsyncMock(return_value=False) + return ctx + + contexts = iter([connection(row), connection()]) + outbox._pool = MagicMock() + outbox._pool.connection = lambda: next(contexts) + + assert await outbox.process_one() is True + sent = sender.send_prepared.await_args.args[0] + assert sent.required_signing_profile == expected + assert sent.body == prepared.body + assert sent.extra_headers == {} + + +# -- Notification outbox ------------------------------------------------------- + + +def test_notification_outbox_stores_the_required_profile_and_reserves_signature_agent() -> None: + pytest.importorskip("psycopg_pool", reason="constructor checks require adcp[pg]") + from adcp import PgNotificationOutbox + + outbox = PgNotificationOutbox( + pool=MagicMock(), + sender=WebhookSender.from_wba(_key(), agent_origin=ORIGIN), + encryption_key=b"x" * 32, + delivery_retry_horizon_seconds=86400, + ) + raw = WebhookSender.prepare_raw(url=HOOK, payload={"notification_type": "extension.changed"}) + prepared = replace(raw, required_signing_profile=WBA) + params, _scope = outbox._prepare_insert( + prepared, + notification_type="extension.changed", + caller_scope_id="caller", + account_id=None, + signing_scope_id=None, + ) + kind, caller, account, url, key, signing_scope, key_id, scope, encrypted, nonce = params[:10] + row = (1, account, caller, kind, url, key, signing_scope, key_id, scope, encrypted, nonce) + assert outbox._open(row) == prepared + # Version 2 makes an earlier SDK quarantine the row instead of ignoring + # the profile; rows without one keep version 1. + assert json.loads(PgNotificationOutbox._protected(prepared))["version"] == 2 + assert json.loads(PgNotificationOutbox._protected(raw))["version"] == 1 + # The default profile stays readable to an earlier SDK, which signs only it. + default = replace(raw, required_signing_profile=ADCP) + assert json.loads(PgNotificationOutbox._protected(default))["version"] == 1 + params, _scope = outbox._prepare_insert( + default, + notification_type="extension.changed", + caller_scope_id="caller", + account_id=None, + signing_scope_id=None, + ) + kind, caller, account, url, key, signing_scope, key_id, scope, encrypted, nonce = params[:10] + row = (1, account, caller, kind, url, key, signing_scope, key_id, scope, encrypted, nonce) + assert outbox._open(row) == default + + with pytest.raises(ValueError): + PgNotificationOutbox._protected( + replace(raw, extra_headers={"Signature-Agent": 'sig1="https://x"'}) + ) + + +# -- Advertised profiles, checked for every outbox ------------------------------ + + +@pytest.mark.parametrize( + "webhook_signing,expected", + [ + (None, None), + (WebhookSigning(supported=False), None), + (WebhookSigning(supported=True), frozenset({ADCP})), + (WebhookSigning(supported=True, profiles=[ADCP, WBA]), frozenset({ADCP, WBA})), + ], +) +def test_advertised_profiles_default_to_the_adcp_profile( + webhook_signing: WebhookSigning | None, expected: frozenset[str] | None +) -> None: + from adcp.decisioning.webhook_emit import advertised_webhook_signing_profiles + + caps = SimpleNamespace(webhook_signing=webhook_signing) + assert advertised_webhook_signing_profiles(caps) == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "config,advertised,refused", + [ + ({"url": HOOK, "signing_profile": WBA}, frozenset({ADCP}), True), + ({"url": HOOK, "signing_profile": WBA}, frozenset({ADCP, WBA}), False), + # The implicit default must be advertised too. + ({"url": HOOK}, frozenset({WBA}), True), + # Legacy authentication selects no signing profile. + ( + {"url": HOOK, "authentication": {"schemes": ["Bearer"], "credentials": "x" * 32}}, + frozenset({WBA}), + False, + ), + ({"url": HOOK, "signing_profile": WBA}, None, False), + ], +) +async def test_resolver_outbox_registrations_are_checked_against_advertised_profiles( + config: dict[str, Any], advertised: frozenset[str] | None, refused: bool +) -> None: + """A sender_resolver outbox cannot vouch for a profile, so the capabilities decide.""" + from adcp.decisioning.dispatch import _push_signing_profile + + registry = MagicMock() + registry.task_webhook_outbox.supports_signing_profile = lambda profile: None + params = {"push_notification_config": config} + call = _push_signing_profile( + params, + registry, + method_name="create_media_buy", + on_failure=None, + advertised_profiles=advertised, + ) + if refused: + with pytest.raises(AdcpError) as caught: + await call + assert caught.value.field == "push_notification_config.signing_profile" + else: + await call From 823d091997534efab39cf8c577a64777e5d99bc0 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sat, 7 Nov 2026 22:52:17 +0000 Subject: [PATCH 2/3] fix(webhooks): keep pre-signing_algorithms sender doubles working in the outboxes The outboxes read the new WebhookSender.signing_algorithms (and the task outbox signing_profiles) directly. A sender-like object without them - the Postgres conformance tests' MagicMock senders, and adopters' own test doubles - failed the check and its rows were quarantined. Both outboxes and the boot validator now share _sender_signing_algorithms and _sender_signing_profiles, which fall back to _auth.alg and, for an RFC 9421 sender, adcp/webhook-signing/v1, as the boot check already did. The Postgres conformance mock senders also declare both properties. Co-Authored-By: Claude Opus 5.5 --- .../decisioning/pg/task_webhook_outbox.py | 6 ++-- src/adcp/decisioning/webhook_emit.py | 13 ++------- src/adcp/notification_outbox_pg.py | 3 +- src/adcp/webhook_sender.py | 29 +++++++++++++++++++ .../test_pg_notification_outbox.py | 2 ++ .../test_pg_task_webhook_outbox.py | 2 ++ tests/test_webhook_sender_wba.py | 19 ++++++++++++ 7 files changed, 61 insertions(+), 13 deletions(-) diff --git a/src/adcp/decisioning/pg/task_webhook_outbox.py b/src/adcp/decisioning/pg/task_webhook_outbox.py index 2d51dc3d88..30bb75419e 100644 --- a/src/adcp/decisioning/pg/task_webhook_outbox.py +++ b/src/adcp/decisioning/pg/task_webhook_outbox.py @@ -37,6 +37,8 @@ WebhookSender, WebhookSenderResolution, WebhookSenderResolver, + _sender_signing_algorithms, + _sender_signing_profiles, ) from adcp.webhook_supervisor import RetryPolicy from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILES @@ -373,7 +375,7 @@ def supports_signing_profile(self, profile: str) -> bool | None: """ if self._sender is None: return None - return profile in self._sender.signing_profiles + return profile in _sender_signing_profiles(self._sender) def protect_registration( self, @@ -839,7 +841,7 @@ async def _resolve_delivery_sender(self, signing_scope_id: str | None) -> Webhoo raise ValueError("resolver returned an invalid sender resolution") sender = resolution.sender self._validate_delivery_sender(sender) - sender_algorithms = sender.signing_algorithms + sender_algorithms = _sender_signing_algorithms(sender) if not sender_algorithms or not sender_algorithms <= resolution.advertised_algorithms: raise ValueError("resolved sender algorithm was not advertised for its scope") sender_timeout = float(getattr(sender, "_timeout", 0.0)) diff --git a/src/adcp/decisioning/webhook_emit.py b/src/adcp/decisioning/webhook_emit.py index 417b361448..b95f7389b3 100644 --- a/src/adcp/decisioning/webhook_emit.py +++ b/src/adcp/decisioning/webhook_emit.py @@ -721,7 +721,9 @@ def validate_webhook_signing_for_capabilities( # skip the cross-check when omitted (no advertisement to violate). advertised_algorithms = getattr(webhook_signing, "algorithms", None) if advertised_algorithms and sender_introspectable: - sender_algs = _sender_algorithms(resolved_sender) + from adcp.webhook_sender import _sender_signing_algorithms + + sender_algs = _sender_signing_algorithms(resolved_sender) advertised_alg_values = [getattr(a, "value", a) for a in advertised_algorithms] unadvertised = sorted(alg for alg in sender_algs if alg not in advertised_alg_values) if not sender_algs or unadvertised: @@ -810,15 +812,6 @@ def advertised_webhook_signing_profiles(capabilities: Any) -> frozenset[str] | N return frozenset(str(getattr(profile, "value", profile)) for profile in profiles) -def _sender_algorithms(sender: Any) -> frozenset[str]: - """Every ``alg`` a sender's signatures may carry; duck-typed for supervisors.""" - algorithms = getattr(sender, "signing_algorithms", None) - if isinstance(algorithms, frozenset): - return algorithms - alg = getattr(getattr(sender, "_auth", None), "alg", None) - return frozenset({alg}) if isinstance(alg, str) else frozenset() - - def external_task_webhook_owner_ready( *, capabilities: DecisioningCapabilities, diff --git a/src/adcp/notification_outbox_pg.py b/src/adcp/notification_outbox_pg.py index 412e963fa0..82d7682048 100644 --- a/src/adcp/notification_outbox_pg.py +++ b/src/adcp/notification_outbox_pg.py @@ -39,6 +39,7 @@ WebhookSender, WebhookSenderResolution, WebhookSenderResolver, + _sender_signing_algorithms, ) from adcp.webhook_supervisor import RetryPolicy from adcp.webhooks import WEBHOOK_SIGNING_PROFILE_ADCP, WEBHOOK_SIGNING_PROFILES @@ -685,7 +686,7 @@ async def _resolve_sender(self, scope: str | None) -> WebhookSender: if not isinstance(resolution, WebhookSenderResolution): raise ValueError("invalid sender resolution") self._validate_sender(resolution.sender) - algorithms = resolution.sender.signing_algorithms + algorithms = _sender_signing_algorithms(resolution.sender) if not algorithms or not algorithms <= resolution.advertised_algorithms: raise ValueError("resolved signing algorithm is not advertised") except Exception: diff --git a/src/adcp/webhook_sender.py b/src/adcp/webhook_sender.py index cdb1fa7e48..49f214c15e 100644 --- a/src/adcp/webhook_sender.py +++ b/src/adcp/webhook_sender.py @@ -165,6 +165,35 @@ def _profile_strategies_for(auth: WebhookAuthStrategy) -> dict[str, WebhookAuthS return {} +def _sender_signing_algorithms(sender: Any) -> frozenset[str]: + """Every ``alg`` a sender's signatures may carry. + + Reads :attr:`WebhookSender.signing_algorithms`, falling back to the + single ``_auth.alg`` for a sender-like object without it (a test double, + or a supervisor's sender built before the property existed). + """ + algorithms = getattr(sender, "signing_algorithms", None) + if isinstance(algorithms, frozenset): + return algorithms + alg = getattr(getattr(sender, "_auth", None), "alg", None) + return frozenset({alg}) if isinstance(alg, str) else frozenset() + + +def _sender_signing_profiles(sender: Any) -> frozenset[str]: + """Every webhook-signing profile a sender can sign under. + + Reads :attr:`WebhookSender.signing_profiles`, falling back for a + sender-like object without it to ``adcp/webhook-signing/v1`` when it + signs with RFC 9421 (the only profile before ``web-bot-auth``), else none. + """ + profiles = getattr(sender, "signing_profiles", None) + if isinstance(profiles, frozenset): + return profiles + if getattr(sender, "signs_with_rfc9421", False) is True: + return frozenset({WEBHOOK_SIGNING_PROFILE_ADCP}) + return frozenset() + + def _wba_strategy(private_key: Any, agent_origin: str) -> WbaSignerStrategy: """Validate a Web Bot Auth key and origin up front, at construction.""" from adcp.signing.wba import _https_origin, public_directory_key diff --git a/tests/conformance/decisioning/test_pg_notification_outbox.py b/tests/conformance/decisioning/test_pg_notification_outbox.py index a838de5856..55529f780d 100644 --- a/tests/conformance/decisioning/test_pg_notification_outbox.py +++ b/tests/conformance/decisioning/test_pg_notification_outbox.py @@ -115,6 +115,8 @@ def sender_mock() -> MagicMock: sender._timeout = 10 sender.signs_with_rfc9421 = True sender._auth.alg = "ed25519" + sender.signing_algorithms = frozenset({"ed25519"}) + sender.signing_profiles = frozenset({"adcp/webhook-signing/v1"}) async def send( value: PreparedWebhook, diff --git a/tests/conformance/decisioning/test_pg_task_webhook_outbox.py b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py index 197b58b378..d64dab05a0 100644 --- a/tests/conformance/decisioning/test_pg_task_webhook_outbox.py +++ b/tests/conformance/decisioning/test_pg_task_webhook_outbox.py @@ -42,6 +42,8 @@ def _sender() -> MagicMock: sender._timeout = 10.0 sender.signs_with_rfc9421 = True sender._auth.alg = "ed25519" + sender.signing_algorithms = frozenset({"ed25519"}) + sender.signing_profiles = frozenset({"adcp/webhook-signing/v1"}) def prepare_mcp(**kwargs: Any) -> PreparedWebhook: key = f"whk_{uuid.uuid4().hex}" diff --git a/tests/test_webhook_sender_wba.py b/tests/test_webhook_sender_wba.py index b7687d4f76..f32f0c58d5 100644 --- a/tests/test_webhook_sender_wba.py +++ b/tests/test_webhook_sender_wba.py @@ -686,3 +686,22 @@ async def test_resolver_outbox_registrations_are_checked_against_advertised_prof assert caught.value.field == "push_notification_config.signing_profile" else: await call + + +def test_sender_like_objects_without_the_new_properties_keep_working() -> None: + """Test doubles predating signing_algorithms / signing_profiles fall back.""" + from adcp.webhook_sender import _sender_signing_algorithms, _sender_signing_profiles + + legacy_double = MagicMock() + legacy_double.signs_with_rfc9421 = True + legacy_double._auth.alg = "ed25519" + assert _sender_signing_algorithms(legacy_double) == frozenset({"ed25519"}) + assert _sender_signing_profiles(legacy_double) == frozenset({ADCP}) + + unsigned = SimpleNamespace(signs_with_rfc9421=False) + assert _sender_signing_algorithms(unsigned) == frozenset() + assert _sender_signing_profiles(unsigned) == frozenset() + + dual = WebhookSender.from_jwk(_WEBHOOK_JWK).with_web_bot_auth(_key(), agent_origin=ORIGIN) + assert _sender_signing_profiles(dual) == frozenset({ADCP, WBA}) + assert _sender_signing_algorithms(dual) == frozenset({"ed25519"}) From db7756b2656a7a095ab20aea6419bda7ab91fdc9 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 8 Nov 2026 02:36:16 +0000 Subject: [PATCH 3/3] fix(reporting): log a skipped web-bot-auth subscription as an error A web-bot-auth subscription that production signing cannot sign is skipped so it does not block the account's other subscribers. That subscriber then receives no notifications, so log the skip at error level (the level the outboxes use for a quarantined delivery) with the subscriber and account ids, instead of a warning that can be lost under volume. Co-Authored-By: Claude Opus 5.5 --- docs/reporting-production.md | 5 +++-- src/adcp/reporting/production/notifications.py | 16 +++++++++++----- tests/test_reporting_webhook_wba.py | 14 +++++++++++--- 3 files changed, 25 insertions(+), 10 deletions(-) diff --git a/docs/reporting-production.md b/docs/reporting-production.md index 55c9465dc7..9a4fe5cd15 100644 --- a/docs/reporting-production.md +++ b/docs/reporting-production.md @@ -145,8 +145,9 @@ algorithm set, no legacy fallback, and an 86,400-second retry horizon. Pass 3.3): the declaration then lists both `profiles`, `ed25519` must be among the algorithms, and every resolved `ReportingSigningMaterial` must carry `web_bot_auth=ReportingWebBotAuthKey(key, agent_origin)`. Without it, a -`web-bot-auth` subscription is skipped with a warning, and the account's other -subscribers are still notified. See +`web-bot-auth` subscription is skipped and logged as an error (alert on it: +that subscriber receives nothing), and the account's other subscribers are +still notified. See [Web Bot Auth webhooks](webhooks/web-bot-auth.md#reporting-notifications). Each queue reserves the first HTTP attempt and its key/body binding before peer I/O, using database time in PostgreSQL. Window insertion, HTTP attempt and ordinal diff --git a/src/adcp/reporting/production/notifications.py b/src/adcp/reporting/production/notifications.py index 93a772d5b6..71f5efcd93 100644 --- a/src/adcp/reporting/production/notifications.py +++ b/src/adcp/reporting/production/notifications.py @@ -126,13 +126,19 @@ def _signable(self, value: ReportingNotificationSubscription) -> bool: A ``web-bot-auth`` subscription without ``web_bot_auth`` signing is skipped, not raised: one such subscriber must not stop deliveries to - every other subscriber of the account. The registration service - should refuse it, since the capability does not list the profile. + every other subscriber of the account. That subscriber then receives + no notifications, so the skip is logged as an error, the level the + outboxes use for a quarantined delivery, for operators to alert on. + The registration service should refuse such a subscription, since + the capability does not list the profile. """ if value.signing_profile == WEBHOOK_SIGNING_PROFILE_WEB_BOT_AUTH and not self.web_bot_auth: - logger.warning( - "[adcp.reporting] skipping a web-bot-auth subscription: production " - "signing is not configured with web_bot_auth=True" + logger.error( + "[adcp.reporting] skipping web-bot-auth subscription %s for account %s: " + "production signing is not configured with web_bot_auth=True, so this " + "subscriber receives no notifications", + value.subscriber_id, + value.account_id, ) return False return True diff --git a/tests/test_reporting_webhook_wba.py b/tests/test_reporting_webhook_wba.py index e5065652b6..83a370a26f 100644 --- a/tests/test_reporting_webhook_wba.py +++ b/tests/test_reporting_webhook_wba.py @@ -11,6 +11,7 @@ import hashlib import json +import logging from dataclasses import asdict from types import SimpleNamespace from typing import Any @@ -229,7 +230,9 @@ async def test_production_web_bot_auth_requires_the_key_in_every_material() -> N @pytest.mark.asyncio -async def test_production_skips_only_the_web_bot_auth_subscriptions_it_cannot_sign() -> None: +async def test_production_skips_only_the_web_bot_auth_subscriptions_it_cannot_sign( + caplog: pytest.LogCaptureFixture, +) -> None: wba = _subscription(subscriber_id="wba_buyer", signing_profile=WBA) adcp = _subscription(subscriber_id="adcp_buyer") @@ -242,8 +245,13 @@ async def get_active(self, **_: Any) -> ReportingNotificationSubscription: query = {"account_id": "acct_a", "notification_type": "reporting.ledger_changed"} without = _SignedSubscriptions(Subscriptions()) - # One unsignable subscriber does not block the account's other subscribers. - assert await without.list_active(**query) == (adcp,) + # One unsignable subscriber does not block the account's other subscribers, + # and the skip is an error-level log naming it, for operators to alert on. + with caplog.at_level(logging.ERROR, logger="adcp.reporting.production.notifications"): + assert await without.list_active(**query) == (adcp,) + (record,) = caplog.records + assert record.levelno == logging.ERROR + assert "wba_buyer" in record.getMessage() and "acct_a" in record.getMessage() assert await without.get_active(subscriber_id="wba_buyer", **query) is None enabled = _SignedSubscriptions(Subscriptions(), web_bot_auth=True)