fix(signing): cap key_origins entry size, route cold-cache jwks_uri=None to JWKS_UNAVAILABLE, host-only diagnostic fallback

Three remaining Argus follow-ups from #789.

1. Per-entry length cap on _extract_key_origins. Each origin value
   is bounded at 512 bytes — well above any legitimate
   scheme+host+port shape but tight enough that a pathological
   multi-kilobyte entry from the 64 KiB capabilities body doesn't
   propagate through downstream comparisons. Oversized entries are
   SKIPPED (not truncated — a truncated host would silently match
   the wrong domain). Constant
   _MAX_KEY_ORIGIN_VALUE_BYTES = 512 documents the choice.

2. Cold-cache jwks_uri=None routes to REQUEST_SIGNATURE_JWKS_UNAVAILABLE.
   Previous behavior coerced None to '' and routed through the
   mismatch path with empty actual_origin — a resolver-side I/O
   failure misclassified as adversarial origin-mismatch on
   dashboards. Now raises JWKS_UNAVAILABLE with
   detail={'purpose': signing_purpose} so the cold-cache shape
   aggregates with other resolver-fetch failures.

3. Diagnostic host-only fallback on canonicalization failure. When
   _origin_host can't canonicalize one side, the mismatch detail's
   expected_origin / actual_origin values must still be HOST-SHAPED
   — previous fallback leaked the full raw URL into the host-labeled
   field, inconsistent with the success-path host-only shape. New
   _diagnostic_host helper falls through to _extract_host (the same
   URL/bare-host parser the canonicalization step uses) for a
   best-effort host, empty string at worst.

Three new tests in test_verify_from_agent_url.py:
- _extract_key_origins skips oversized entries
- mismatch detail uses host-only fallback (no URL leakage)
- jwks_uri=None routes to JWKS_UNAVAILABLE

622 tests across impacted surface remain green. ruff + mypy clean.

Refs Argus second-pass review on #789 (cold-cache routing, detail
shape, per-entry length cap).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bokelley

src/adcp/signing/agent_resolver.py

@@ -329,6 +329,14 @@ def _extract_brand_json_url(capabilities: dict[str, Any]) -> str:
     return brand_json_url
 
 
+#: Per-entry size clamp on ``identity.key_origins`` values. DNS hostname
+#: limit is 253 octets (RFC 1035); origin strings carry scheme+host so
+#: the practical cap is a bit higher, but 512 is well above any
+#: legitimate value while still bounding the surface against a
+#: pathologically-large entry from a 64 KiB capabilities body.
+_MAX_KEY_ORIGIN_VALUE_BYTES = 512
+
+
 def _extract_key_origins(capabilities: dict[str, Any]) -> dict[str, str] | None:
     """Pluck ``identity.key_origins`` from the capabilities body.
 
@@ -339,6 +347,14 @@ def _extract_key_origins(capabilities: dict[str, Any]) -> dict[str, str] | None:
     purpose is actually exercised). Filters values to strings — a
     malformed entry is skipped rather than poisoning the whole map.
 
+    **Per-entry length cap (``_MAX_KEY_ORIGIN_VALUE_BYTES``).** Each
+    origin value is bounded to 512 bytes — well above any legitimate
+    ``scheme + host + port`` shape but tight enough that a pathological
+    multi-kilobyte value from the 64 KiB capabilities body doesn't
+    propagate through downstream comparisons. Entries exceeding the cap
+    are skipped (the verifier then surfaces the purpose as missing on
+    the consistency check).
+
     Forward-compat with operators on 3.0 schemas: the map travels under
     ``additionalProperties: true`` and the SDK reads it as a plain dict
     rather than via the typed Pydantic surface (which won't carry the
@@ -352,8 +368,13 @@ def _extract_key_origins(capabilities: dict[str, Any]) -> dict[str, str] | None:
         return None
     out: dict[str, str] = {}
     for purpose, origin in raw.items():
-        if isinstance(purpose, str) and isinstance(origin, str) and origin:
-            out[purpose] = origin
+        if not (isinstance(purpose, str) and isinstance(origin, str) and origin):
+            continue
+        if len(origin.encode("utf-8")) > _MAX_KEY_ORIGIN_VALUE_BYTES:
+            # Length-capped entry — skip rather than truncate (a
+            # truncated host would silently match the wrong domain).
+            continue
+        out[purpose] = origin
     return out or None
 
 

src/adcp/signing/key_origins.py

@@ -151,15 +151,34 @@ def check_key_origin_consistency(
             detail={
                 "purpose": purpose,
                 # Use the canonicalized values when available; fall back
-                # to the raw inputs for diagnostic accuracy when one
-                # side failed to canonicalize. Spec wording is
+                # to a best-effort host extraction (NOT the raw URL — the
+                # field name promises a host, and surfacing a full URL on
+                # canonicalization failure was inconsistent with the
+                # success path's host-only shape). Spec wording is
                 # ``expected_origin`` / ``actual_origin`` verbatim.
-                "expected_origin": declared_host if declared_host is not None else declared,
-                "actual_origin": actual_host if actual_host is not None else jwks_uri,
+                "expected_origin": _diagnostic_host(declared_host, declared),
+                "actual_origin": _diagnostic_host(actual_host, jwks_uri),
             },
         )
 
 
+def _diagnostic_host(canonical: str | None, raw: str) -> str:
+    """Return ``canonical`` if present, else a best-effort host from
+    ``raw``, else the empty string.
+
+    Used to keep ``expected_origin`` / ``actual_origin`` host-shaped
+    in the mismatch detail payload even when canonicalization failed.
+    Falls through to ``_extract_host`` (the same URL/bare-host parser
+    the canonicalization step uses) for the best-effort path, so the
+    diagnostic value still reflects "the host the operator/verifier
+    pointed at" rather than the full URL surface.
+    """
+    if canonical is not None:
+        return canonical
+    host = _extract_host(raw)
+    return host or ""
+
+
 def _origin_host(value: str) -> str | None:
     """Return the host portion of a URL or bare origin, canonicalized
     for byte-equality comparison.

