diff --git a/docs/adcp-3.3-sdk-inventory.md b/docs/adcp-3.3-sdk-inventory.md index 344be33ee5..c772a48075 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..9a4fe5cd15 100644 --- a/docs/reporting-production.md +++ b/docs/reporting-production.md @@ -140,7 +140,15 @@ 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 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 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 504440e1e9..72e6f94500 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..30bb75419e 100644 --- a/src/adcp/decisioning/pg/task_webhook_outbox.py +++ b/src/adcp/decisioning/pg/task_webhook_outbox.py @@ -37,8 +37,11 @@ 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 if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool @@ -164,14 +167,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 +182,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 +204,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 +219,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 +293,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 +332,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 +359,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 _sender_signing_profiles(self._sender) def protect_registration( self, @@ -365,31 +388,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 +450,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 +475,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 +498,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 +527,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 +601,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 +629,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 +766,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 +841,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(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)) if self._lease_seconds < sender_timeout + 5: @@ -874,58 +916,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..b95f7389b3 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,13 @@ 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) + 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] - 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 +748,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 +796,22 @@ 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 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..82d7682048 100644 --- a/src/adcp/notification_outbox_pg.py +++ b/src/adcp/notification_outbox_pg.py @@ -39,8 +39,10 @@ WebhookSender, WebhookSenderResolution, WebhookSenderResolver, + _sender_signing_algorithms, ) 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 +417,7 @@ def _protected(prepared: PreparedWebhook) -> bytes: "signature", "signature-input", "content-digest", + "signature-agent", } ), ) @@ -423,14 +426,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 +627,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 +635,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 +647,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 +686,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 = _sender_signing_algorithms(resolution.sender) + 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..71f5efcd93 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,35 @@ 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. 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.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 + 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 +157,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 +185,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 +260,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..49f214c15e 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,68 @@ 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 _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 + + 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 +284,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 +319,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 +451,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 +481,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 +630,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 +888,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. + + Compare with ``webhook_signing.profiles`` in the seller's + capabilities: a registration naming a profile outside this set is + refused with :class:`WebhookSigningProfileUnavailableError`. + """ + return frozenset(self._profile_strategies) + + @property + def signing_algorithms(self) -> frozenset[str]: + """The RFC 9421 ``alg`` values this sender's signatures may carry. - ``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``. + Empty for the bearer and HMAC senders. Compare with + ``webhook_signing.algorithms`` in the seller's capabilities. """ - return WEBHOOK_SIGNING_PROFILE_ADCP if self.signs_with_rfc9421 else None + 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 +952,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 +1241,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 +2221,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 +2248,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 +2259,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 +2296,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 +2314,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 +2380,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 +2389,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/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_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..83a370a26f --- /dev/null +++ b/tests/test_reporting_webhook_wba.py @@ -0,0 +1,259 @@ +"""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 +import logging +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( + caplog: pytest.LogCaptureFixture, +) -> 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, + # 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) + 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..f32f0c58d5 --- /dev/null +++ b/tests/test_webhook_sender_wba.py @@ -0,0 +1,707 @@ +"""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 + + +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"})