hardening(identity): typed-field fallback + per-element extras guard + set/frozenset reject

vvillait88 · claude · vvillait88 · commit 48bced1be8e5 · 2026-05-09T05:27:48.000-07:00
Three reviewer LOW findings on the UCP profile path:

* `build_ucp_profile` now falls back to the typed `AssessResult.operator_verification`
  (and ad-hoc `account_verification`) field when `data.raw` doesn't carry the block.
  Production callers populate `raw`, but a hand-constructed `AssessResult(raw=None)`
  was silently dropping the operator verification block, diverging from the node
  sibling's typed-field read path.
* `UCPService`, `UCPCapability`, and `UCPSigningKey` now mirror `UCPProfile.to_dict`'s
  reserved-key collision guard so vendor `extras` can't silently overwrite a
  canonical declared field via `out.update(extras)`. `UCPPaymentHandler` has no
  `extras` attribute and is unaffected.
* `_reject_unsafe_numbers` now rejects `set` / `frozenset` outright with a typed
  `ValueError` (mirroring node's `stableStringify: Set values are not allowed`).
  Previously an empty set or a set-of-valid-strings fell through cleanly and
  surfaced a raw `TypeError` from `json.dumps` later.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/agentscore_commerce/identity/ucp.py b/agentscore_commerce/identity/ucp.py
@@ -50,6 +50,8 @@ class UCPSigningKey:
     extras: dict[str, Any] = field(default_factory=dict)
     """Additional JWK fields (x, y, n, e, etc.) merged into the serialized output."""
 
+    _RESERVED = frozenset({"kid", "kty", "alg", "use", "crv"})
+
     def to_dict(self) -> dict[str, Any]:
         out: dict[str, Any] = {"kid": self.kid, "kty": self.kty}
         if self.alg is not None:
@@ -58,7 +60,11 @@ def to_dict(self) -> dict[str, Any]:
             out["use"] = self.use
         if self.crv is not None:
             out["crv"] = self.crv
-        out.update(self.extras)
+        for k, v in self.extras.items():
+            if k in self._RESERVED:
+                msg = f"UCPSigningKey.extras key {k!r} collides with a reserved field; rejected."
+                raise ValueError(msg)
+            out[k] = v
         return out
 
     @classmethod
@@ -110,13 +116,19 @@ class UCPService:
     version: str | None = None
     extras: dict[str, Any] = field(default_factory=dict)
 
+    _RESERVED = frozenset({"type", "url", "version"})
+
     def to_dict(self) -> dict[str, Any]:
         out: dict[str, Any] = {"type": self.type}
         if self.url is not None:
             out["url"] = self.url
         if self.version is not None:
             out["version"] = self.version
-        out.update(self.extras)
+        for k, v in self.extras.items():
+            if k in self._RESERVED:
+                msg = f"UCPService.extras key {k!r} collides with a reserved field; rejected."
+                raise ValueError(msg)
+            out[k] = v
         return out
 
 
@@ -129,13 +141,19 @@ class UCPCapability:
     version: str | None = None
     extras: dict[str, Any] = field(default_factory=dict)
 
+    _RESERVED = frozenset({"name", "schema", "version"})
+
     def to_dict(self) -> dict[str, Any]:
         out: dict[str, Any] = {"name": self.name}
         if self.schema is not None:
             out["schema"] = self.schema
         if self.version is not None:
             out["version"] = self.version
-        out.update(self.extras)
+        for k, v in self.extras.items():
+            if k in self._RESERVED:
+                msg = f"UCPCapability.extras key {k!r} collides with a reserved field; rejected."
+                raise ValueError(msg)
+            out[k] = v
         return out
 
 
@@ -256,7 +274,24 @@ async def ucp_profile():
     if data is not None and data.resolved_operator:
         raw = data.raw or {}
         operator_verification = raw.get("operator_verification") if isinstance(raw, dict) else None
+        if not operator_verification:
+            # Fallback to the typed AssessResult.operator_verification field when
+            # `raw` doesn't carry it. Mirrors the node sibling's typed-field read
+            # path so a hand-constructed AssessResult (no `raw`) still surfaces
+            # the operator verification block in the UCP capability claims.
+            typed_op = getattr(data, "operator_verification", None)
+            if typed_op is not None and not isinstance(typed_op, dict):
+                # Convert OperatorVerification dataclass to a plain dict.
+                operator_verification = {
+                    "level": getattr(typed_op, "level", None),
+                    "operator_type": getattr(typed_op, "operator_type", None),
+                    "verified_at": getattr(typed_op, "verified_at", None),
+                }
+            else:
+                operator_verification = typed_op
         account_verification = raw.get("account_verification") if isinstance(raw, dict) else None
+        if not account_verification:
+            account_verification = getattr(data, "account_verification", None)
         if not isinstance(operator_verification, dict):
             operator_verification = {}
         if not isinstance(account_verification, dict):
diff --git a/agentscore_commerce/identity/ucp_jwks.py b/agentscore_commerce/identity/ucp_jwks.py
@@ -211,11 +211,22 @@ def _reject_unsafe_numbers(value: Any) -> None:
             )
             raise ValueError(msg)
         return