src/adcp/signing/verifier.py

@@ -43,6 +43,7 @@
     REQUEST_SIGNATURE_DIGEST_MISMATCH,
     REQUEST_SIGNATURE_HEADER_MALFORMED,
     REQUEST_SIGNATURE_INVALID,
+    REQUEST_SIGNATURE_JWKS_UNAVAILABLE,
     REQUEST_SIGNATURE_KEY_PURPOSE_INVALID,
     REQUEST_SIGNATURE_KEY_REVOKED,
     REQUEST_SIGNATURE_KEY_UNKNOWN,
@@ -576,12 +577,27 @@ def _maybe_check_key_origin(
             )
         return
     jwks_uri = getattr(resolver, "jwks_uri", None)
-    # ``jwks_uri`` may be ``None`` if the brand-json resolver hasn't
-    # populated it yet (cold cache + failed refresh). The consistency
-    # check fails closed on a missing actual host — same posture as
-    # ``_origin_host`` returning ``None``.
+    if not jwks_uri:
+        # A brand-json resolver that hasn't populated ``jwks_uri`` (cold
+        # cache + failed refresh, or a misconfigured custom resolver) is
+        # a resolver-side I/O failure, not a key-origin mismatch — the
+        # verifier has no resolved host to compare. Surface as
+        # ``REQUEST_SIGNATURE_JWKS_UNAVAILABLE`` so dashboards aggregate
+        # this cold-cache shape with other resolver-fetch failures
+        # rather than with adversarial origin-mismatch traffic.
+        raise SignatureVerificationError(
+            REQUEST_SIGNATURE_JWKS_UNAVAILABLE,
+            step=7,
+            message=(
+                "brand-json resolver did not populate jwks_uri (cold cache "
+                "or misconfigured resolver); key_origins consistency check "
+                "cannot proceed without a resolved host to compare against "
+                f"identity.key_origins.{signing_purpose}"
+            ),
+            detail={"purpose": signing_purpose},
+        )
     check_key_origin_consistency(
-        jwks_uri=jwks_uri or "",
+        jwks_uri=jwks_uri,
         key_origins=expected_key_origins,
         purpose=signing_purpose,
         posture=posture,

tests/test_verify_from_agent_url.py

@@ -620,3 +620,92 @@ def test_no_warning_when_both_attributes_align() -> None:
         and "jwks_source" in str(w.message)
     ]
     assert misconfig == []
