fix(identity): never send a bound origin to the 3.2 profile, and treat a missing kid as invalid

docs/agent-identity-3-3-migration.md

@@ -249,12 +249,13 @@ jws_artifacts = WbaJwsArtifactVerifier(
     signer_keys=directory_keys,     # DiscoveredSignerKeys: O's key directory
     revocation=origin_revocations,  # OriginRevocationChecker: O's list
     archive=HttpGovernanceKeyArchive(),  # history for governance contexts only
+    bound_origins=is_bound,         # R5 bindings: no downgrade to 3.2
 )
 ```
 
 - **O.** The canonical origin of `iss` for a governance context, `agent_url` for designated-task response signing, and `rights_agent.url` for a rights attestation. Pass the URL from your own context (the governance agent you expect, the agent you called, the rights agent named in the constraint you hold), never one read from the unverified token.
-- **Selection.** `select_jws_artifact_profile` (or `jws_artifacts.select`) derives the trust record from O. A record that lists O, with any role, selects the key directory. A record that doesn't list O, an operator that has never published one, or an O with no registrable domain selects the 3.2 profile. A record that can't be read (an outage, an invalid record, a 404/410 after a 200, local capacity) raises `JwsArtifactError` `jws_key_unavailable`: a fetch failure never selects 3.2.
-- **Key rules.** `alg` is `EdDSA` and `kid` is the key's JWK thumbprint, found in O's directory. No `use`, `key_ops` or `adcp_use` check applies: purpose separation is by origin. A key that another origin publishes is `jws_key_unknown`; a directory that can't be read is `jws_key_unavailable`.
+- **Selection.** `select_jws_artifact_profile` (or `jws_artifacts.select`) derives the trust record from O. A record that lists O, with any role, selects the key directory. A record that doesn't list O, an operator that has never published one, or an O with no registrable domain selects the 3.2 profile. A record that can't be read (an outage, an invalid record, a 404/410 after a 200, local capacity) raises `JwsArtifactError` `jws_key_unavailable`: a fetch failure never selects 3.2. With `bound_origins`, an O the verifier has bound (R5) is never sent to 3.2: it raises `jws_invalid`, and a lookup that raises is `jws_key_unavailable` (R4, no downgrade). The lookup takes the origin and returns whether a principal is bound to it, sync or async.
+- **Key rules.** `alg` is `EdDSA` and `kid` is the key's JWK thumbprint, found in O's directory. A missing `kid` is `jws_invalid` (`governance_token_invalid` on both governance paths). No `use`, `key_ops` or `adcp_use` check applies: purpose separation is by origin. A key that another origin publishes is `jws_key_unknown`; a directory that can't be read is `jws_key_unavailable`.
 - **Revocation.** O's list. A revoked key is `jws_key_revoked`. A stale list, one that can't be checked, or (unless `revocation_absent="accept"`) an origin that serves none is `jws_revocation_unavailable`.
 - **Governance contexts.** `verify_governance_authorization(..., web_bot_auth=jws_artifacts)` selects the profile from `expected_issuer`. On the key-directory path `jwks` and `revocation_checker` are not used, and every other check is unchanged: keep the brand.json cross-check of `iss` (`resolve_governance_jwks`, or `resolve_issuer_jwks` in `make_governance_enforcement_middleware`), `is_jti_revoked` and `revocation_resolver`. Outages are `governance_jwks_unavailable` (new in the error-code literal), revocation problems `governance_token_revoked`, and an origin that serves no list is refused because a governance context authorizes spend. Unlisted issuers stay on the 3.2 path.
 - **History.** `verify_historical_governance_context(token=, expected_issuer=, web_bot_auth=)` verifies a retained governance context for audit: a key rotated out of the directory is taken from the governance key archive at O (`/.well-known/jwks-archive.json`, matched by thumbprint; `key_source="archive"`). It checks the header, issuer, key, signature and O's revocation list (an origin with no list has revoked nothing), not expiry, audience or replay. An unlisted issuer returns a failure with `profile="3.2"`.

src/adcp/governance.py

@@ -1181,7 +1181,7 @@ def reject(
     token = cast(str, token)
     kid = header.get("kid")
     if not isinstance(kid, str) or not kid:
-        return reject("governance_key_unknown", "governance signing kid is missing")
+        return reject("governance_token_invalid", "governance signing kid is missing")
     # AdCP 3.3 R4: the trust record derived from the expected issuer's origin
     # selects the profile. It is never read from the unverified token, and a
     # record that cannot be read never falls back to the 3.2 profile.

src/adcp/signing/jws_artifacts.py

@@ -42,9 +42,10 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import time
 from collections import OrderedDict
-from collections.abc import Callable, Mapping
+from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
 from typing import Any, Literal, Protocol
 from urllib.parse import urlsplit
@@ -213,8 +214,16 @@ def _lists_origin(record: EvaluatedTrustRecord, origin: str) -> bool:
     return False
 
 
+#: Whether the verifier has bound ``origin`` to a principal (R5). Raises
+#: when the binding store cannot be read.
+OriginBindingLookup = Callable[[str], bool | Awaitable[bool]]
+
+
 async def select_jws_artifact_profile(
-    signing_agent_url: str, records: TrustRecordEvaluator
+    signing_agent_url: str,
+    records: TrustRecordEvaluator,
+    *,
+    bound_origins: OriginBindingLookup | None = None,
 ) -> JwsArtifactProfile:
     """Apply R4's selection rule to the signing agent URL the caller trusts.
 
@@ -225,24 +234,47 @@ async def select_jws_artifact_profile(
     cannot be read (an outage, an invalid record, a 404/410 after the
     operator served one, or local capacity) raises :class:`JwsArtifactError`
     ``jws_key_unavailable``: a fetch failure never selects the 3.2 profile.
+
+    ``bound_origins`` enforces R4's no downgrade: an ``O`` the verifier has
+    bound (R5) is never sent to the 3.2 profile, and raises ``jws_invalid``
+    instead. A lookup that raises is ``jws_key_unavailable``.
     """
     origin = _signing_origin(signing_agent_url)
+
+    async def legacy(
+        reason: Literal["unlisted", "no_trust_record"], record: EvaluatedTrustRecord | None = None
+    ) -> JwsArtifactProfile:
+        if bound_origins is not None:
+            try:
+                bound = bound_origins(origin)
+                if inspect.isawaitable(bound):
+                    bound = await bound
+            except Exception as exc:
+                raise JwsArtifactError(
+                    "jws_key_unavailable", "origin binding lookup failed"
+                ) from exc
+            if bound:
+                raise JwsArtifactError(
+                    "jws_invalid", "a bound origin is never verified under the 3.2 profile"
+                )
+        return JwsArtifactProfile("3.2", origin, reason, record)
+
     try:
         has_operator = registrable_domain(origin) is not None
     except ValueError:
         has_operator = False
     if not has_operator:
-        return JwsArtifactProfile("3.2", origin, "no_trust_record")
+        return await legacy("no_trust_record")
     try:
         record = await records.evaluate(origin)
     except TrustRecordUnavailableError as exc:
         raise JwsArtifactError("jws_key_unavailable", "issuer trust record is unavailable") from exc
     except TrustRecordError:
         # Absent (404/410 before any 200) or no registrable domain.
-        return JwsArtifactProfile("3.2", origin, "no_trust_record")
+        return await legacy("no_trust_record")
     if _lists_origin(record, origin):
         return JwsArtifactProfile("web-bot-auth", origin, "listed", record)
-    return JwsArtifactProfile("3.2", origin, "unlisted", record)
+    return await legacy("unlisted", record)
 
 
 def _default_registry() -> JWSRegistry:
@@ -341,7 +373,7 @@ async def verify_wba_jws_artifact(
         raise JwsArtifactError("jws_invalid", "the b64 header is not supported")
     kid = header.get("kid")
     if not isinstance(kid, str) or not kid or len(kid) > 256:
-        raise JwsArtifactError("jws_key_unknown", "kid is missing")
+        raise JwsArtifactError("jws_invalid", "kid is missing")
 
     key: Mapping[str, Any] | None = None
     key_source: Literal["directory", "archive"] = "directory"
@@ -427,16 +459,22 @@ class WbaJwsArtifactVerifier:
     * ``archive``: the governance key archive, for historical verification of
       a governance context (:class:`HttpGovernanceKeyArchive`). ``None``
       disables history.
+    * ``bound_origins``: whether an origin is bound to a principal (R5), for
+      R4's no downgrade. Usually the binding store the request path's
+      principal binding writes to.
     """
 
     records: TrustRecordEvaluator
     signer_keys: WbaSignerKeyResolver
     revocation: Any
     archive: GovernanceKeyArchive | None = None
+    bound_origins: OriginBindingLookup | None = None
 
     async def select(self, signing_agent_url: str) -> JwsArtifactProfile:
-        """:func:`select_jws_artifact_profile` with this verifier's trust records."""
-        return await select_jws_artifact_profile(signing_agent_url, self.records)
+        """:func:`select_jws_artifact_profile` with this verifier's records and bindings."""
+        return await select_jws_artifact_profile(
+            signing_agent_url, self.records, bound_origins=self.bound_origins
+        )
 
     async def verify(
         self,

tests/test_jws_artifacts_wba.py

@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import base64
 import json
 from collections.abc import Mapping
 from typing import Any
@@ -243,6 +244,67 @@ async def test_a_signing_agent_url_must_be_an_https_origin(url):
     assert caught.value.code == "jws_invalid"
 
 
+def _bound(*origins: str, calls: list[str] | None = None):
+    async def lookup(origin: str) -> bool:
+        if calls is not None:
+            calls.append(origin)
+        return origin in origins
+
+    return lookup
+
+
+_UNLISTED = {trust_record_url(GOV): {"agents": [{"url": BRAND_AGENT, "roles": ["adcp:buying"]}]}}
+
+
+@pytest.mark.parametrize(
+    ("url", "documents"),
+    [(GOV_ISSUER, _UNLISTED), (GOV_ISSUER, {}), ("https://203.0.113.7/gov", {})],
+    ids=["unlisted", "never-published", "no-registrable-domain"],
+)
+async def test_a_bound_origin_is_never_sent_to_3_2(url, documents):
+    """R4, no downgrade: once O is bound (R5), the 3.2 profile is refused for O."""
+    origin = url.rsplit("/", 1)[0]
+    with pytest.raises(JwsArtifactError) as caught:
+        await select_jws_artifact_profile(url, _records(documents), bound_origins=_bound(origin))
+    assert caught.value.code == "jws_invalid"
+
+
+async def test_an_unbound_origin_still_selects_3_2():
+    calls: list[str] = []
+    profile = await select_jws_artifact_profile(
+        GOV_ISSUER, _records(_UNLISTED), bound_origins=_bound(BRAND_AGENT, calls=calls)
+    )
+    assert (profile.profile, profile.reason) == ("3.2", "unlisted")
+    assert calls == [GOV]
+
+
+async def test_a_listed_origin_does_not_consult_the_bindings():
+    calls: list[str] = []
+    profile = await select_jws_artifact_profile(
+        GOV_ISSUER, _records(_listing(GOV)), bound_origins=_bound(GOV, calls=calls)
+    )
+    assert profile.profile == "web-bot-auth"
+    assert calls == []
+
+
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_a_binding_lookup_may_be_sync_or_async(sync):
+    lookup = (lambda origin: origin == GOV) if sync else _bound(GOV)
+    with pytest.raises(JwsArtifactError) as caught:
+        await select_jws_artifact_profile(GOV_ISSUER, _records(_UNLISTED), bound_origins=lookup)
+    assert caught.value.code == "jws_invalid"
+
+
+async def test_a_binding_lookup_that_fails_is_unavailable():
+    def lookup(origin: str) -> bool:
+        raise RuntimeError("binding store down")
+
+    with pytest.raises(JwsArtifactError) as caught:
+        await select_jws_artifact_profile(GOV_ISSUER, _records(_UNLISTED), bound_origins=lookup)
+    assert caught.value.code == "jws_key_unavailable"
+    assert caught.value.transient
+
+
 # ---------------------------------------------------------------------------
 # Governance context, online
 # ---------------------------------------------------------------------------
@@ -296,6 +358,48 @@ async def test_an_unlisted_issuer_stays_on_the_3_2_profile():
     assert directory.calls == []
 
 
+async def test_a_bound_issuer_is_refused_rather_than_verified_on_3_2():
+    from adcp.signing import StaticJwksResolver
+
+    private = ed25519.Ed25519PrivateKey.generate()
+    jwk = {
+        **OKPKey.import_key(private).as_dict(private=False),
+        "kid": "gov-2026",
+        "use": "sig",
+        "key_ops": ["verify"],
+        "adcp_use": "governance-signing",
+    }
+    verifier = WbaJwsArtifactVerifier(
+        records=_records({}),
+        signer_keys=Directory(),
+        revocation=Revocation(),
+        archive=Archive(),
+        bound_origins=_bound(GOV),
+    )
+    token = _governance_token(private, "gov-2026")
+
+    online = await _verify_governance(token, verifier, StaticJwksResolver({"keys": [jwk]}))
+    historical = await verify_historical_governance_context(
+        token=token, expected_issuer=GOV_ISSUER, web_bot_auth=verifier
+    )
+
+    assert (online.ok, online.error) == (False, "governance_token_invalid")
+    assert (historical.ok, historical.error) == (False, "governance_token_invalid")
+    assert historical.profile != "3.2"
+
+
+@pytest.mark.parametrize("listed", [True, False], ids=["wba", "3.2"])
+async def test_a_governance_token_without_a_kid_is_invalid(listed):
+    private, jwk = _wba_key()
+    encoded_header, payload, signature = _governance_token(private, jwk["kid"]).split(".")
+    header = json.loads(base64.urlsafe_b64decode(encoded_header + "=" * (-len(encoded_header) % 4)))
+    del header["kid"]
+    token = ".".join((b64url_encode(json.dumps(header).encode()), payload, signature))
+    verifier = _verifier(_listing(GOV) if listed else {}, Directory({GOV: [jwk]}))
+    result = await _verify_governance(token, verifier, ForbiddenJwks())
+    assert (result.ok, result.error) == (False, "governance_token_invalid")
+
+
 async def test_the_3_2_profile_still_checks_key_purpose():
     from adcp.signing import StaticJwksResolver
 
@@ -780,8 +884,8 @@ async def test_malformed_tokens_are_invalid(token, code):
     [
         (b'{"alg":"EdDSA","alg":"EdDSA","typ":"test+jws"}', "jws_invalid"),
         (b'["EdDSA"]', "jws_invalid"),
-        ({"alg": "EdDSA", "typ": "test+jws"}, "jws_key_unknown"),
-        ({"alg": "EdDSA", "typ": "test+jws", "kid": 7}, "jws_key_unknown"),
+        ({"alg": "EdDSA", "typ": "test+jws"}, "jws_invalid"),
+        ({"alg": "EdDSA", "typ": "test+jws", "kid": 7}, "jws_invalid"),
     ],
     ids=["duplicate-member", "not-an-object", "no-kid", "non-string-kid"],
 )