+    # Reject set / frozenset with a typed message (mirrors the node sibling's
+    # "Set values are not allowed" rejection in stableStringify). Without this,
+    # an empty set or a set-of-valid-strings falls through `_reject_unsafe_numbers`
+    # cleanly and surfaces a raw `TypeError` from `json.dumps` later. Sets aren't
+    # representable in JSON; convert to a sorted list before passing.
+    if isinstance(value, set | frozenset):
+        msg = (
+            f"{type(value).__name__} values are not allowed in canonicalized JSON. "
+            "Convert to a sorted list before passing."
+        )
+        raise ValueError(msg)
     if isinstance(value, dict):
         for k, v in value.items():
             _reject_unsafe_numbers(k)
             _reject_unsafe_numbers(v)
-    elif isinstance(value, list | tuple | set | frozenset):
+    elif isinstance(value, list | tuple):
         for v in value:
             _reject_unsafe_numbers(v)
 
diff --git a/tests/test_ucp.py b/tests/test_ucp.py
@@ -5,6 +5,7 @@
 from agentscore_commerce.identity import (
     AGENTSCORE_UCP_CAPABILITY,
     AssessResult,
+    OperatorVerification,
     UCPCapability,
     UCPPaymentHandler,
     UCPService,
@@ -179,3 +180,133 @@ def test_coerces_null_verified_at_to_none() -> None:
 
 def test_coerces_empty_string_verified_at_to_none() -> None:
     assert _claims_of({"verified_at": ""})["verified_at"] is None
+
+
+# Typed-field fallback: production callers populate `data.raw`, but a
+# hand-constructed AssessResult (no raw) should still surface the operator
+# verification block via the typed `AssessResult.operator_verification` field
+# and the (optional) `account_verification` attribute. Mirrors the node sibling's
+# typed-field read path.
+
+
+def test_typed_operator_verification_fallback_when_raw_is_none() -> None:
+    result = AssessResult(
+        allow=True,
+        resolved_operator="op_typed",
+        operator_verification=OperatorVerification(
+            level="enhanced",
+            operator_type="api",
+            verified_at="2026-04-01T00:00:00Z",
+        ),
+        raw=None,
+    )
+    profile = build_ucp_profile(**_base_kwargs(), data=result)
+    d = profile.to_dict()
+    cap = next(c for c in d["capabilities"] if c["name"] == AGENTSCORE_UCP_CAPABILITY)
+    claims = cap["claims"]
+    assert claims["operator_id"] == "op_typed"
+    assert claims["kyc_level"] == "enhanced"
+    assert claims["verified_at"] == "2026-04-01T00:00:00Z"
+
+
+def test_typed_account_verification_fallback_via_setattr() -> None:
+    # `AssessResult` doesn't declare `account_verification` as a typed field, but
+    # a caller can still attach one ad-hoc. The fallback reads it via getattr so
+    # parity with the node sibling holds whichever way the caller populates it.
+    result = AssessResult(
+        allow=True,
+        resolved_operator="op_typed",
+        operator_verification=OperatorVerification(level="verified"),
+        raw=None,
+    )
+    result.account_verification = {  # type: ignore[attr-defined]
+        "kyc_level": "verified",
+        "age_bracket": "21+",
+        "jurisdiction": "US",
+        "sanctions_clear": True,
+    }
+    profile = build_ucp_profile(**_base_kwargs(), data=result)
+    d = profile.to_dict()
+    cap = next(c for c in d["capabilities"] if c["name"] == AGENTSCORE_UCP_CAPABILITY)
+    claims = cap["claims"]
+    assert claims["kyc_level"] == "verified"
+    assert claims["age_bracket"] == "21+"
+    assert claims["jurisdiction"] == "US"
+    assert claims["sanctions_clear"] is True
+
+
+def test_raw_takes_precedence_over_typed_fallback() -> None:
+    # When raw carries `operator_verification`, the typed-field fallback is NOT
+    # consulted. Production callers populate raw and the typed fields stay
+    # in sync; this test pins the precedence so a typed mismatch can't silently
+    # override the raw payload.
+    result = AssessResult(
+        allow=True,
+        resolved_operator="op_xyz",
+        operator_verification=OperatorVerification(level="enhanced"),
+        raw={
+            "operator_verification": {"level": "verified"},
+            "account_verification": {"kyc_level": "verified"},
+        },
+    )
+    profile = build_ucp_profile(**_base_kwargs(), data=result)
+    cap = next(c for c in profile.capabilities if c.name == AGENTSCORE_UCP_CAPABILITY)
+    # `kyc_level` reads from raw account_verification.kyc_level first.
+    assert cap.extras["claims"]["kyc_level"] == "verified"
+
+
+# Per-element to_dict reserved-key collision guard. Mirrors the parent
+# UCPProfile.to_dict guard so vendor extras can't silently overwrite a canonical
+# field on UCPService / UCPCapability / UCPSigningKey via `out.update(extras)`.
+
+
+def test_ucp_service_extras_collision_with_type_rejected() -> None:
+    svc = UCPService(type="rest", extras={"type": "different"})
+    with pytest.raises(ValueError, match=r"UCPService\.extras key 'type' collides"):
+        svc.to_dict()
+
+
+def test_ucp_service_extras_collision_with_url_rejected() -> None:
+    svc = UCPService(type="rest", url="https://x.example", extras={"url": "https://attacker.example"})
+    with pytest.raises(ValueError, match=r"UCPService\.extras key 'url' collides"):
+        svc.to_dict()
+
+
+def test_ucp_service_extras_non_reserved_pass_through() -> None:
+    svc = UCPService(type="rest", url="https://x.example", extras={"region": "us-west-1"})
+    assert svc.to_dict() == {"type": "rest", "url": "https://x.example", "region": "us-west-1"}
+
+
+def test_ucp_capability_extras_collision_with_name_rejected() -> None:
+    cap = UCPCapability(name="checkout", extras={"name": "different"})
+    with pytest.raises(ValueError, match=r"UCPCapability\.extras key 'name' collides"):
+        cap.to_dict()
+
+
+def test_ucp_capability_extras_collision_with_schema_rejected() -> None:
+    cap = UCPCapability(name="checkout", schema="https://x/y", extras={"schema": "https://attacker"})
+    with pytest.raises(ValueError, match=r"UCPCapability\.extras key 'schema' collides"):
+        cap.to_dict()
+
+
+def test_ucp_capability_extras_non_reserved_pass_through() -> None:
+    cap = UCPCapability(name="checkout", extras={"claims": {"k": "v"}})
+    assert cap.to_dict() == {"name": "checkout", "claims": {"k": "v"}}
+
+
+def test_ucp_signing_key_extras_collision_with_kid_rejected() -> None:
+    sk = UCPSigningKey(kid="me", kty="EC", extras={"kid": "attacker"})
+    with pytest.raises(ValueError, match=r"UCPSigningKey\.extras key 'kid' collides"):
+        sk.to_dict()
+
+
+def test_ucp_signing_key_extras_collision_with_kty_rejected() -> None:
+    sk = UCPSigningKey(kid="me", kty="EC", extras={"kty": "RSA"})
+    with pytest.raises(ValueError, match=r"UCPSigningKey\.extras key 'kty' collides"):
+        sk.to_dict()
+
+
+def test_ucp_signing_key_extras_non_reserved_pass_through() -> None:
+    sk = UCPSigningKey(kid="me", kty="EC", alg="ES256", crv="P-256", extras={"x": "abc", "y": "def"})
+    out = sk.to_dict()
+    assert out == {"kid": "me", "kty": "EC", "alg": "ES256", "crv": "P-256", "x": "abc", "y": "def"}
diff --git a/tests/test_ucp_jwks.py b/tests/test_ucp_jwks.py
@@ -314,16 +314,34 @@ def test_accepts_int_and_string(self) -> None:
         signed = sign_ucp_profile(profile, signing_key=signer.private_key, kid="k")
         assert verify_ucp_profile(signed, build_jwks_response([signer.public_jwk])) is True
 
-    def test_rejects_float_in_set(self) -> None:
+    def test_rejects_set_values_outright(self) -> None:
+        # `set` is not representable in JSON; the canonicalizer rejects it with a
+        # typed message before any element-level checks run. Mirrors node's
+        # `stableStringify: Set values are not allowed`.
         signer = generate_ucp_signing_key(kid="k")
         profile = {**_base_profile([signer.public_jwk]), "extras": {"vals": {0.5}}}
-        with pytest.raises(ValueError, match="rejects float"):
+        with pytest.raises(ValueError, match="set values are not allowed"):
             sign_ucp_profile(profile, signing_key=signer.private_key, kid="k")
 
-    def test_rejects_float_in_frozenset(self) -> None:
+    def test_rejects_frozenset_values_outright(self) -> None:
         signer = generate_ucp_signing_key(kid="k")
         profile = {**_base_profile([signer.public_jwk]), "extras": {"vals": frozenset({0.25})}}
-        with pytest.raises(ValueError, match="rejects float"):
+        with pytest.raises(ValueError, match="frozenset values are not allowed"):
+            sign_ucp_profile(profile, signing_key=signer.private_key, kid="k")
+
+    def test_rejects_empty_set_with_typed_message(self) -> None:
+        # Empty set + set-of-valid-strings would fall through `_reject_unsafe_numbers`
+        # cleanly and surface a raw `TypeError` from `json.dumps` later. The typed
+        # reject ensures callers get a guiding ValueError instead.
+        signer = generate_ucp_signing_key(kid="k")
+        profile = {**_base_profile([signer.public_jwk]), "extras": {"vals": set()}}
+        with pytest.raises(ValueError, match="set values are not allowed"):
+            sign_ucp_profile(profile, signing_key=signer.private_key, kid="k")
+
+    def test_rejects_set_of_valid_strings_with_typed_message(self) -> None:
+        signer = generate_ucp_signing_key(kid="k")
+        profile = {**_base_profile([signer.public_jwk]), "extras": {"vals": {"valid", "strings"}}}
+        with pytest.raises(ValueError, match="set values are not allowed"):
             sign_ucp_profile(profile, signing_key=signer.private_key, kid="k")
 
     def test_accepts_max_safe_int_boundary(self) -> None:
