fix(governance): require crit markers with their claims, and run the signing tests CI skips

pyproject.toml

@@ -158,6 +158,12 @@ dev = [
     # tests/test_mcp_middleware_composition.py and future integration
     # tests that exercise the streamable-HTTP ASGI app in-process.
     "asgi-lifespan>=2.1.0",
+    # The signing suite's end-to-end tests serve a FastAPI app, and its
+    # Standard Webhooks interop tests verify with svix. Each skips its module
+    # (pytest.importorskip) when the package is missing, so without these
+    # CI never runs them (#1524).
+    "fastapi>=0.100",
+    "svix>=1.0",
     # mypy stubs for the protobuf runtime we use via a2a-sdk 1.0
     # (``google.protobuf.json_format``, ``struct_pb2``, ``timestamp_pb2``).
     # Without these mypy flags every import as ``import-untyped``.

src/adcp/governance.py

@@ -1145,7 +1145,11 @@ def reject(
     except (TypeError, ValueError):
         return reject("governance_token_invalid", "governance_context contains invalid JSON")
     alg = header.get("alg")
-    if alg not in {"EdDSA", "ES256"} or header.get("typ") != GOVERNANCE_AUTHORIZATION_TYP:
+    if (
+        type(alg) is not str
+        or alg not in {"EdDSA", "ES256"}
+        or header.get("typ") != GOVERNANCE_AUTHORIZATION_TYP
+    ):
         return reject("governance_token_invalid", "unsupported governance JWS alg or typ")
     critical = header.get("crit")
     if (
@@ -1156,9 +1160,10 @@ def reject(
     ):
         return reject("governance_token_invalid", "invalid governance JWS crit header")
     for name in GOVERNANCE_AUTHORIZATION_CRITICAL_CLAIMS:
-        has_claim = name in claims
-        has_marker = name in critical and header.get(name) is True
-        if has_claim != has_marker:
+        # security.mdx checklist step 4: the marker, its crit entry, and its
+        # claim appear together, and the marker is true.
+        together = (name in claims) == (name in critical) == (name in header)
+        if not together or header.get(name, True) is not True:
             return reject(
                 "governance_token_invalid", f"{name} claim and critical marker must match"
             )

src/adcp/signing/crypto.py

@@ -40,9 +40,19 @@
 
 
 def b64url_decode(s: str) -> bytes:
-    """Decode base64url, tolerating missing padding."""
-    pad = "=" * (-len(s) % 4)
-    return base64.urlsafe_b64decode(s + pad)
+    """Decode RFC 7515 base64url: the URL-safe alphabet, no padding, one spelling.
+
+    RFC 4648 section 3.3 requires refusing characters outside the alphabet,
+    so padding, ``+``, ``/``, and any other character raise ``ValueError``,
+    and so do non-zero trailing bits (section 3.5), which would give the same
+    bytes a second spelling.
+    """
+    if "=" in s or "+" in s or "/" in s:
+        raise ValueError("not unpadded base64url")
+    raw = base64.b64decode(s + "=" * (-len(s) % 4), altchars=b"-_", validate=True)
+    if b64url_encode(raw) != s:
+        raise ValueError("non-canonical base64url")
+    return raw
 
 
 def b64url_encode(b: bytes) -> str:

tests/conformance/signing/test_crypto.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import json
+import string
 from pathlib import Path
 
 import pytest
@@ -161,3 +162,32 @@ def test_jwk_with_wrong_length_x_rejected() -> None:
     }
     with pytest.raises(ValueError, match="length"):
         public_key_from_jwk(bad)
+
+
+_B64URL = string.ascii_uppercase + string.ascii_lowercase + string.digits + "-_"
+
+
+@pytest.mark.parametrize(
+    "spoil",
+    [
+        lambda x: x[:10] + "!!" + x[10:],  # characters outside the alphabet
+        lambda x: x[:10] + "+/" + x[12:],  # the standard alphabet
+        lambda x: x + "=",  # padding
+        lambda x: x[:-1] + _B64URL[_B64URL.index(x[-1]) ^ 1],  # same bytes, non-zero pad bits
+    ],
+    ids=["alphabet", "standard-alphabet", "padding", "trailing-bits"],
+)
+def test_jwk_members_are_strict_base64url(spoil) -> None:
+    """RFC 7515 section 2 base64url: no padding, no other characters, one spelling.
+
+    Verifier step 8 decodes the signer's JWK with ``public_key_from_jwk``, so a
+    lenient decoder would accept a key whose members a strict verifier refuses.
+    """
+    from adcp.signing.crypto import b64url_decode
+
+    jwk = {k: v for k, v in KEYS["test-ed25519-2026"].items() if not k.startswith("_")}
+    assert public_key_from_jwk(jwk)
+    with pytest.raises(ValueError):
+        public_key_from_jwk({**jwk, "x": spoil(jwk["x"])})
+    with pytest.raises(ValueError):
+        b64url_decode(spoil(jwk["x"]))

tests/conformance/signing/test_e2e_fastapi.py

@@ -64,8 +64,14 @@ async def create_media_buy(request: Request) -> Any:
     return app
 
 
-@pytest.mark.parametrize("policy,cover_digest", [("either", False), ("required", True)])
-async def test_signed_request_verifies_end_to_end(policy: str, cover_digest: bool) -> None:
+# AdCP 3.2 signers must cover content-digest over a non-empty body, so the
+# digest-off case signs with the 3.1 profile, the verifier's default.
+@pytest.mark.parametrize(
+    "policy,cover_digest,profile", [("either", False, "3.1"), ("required", True, "3.2")]
+)
+async def test_signed_request_verifies_end_to_end(
+    policy: str, cover_digest: bool, profile: str
+) -> None:
     app = _build_app(policy)
     private_key = private_key_from_jwk(ED25519_KEY, d_field="_private_d_for_test_only")
 
@@ -82,7 +88,7 @@ async def test_signed_request_verifies_end_to_end(policy: str, cover_digest: boo
         key_id="test-ed25519-2026",
         alg="ed25519",
         cover_content_digest=cover_digest,
-        signing_profile_version="3.2",
+        signing_profile_version=profile,
     )
     request_headers = {**headers, **signed.as_dict()}
 

tests/conformance/signing/test_webhook_e2e_fastapi.py

@@ -138,7 +138,11 @@ async def test_signed_webhook_verifies_end_to_end() -> None:
 @pytest.mark.asyncio
 async def test_duplicate_retry_dedupes_over_real_http() -> None:
     """The sender's retry MUST get a 200 with duplicate=True, and the
-    receiver MUST NOT reprocess. Real HTTP roundtrip twice."""
+    receiver MUST NOT reprocess. Real HTTP roundtrip twice.
+
+    A retry repeats the body under a fresh signature, as
+    ``WebhookSender.resend`` does: the receiver refuses a replayed nonce
+    before it reaches deduplication."""
     app, _ = _build_app()
 
     payload = create_mcp_webhook_payload(
@@ -149,12 +153,13 @@ async def test_duplicate_retry_dedupes_over_real_http() -> None:
         idempotency_key="whk_e2e_dupeaaaaaaaaaaaaaaa",
     )
     body = json.dumps(to_wire_dict(payload)).encode("utf-8")
-    headers = _sign_and_send_body(body)
 
     transport = httpx.ASGITransport(app=app)
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
-        first = await client.post("/webhooks/adcp", content=body, headers=headers)
-        second = await client.post("/webhooks/adcp", content=body, headers=headers)
+        first = await client.post("/webhooks/adcp", content=body, headers=_sign_and_send_body(body))
+        second = await client.post(
+            "/webhooks/adcp", content=body, headers=_sign_and_send_body(body)
+        )
 
     assert first.status_code == 200
     assert first.json()["duplicate"] is False

tests/conformance/signing/test_webhook_sender_alt_auth.py

@@ -98,6 +98,7 @@ async def test_bearer_token_attaches_authorization_header() -> None:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_bearer",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )
@@ -124,6 +125,7 @@ async def test_bearer_extra_headers_cannot_override_authorization() -> None:
             await sender.send_mcp(
                 url="http://test/webhooks/adcp",
                 task_id="t",
+                operation_id="op_test",
                 task_type="create_media_buy",
                 status="completed",
                 extra_headers={"Authorization": "Bearer attacker"},
@@ -148,6 +150,7 @@ async def test_adcp_legacy_hmac_round_trips_against_verifier() -> None:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_legacy_hmac",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )
@@ -184,6 +187,7 @@ async def test_adcp_legacy_hmac_resend_re_signs_with_fresh_timestamp() -> None:
         first = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_resend",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )
@@ -267,6 +271,7 @@ async def test_standard_webhooks_round_trips_against_verifier() -> None:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_sw",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )
@@ -300,6 +305,7 @@ async def test_standard_webhooks_resend_uses_fresh_msg_id_and_timestamp() -> Non
         first = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_sw_resend",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )
@@ -331,6 +337,7 @@ async def test_standard_webhooks_svix_verifier_interop() -> None:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_sw_svix",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )

tests/conformance/signing/test_webhook_sender_e2e.py

@@ -14,8 +14,10 @@
 
 from __future__ import annotations
 
+import contextlib
 import copy
 import json
+from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import Any
 
@@ -109,6 +111,7 @@ async def test_sender_to_receiver_happy_path() -> None:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_sender_e2e",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
             result={"media_buy_id": "mb_1"},
@@ -126,21 +129,32 @@ async def test_sender_to_receiver_happy_path() -> None:
 async def test_retry_with_same_idempotency_key_dedupes() -> None:
     """The whole reason WebhookDeliveryResult exposes idempotency_key: so the
     retry call reuses it and the receiver dedupes correctly. Without this,
-    every retry processes as a fresh event."""
+    every retry processes as a fresh event.
+
+    The key binds the payload, so the retry repeats the first call's
+    ``timestamp``: a fresh one makes a different body under the same key,
+    which the receiver refuses with 409."""
+    from datetime import datetime, timezone
+
+    sent_at = datetime(2026, 10, 18, 12, 0, tzinfo=timezone.utc)
     app, _ = _build_receiver_app()
     async with _build_sender(app) as sender:
         first = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_retry",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
+            timestamp=sent_at,
         )
         # Simulate caller deciding to retry (e.g., if first had been non-2xx).
         retry = await sender.send_mcp(
             url=first.url,
             task_id="task_retry",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
+            timestamp=sent_at,
             idempotency_key=first.idempotency_key,
         )
 
@@ -163,6 +177,7 @@ async def test_extra_headers_cannot_override_signature_bindings() -> None:
             await sender.send_mcp(
                 url="http://test/webhooks/adcp",
                 task_id="t",
+                operation_id="op_test",
                 task_type="create_media_buy",
                 status="completed",
                 extra_headers={"Signature": "attacker-controlled"},
@@ -171,6 +186,7 @@ async def test_extra_headers_cannot_override_signature_bindings() -> None:
             await sender.send_mcp(
                 url="http://test/webhooks/adcp",
                 task_id="t",
+                operation_id="op_test",
                 task_type="create_media_buy",
                 status="completed",
                 extra_headers={"Content-Digest": "forged"},
@@ -318,6 +334,7 @@ async def capture(request: Request) -> JSONResponse:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp/capture",
             task_id="task_byte_identity",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
             result={"media_buy_id": "mb_bytes"},
@@ -338,6 +355,7 @@ async def test_resend_replays_identical_bytes_with_fresh_signature() -> None:
         first = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_resend",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
             result={"media_buy_id": "mb_resend"},
@@ -374,6 +392,7 @@ async def test_extra_header_case_insensitive_rejection() -> None:
                 await sender.send_mcp(
                     url="http://test/webhooks/adcp",
                     task_id="t",
+                    operation_id="op_test",
                     task_type="create_media_buy",
                     status="completed",
                     extra_headers={bad: "attacker"},
@@ -397,6 +416,7 @@ async def test_receiver_rejects_when_sender_key_wrong() -> None:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="t",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )
@@ -419,6 +439,7 @@ async def test_owned_client_rejects_loopback_destination() -> None:
             await sender.send_mcp(
                 url="https://127.0.0.1/webhooks/adcp",
                 task_id="task_ssrf",
+                operation_id="op_test",
                 task_type="create_media_buy",
                 status="completed",
             )
@@ -451,6 +472,7 @@ async def test_owned_client_rejects_disallowed_port_when_hardening_configured()
                 await sender.send_mcp(
                     url="https://example.com:6379/webhooks/adcp",
                     task_id="task_port",
+                    operation_id="op_test",
                     task_type="create_media_buy",
                     status="completed",
                 )
@@ -538,6 +560,7 @@ def fake_build(uri: str, **_kwargs: Any) -> Any:
             result = await sender.send_mcp(
                 url="http://test:9443/webhooks/adcp",
                 task_id="task_nonstd",
+                operation_id="op_test",
                 task_type="create_media_buy",
                 status="completed",
             )
@@ -575,6 +598,7 @@ async def echo(_request: Request) -> JSONResponse:
         result = await sender.send_mcp(
             url="http://test/webhooks/adcp",
             task_id="task_op_trust",
+            operation_id="op_test",
             task_type="create_media_buy",
             status="completed",
         )
@@ -616,7 +640,9 @@ def __init__(self, **kwargs: Any) -> None:
             # delivery path. The last writer wins; the per-request call
             # is the one we care about.
             captured_kwargs.update(kwargs)
-            self._response = MagicMock(status_code=200, headers={}, content=b"{}")
+            self._response = MagicMock(
+                status_code=200, headers={}, content=b"{}", is_stream_consumed=True
+            )
 
         async def __aenter__(self) -> _FakeAsyncClient:
             return self
@@ -627,8 +653,10 @@ async def __aexit__(self, *_args: Any) -> None:
         async def aclose(self) -> None:
             return None
 
-        async def post(self, *_args: Any, **_kwargs: Any) -> Any:
-            return self._response
+        # The sender posts with ``client.stream`` to bound the response body.
+        @contextlib.asynccontextmanager
+        async def stream(self, *_args: Any, **_kwargs: Any) -> AsyncIterator[Any]:
+            yield self._response
 
     # Set HTTPS_PROXY pointing at an unreachable address. With
     # trust_env=True (the httpx default), this would override the
@@ -643,6 +671,7 @@ async def post(self, *_args: Any, **_kwargs: Any) -> Any:
                     await sender.send_mcp(
                         url="https://buyer.example.com/webhooks/adcp",
                         task_id="task_proxy",
+                        operation_id="op_test",
                         task_type="create_media_buy",
                         status="completed",
                     )
@@ -698,6 +727,7 @@ def reserved_headers(self) -> frozenset[str]:
             await sender.send_mcp(
                 url="https://127.0.0.1/webhooks/adcp",
                 task_id="task_no_sign",
+                operation_id="op_test",
                 task_type="create_media_buy",
                 status="completed",
             )