feat: add webhook proof-of-control helper

README.md

@@ -528,22 +528,45 @@ kind (`send_revocation_notification`, `send_artifact_webhook`,
 Validate buyer-provided webhook URLs before storing durable subscriptions:
 
 ```python
-from adcp.webhooks import WebhookDestinationPolicy, validate_webhook_destination_url
-
-validate_webhook_destination_url(
-    request.push_notification_config.url,
-    field="push_notification_config.url",
-    policy=WebhookDestinationPolicy.production(),
+from adcp import (
+    WebhookChallengeError,
+    WebhookDestinationPolicy,
+    WebhookSender,
+    challenge_webhook_destination,
 )
+
+sender = WebhookSender.from_jwk(webhook_signing_jwk_with_private_d)
+
+try:
+    # Call only for new or changed active durable subscriptions. Persist
+    # active=False configs without challenging; challenge before activation.
+    auth_kwargs = (
+        {"authentication": config.authentication}
+        if config.authentication
+        else {"sender": sender}
+    )
+    await challenge_webhook_destination(
+        url=config.url,
+        account_id=account_id,
+        subscriber_id=config.subscriber_id,
+        **auth_kwargs,
+        field="accounts[0].notification_configs[0].url",
+        policy=WebhookDestinationPolicy.production(),
+    )
+except WebhookChallengeError as exc:
+    return {"errors": [exc.to_error()]}
 ```
 
 Use `WebhookDestinationPolicy.local_development()` only for local tests that
 need `http://localhost` or private-network destinations. Production validation
 requires HTTPS and rejects loopback, private, link-local, reserved, and cloud
-metadata destinations using the same SSRF classifier as `WebhookSender`. The
-validation result includes both `original_url` and `effective_url`; sellers
-should normally persist the buyer's original URL and reapply the same
-policy/hooks at send time, rather than storing a Docker or test rewrite.
+metadata destinations. The challenge helper uses the SDK-managed pinned
+transport and fails unless the receiver echoes the challenge in a JSON
+`challenge` or `token` field. For omitted `authentication`, pass an RFC 9421
+`WebhookSender.from_jwk(...)` without `client=`. Bearer/HMAC durable configs
+must pass `authentication=config.authentication`; custom egress clients,
+proxies, and mTLS transports are for normal delivery paths, not registration
+challenges.
 
 Wholesale feed notifications use stable types from `adcp` / `adcp.types`:
 

docs/handler-authoring.md

@@ -1220,15 +1220,46 @@ Validate durable buyer endpoints before persisting `push_notification_config.url
 or `accounts[].notification_configs[].url` from `sync_accounts`:
 
 ```python
-from adcp.webhooks import (
+from adcp import (
+    WebhookChallengeError,
     WebhookDestinationPolicy,
-    WebhookDestinationValidationError,
-    validate_webhook_destination_url,
+    WebhookSender,
+    challenge_webhook_destination,
 )
 
+sender = WebhookSender.from_jwk(webhook_signing_jwk_with_private_d)
+
+try:
+    if config.active is not False:
+        # Challenge new or changed active durable subscriptions before
+        # activation. Inactive configs are stored without a network challenge.
+        auth_kwargs = (
+            {"authentication": config.authentication}
+            if config.authentication
+            else {"sender": sender}
+        )
+        await challenge_webhook_destination(
+            url=str(config.url),
+            account_id=account_id,
+            subscriber_id=config.subscriber_id,
+            **auth_kwargs,
+            policy=WebhookDestinationPolicy.production(),
+            field="accounts[0].notification_configs[0].url",
+        )
+except WebhookChallengeError as exc:
+    return {"errors": [exc.to_error()]}
+```
+
+If you only need to preflight a URL before deciding whether the subscription
+changed, use `validate_webhook_destination_url` directly:
+
+```python
+from adcp import WebhookDestinationPolicy
+from adcp.webhooks import WebhookDestinationValidationError, validate_webhook_destination_url
+
 try:
-    validate_webhook_destination_url(
-        config.url,
+    validation = validate_webhook_destination_url(
+        str(config.url),
         field="accounts[0].notification_configs[0].url",
         policy=WebhookDestinationPolicy.production(),
     )
@@ -1241,8 +1272,14 @@ reserved, and cloud metadata destinations. Use
 `WebhookDestinationPolicy.local_development()` only for local fixtures that
 need `http://localhost` or private-network endpoints. The helper returns both
 `original_url` and `effective_url`; persist the buyer's original URL in durable
-subscription state, and reapply the same policy/hooks when sending. Do not
-persist a Docker or test rewrite as the buyer's registered endpoint.
+subscription state. The proof-of-control helper uses the SDK-managed pinned
+transport, posts a `webhook.challenge`, and requires the receiver to echo the
+challenge value in a JSON `challenge` or `token` field. For omitted
+`authentication`, pass an RFC 9421 `WebhookSender.from_jwk(...)` without
+`client=`. Bearer/HMAC durable configs must pass
+`authentication=config.authentication`; custom egress clients, proxies, and
+mTLS transports are reserved for normal delivery paths, not registration
+proof-of-control.
 
 ### Wholesale feed notifications
 

src/adcp/__init__.py

@@ -487,18 +487,26 @@
 from adcp.webhooks import (
     LegacyHmacFallback,
     MemoryBackend,
+    WebhookChallengeError,
+    WebhookChallengeResult,
     WebhookDedupStore,
+    WebhookDestinationPolicy,
     WebhookReceiver,
     WebhookReceiverConfig,
+    WebhookSender,
     WebhookVerifyOptions,
+    challenge_webhook_destination,
     create_a2a_webhook_payload,
     create_mcp_webhook_payload,
+    create_webhook_challenge_payload,
     extract_webhook_result_data,
+    generate_webhook_challenge_value,
     generate_webhook_idempotency_key,
     get_adcp_signed_headers_for_webhook,
     sign_legacy_webhook,
     sign_webhook,
     to_wire_dict,
+    validate_webhook_challenge_response,
 )
 
 try:
@@ -670,14 +678,22 @@ def get_adcp_version() -> str:
     # Webhook utilities
     "create_mcp_webhook_payload",
     "create_a2a_webhook_payload",
+    "create_webhook_challenge_payload",
+    "challenge_webhook_destination",
+    "validate_webhook_challenge_response",
     "get_adcp_signed_headers_for_webhook",
     "extract_webhook_result_data",
+    "generate_webhook_challenge_value",
     "generate_webhook_idempotency_key",
     "sign_legacy_webhook",
     "sign_webhook",
     "to_wire_dict",
+    "WebhookChallengeError",
+    "WebhookChallengeResult",
+    "WebhookDestinationPolicy",
     "WebhookReceiver",
     "WebhookReceiverConfig",
+    "WebhookSender",
     "WebhookVerifyOptions",
     "WebhookDedupStore",
     "MemoryBackend",

src/adcp/webhook_auth.py

@@ -49,7 +49,7 @@
 # the strategy emits cannot be overridden by an extra_headers payload,
 # nor can Content-Type (changing it would break body framing on the
 # receiver).
-_BASE_RESERVED: frozenset[str] = frozenset({"content-type"})
+_BASE_RESERVED: frozenset[str] = frozenset({"content-length", "content-type", "host"})
 
 
 class WebhookAuthStrategy(Protocol):
@@ -203,7 +203,8 @@ def merge_extra_headers(
     if not extra:
         return merged
     for key in extra:
-        if str(key).lower() in reserved:
+        normalized = str(key).lower()
+        if normalized in reserved or normalized.startswith(":"):
             raise ValueError(
                 f"extra_headers may not override auth-binding or content-type " f"header {key!r}"
             )

src/adcp/webhook_sender.py

@@ -74,6 +74,7 @@
 )
 from adcp.webhooks import (
     create_mcp_webhook_payload,
+    create_webhook_challenge_payload,
     generate_webhook_idempotency_key,
     to_wire_dict,
 )
@@ -836,6 +837,41 @@ async def send_wholesale_feed_to_subscription(
             extra_headers=extra_headers,
         )
 
+    async def send_webhook_challenge(
+        self,
+        *,
+        url: str,
+        account_id: str,
+        subscriber_id: str,
+        challenge: str | None = None,
+        extra_headers: Mapping[str, str] | None = None,
+    ) -> WebhookDeliveryResult:
+        """POST a signed durable-subscription proof-of-control challenge.
+
+        The body matches the durable ``notification_configs[]`` challenge
+        shape and intentionally does not inject ``idempotency_key``:
+
+        ``{"type":"webhook.challenge","challenge":"...", ...}``
+
+        Pair this low-level sender method with
+        :func:`adcp.webhooks.challenge_webhook_destination` when you also
+        want URL validation and response echo checking in one call.
+        """
+
+        payload = create_webhook_challenge_payload(
+            account_id=account_id,
+            subscriber_id=subscriber_id,
+            challenge=challenge,
+        )
+        challenge_value = str(payload["challenge"])
+        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+        return await self._send_bytes(
+            url=url,
+            body=body,
+            idempotency_key=challenge_value,
+            extra_headers=extra_headers,
+        )
+
     async def send_raw(
         self,
         *,