+
+
+# ---- _extract_key_origins length cap (Argus follow-up nit #1) ----
+
+
+def test_extract_key_origins_caps_oversized_entries() -> None:
+    """Each origin value must be clamped at 512 bytes — well above any
+    legitimate ``scheme+host+port`` shape but tight enough that a
+    pathological multi-kilobyte value from the 64 KiB capabilities body
+    doesn't propagate through downstream comparisons. Oversized entries
+    are SKIPPED (not truncated — a truncated host would silently match
+    the wrong domain)."""
+    from adcp.signing.agent_resolver import _extract_key_origins
+
+    huge = "https://" + "x" * 1024 + ".com"
+    legit = "https://keys.brand.com"
+    result = _extract_key_origins(
+        {
+            "identity": {
+                "key_origins": {
+                    "request_signing": legit,
+                    "webhook_signing": huge,  # skipped
+                }
+            }
+        }
+    )
+    assert result == {"request_signing": legit}
+
+
+# ---- Diagnostic host-only fallback (Argus follow-up nit #2) ----
+
+
+def test_mismatch_detail_uses_host_only_fallback_on_canonicalization_failure() -> None:
+    """When ``_origin_host`` can't canonicalize one side (e.g. spaces
+    in the host), the mismatch ``expected_origin`` / ``actual_origin``
+    detail values must still be HOST-SHAPED — not the full raw URL.
+    Previous behavior leaked the full URL into the host-labeled field,
+    inconsistent with the success path. Now the diagnostic uses a
+    best-effort host extraction via ``_extract_host``."""
+    from adcp.signing.errors import REQUEST_SIGNATURE_KEY_ORIGIN_MISMATCH
+    from adcp.signing.key_origins import check_key_origin_consistency
+
+    with pytest.raises(SignatureVerificationError) as exc_info:
+        check_key_origin_consistency(
+            jwks_uri="https://keys.brand.example/jwks.json",
+            key_origins={"request_signing": "not a host with spaces"},
+            purpose="request_signing",
+        )
+    assert exc_info.value.code == REQUEST_SIGNATURE_KEY_ORIGIN_MISMATCH
+    detail = exc_info.value.detail
+    assert detail is not None
+    # actual_origin canonicalizes cleanly to the host (no URL form).
+    assert detail["actual_origin"] == "keys.brand.example"
+    # expected_origin failed to canonicalize but still doesn't leak the
+    # full raw string with quoting artifacts — empty string at worst,
+    # never the full URL.
+    assert "/" not in detail["expected_origin"]
+
+
+# ---- jwks_uri=None routes to JWKS_UNAVAILABLE (Argus follow-up #4) ----
+
+
+def test_maybe_check_key_origin_jwks_uri_none_routes_to_jwks_unavailable() -> None:
+    """A brand-json resolver that hasn't populated ``jwks_uri`` (cold
+    cache + failed refresh, or a misconfigured custom resolver) is a
+    resolver-side I/O failure, not an origin mismatch. The verifier
+    must surface ``REQUEST_SIGNATURE_JWKS_UNAVAILABLE`` so dashboards
+    aggregate this cold-cache shape with other resolver-fetch
+    failures rather than with adversarial origin-mismatch traffic."""
+    from adcp.signing.errors import REQUEST_SIGNATURE_JWKS_UNAVAILABLE
+    from adcp.signing.verifier import _maybe_check_key_origin
+
+    class _BrandJsonResolverWithNoJwksUri:
+        jwks_source = "brand_json"
+        # ``jwks_uri`` deliberately absent / None — cold cache shape.
+        jwks_uri = None
+
+        def __call__(self, keyid: str) -> dict | None:  # type: ignore[type-arg]
+            return None
+
+    with pytest.raises(SignatureVerificationError) as exc_info:
+        _maybe_check_key_origin(
+            resolver=_BrandJsonResolverWithNoJwksUri(),  # type: ignore[arg-type]
+            expected_key_origins={"request_signing": "https://keys.brand.example"},
+            signing_purpose="request_signing",
+            posture=None,
+        )
+    assert exc_info.value.code == REQUEST_SIGNATURE_JWKS_UNAVAILABLE
+    assert exc_info.value.detail == {"purpose": "request_signing"}