fix(delegation,attestation): enforce RFC 8785 safe-integer bounds and canonical lowercase hex on HolderProof and CallerPossession (#274)

src/ca2a_runtime/attestation.py

@@ -38,6 +38,7 @@
 
 import hashlib
 import hmac
+import re
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any
@@ -221,6 +222,8 @@ def appraise_caller(
 CALLER_OFFER_VERSION = "ca2a-caller-offer-v2"
 _POSSESSION_KEY_INFO = b"ca2a/caller-possession/v2/key"
 _POSSESSION_STATEMENT = b"ca2a/caller-possession/v2/statement\x00"
+_POSSESSION_MAC_RE = re.compile(r"[0-9a-f]{64}")
+_POSSESSION_FIELDS = frozenset({"version", "mac"})
 
 
 @dataclass(frozen=True)
@@ -239,11 +242,17 @@ def from_dict(cls, data: Any) -> CallerPossession:
             raise AttestationFailed(
                 "caller_possession must be an object", detail=f"got {type(data).__name__}"
             )
+        unknown = sorted(set(data) - _POSSESSION_FIELDS)
+        if unknown:
+            raise AttestationFailed(
+                "caller_possession contains unknown fields",
+                detail=f"unknown={unknown}",
+            )
         version, mac = data.get("version"), data.get("mac")
         if not isinstance(version, str) or not version:
             raise AttestationFailed("caller_possession.version must be a non-empty string")
-        if not isinstance(mac, str) or not mac:
-            raise AttestationFailed("caller_possession.mac must be a non-empty string")
+        if not isinstance(mac, str) or not _POSSESSION_MAC_RE.fullmatch(mac):
+            raise AttestationFailed("caller_possession.mac must be 64 lowercase hex characters")
         return cls(version=version, mac=mac)
 
 
@@ -359,6 +368,8 @@ def verify_caller_possession(
             "caller proof of possession has an unsupported version",
             detail=f"got {possession.version!r}, expected {CALLER_OFFER_VERSION!r}",
         )
+    if not isinstance(possession.mac, str) or not _POSSESSION_MAC_RE.fullmatch(possession.mac):
+        raise AttestationFailed("caller_possession.mac must be 64 lowercase hex characters")
     callee_channel_key = callee_private_key.public_key().public_bytes_raw().hex()
     if transcript.get("callee_channel_key") != callee_channel_key:
         raise AttestationFailed("possession transcript names a different callee channel key")

src/ca2a_runtime/delegation/credential.py

@@ -36,7 +36,7 @@
     Ed25519PublicKey,
 )
 
-from ca2a_runtime.canonical import canonicalize
+from ca2a_runtime.canonical import _MAX_SAFE_INTEGER, canonicalize
 from ca2a_runtime.delegation.revocation import (
     REVOCATION_NOT_CHECKED,
     RevocationSnapshot,
@@ -102,9 +102,17 @@ def __post_init__(self) -> None:
         # constructor, and sign() would otherwise put a bound outside the
         # documented wire format into a signed body that this implementation's
         # own parser rejects. replace() re-runs this, so no path around it.
+        if (
+            isinstance(self.depth, bool)
+            or not isinstance(self.depth, int)
+            or not 0 <= self.depth <= _MAX_SAFE_INTEGER
+        ):
+            raise InvalidCredential("depth must be a non-negative safe integer")
         for name, bound in (("not_before", self.not_before), ("not_after", self.not_after)):
             if bound is not None and (
-                isinstance(bound, bool) or not isinstance(bound, int) or bound < 0
+                isinstance(bound, bool)
+                or not isinstance(bound, int)
+                or not 0 <= bound <= _MAX_SAFE_INTEGER
             ):
                 raise InvalidCredential(f"{name} must be a non-negative integer when present")
         if (
@@ -205,7 +213,11 @@ def from_dict(cls, data: dict[str, Any]) -> DelegationCredential:
             or len(set(scope)) != len(scope)
         ):
             raise InvalidCredential("scope must be a non-empty array of unique non-empty strings")
-        if isinstance(depth, bool) or not isinstance(depth, int) or depth < 0:
+        if (
+            isinstance(depth, bool)
+            or not isinstance(depth, int)
+            or not 0 <= depth <= _MAX_SAFE_INTEGER
+        ):
             raise InvalidCredential("depth must be a non-negative integer")
         if parent_id is not None and (not isinstance(parent_id, str) or not parent_id):
             raise InvalidCredential("parent_id must be a non-empty string or null")
@@ -218,7 +230,9 @@ def from_dict(cls, data: dict[str, Any]) -> DelegationCredential:
         # so null would be a second wire form for the same signed object.
         for name, bound in (("not_before", not_before), ("not_after", not_after)):
             if name in data and (
-                isinstance(bound, bool) or not isinstance(bound, int) or bound < 0
+                isinstance(bound, bool)
+                or not isinstance(bound, int)
+                or not 0 <= bound <= _MAX_SAFE_INTEGER
             ):
                 raise InvalidCredential(f"{name} must be a non-negative integer when present")
         if not_before is not None and not_after is not None and not_before > not_after:
@@ -279,13 +293,15 @@ def verify_chain(
     # compared against holds the same line, or True / 1.5 / -1 from a library
     # caller would silently decide validity. (The CLI is already argparse-typed.)
     if at_time is not None and (
-        isinstance(at_time, bool) or not isinstance(at_time, int) or at_time < 0
+        isinstance(at_time, bool)
+        or not isinstance(at_time, int)
+        or not 0 <= at_time <= _MAX_SAFE_INTEGER
     ):
         raise ValueError("at_time must be a non-negative integer or None")
     if max_revocation_staleness is not None and (
         isinstance(max_revocation_staleness, bool)
         or not isinstance(max_revocation_staleness, int)
-        or max_revocation_staleness < 0
+        or not 0 <= max_revocation_staleness <= _MAX_SAFE_INTEGER
     ):
         raise ValueError("max_revocation_staleness must be a non-negative integer or None")
 

src/ca2a_runtime/delegation/holder.py

@@ -67,6 +67,7 @@
 from __future__ import annotations
 
 import hashlib
+import re
 from dataclasses import dataclass
 from typing import Any
 
@@ -84,6 +85,8 @@
 #: Domain separator. A signature made for a holder proof must never be
 #: mistakable for a signature over a credential body or a TRACE record.
 PROOF_DOMAIN = "ca2a-holder-proof-v1"
+_HEX_64_RE = re.compile(r"[0-9a-f]{128}")
+_HOLDER_PROOF_FIELDS = frozenset({"challenge", "signature"})
 
 __all__ = [
     "PROOF_DOMAIN",
@@ -154,12 +157,20 @@ def from_dict(cls, data: Any) -> HolderProof:
                 "holder_proof must be an object",
                 detail=f"got {type(data).__name__}",
             )
+        unknown = sorted(set(data) - _HOLDER_PROOF_FIELDS)
+        if unknown:
+            raise HolderProofInvalid(
+                "holder_proof contains unknown fields",
+                detail=f"unknown={unknown}",
+            )
         challenge = data.get("challenge")
         signature = data.get("signature")
         if not isinstance(challenge, str) or not challenge:
             raise HolderProofInvalid("holder_proof.challenge must be a non-empty string")
-        if not isinstance(signature, str) or not signature:
-            raise HolderProofInvalid("holder_proof.signature must be a non-empty string")
+        if not isinstance(signature, str) or not _HEX_64_RE.fullmatch(signature):
+            raise HolderProofInvalid(
+                "holder_proof.signature must be a lowercase 64-byte Ed25519 signature in hex"
+            )
         return cls(challenge=challenge, signature=signature)
 
 
@@ -254,6 +265,10 @@ def verify_holder_proof(
         caller_channel_key=caller_channel_key,
         parent_record_hash=parent_record_hash,
     )
+    if not isinstance(proof.signature, str) or not _HEX_64_RE.fullmatch(proof.signature):
+        raise HolderProofInvalid(
+            "holder_proof.signature must be a lowercase 64-byte Ed25519 signature in hex"
+        )
     try:
         pub = Ed25519PublicKey.from_public_bytes(bytes.fromhex(leaf.subject))
         pub.verify(bytes.fromhex(proof.signature), canonicalize(body))

src/ca2a_runtime/delegation/revocation.py

@@ -55,6 +55,7 @@
     Ed25519PublicKey,
 )
 
+from ca2a_runtime.canonical import _MAX_SAFE_INTEGER
 from ca2a_runtime.canonical import canonicalize as canonical_bytes
 from ca2a_runtime.errors import InvalidRevocation
 
@@ -82,7 +83,9 @@ def credential_digest(credential: DelegationCredential) -> str:
 
 
 def _non_negative_int(value: Any) -> bool:
-    return isinstance(value, int) and not isinstance(value, bool) and value >= 0
+    return (
+        isinstance(value, int) and not isinstance(value, bool) and 0 <= value <= _MAX_SAFE_INTEGER
+    )
 
 
 @dataclass(frozen=True)

tests/unit/test_holder_binding.py

@@ -389,13 +389,70 @@ def test_missing_challenge_context_fails_closed() -> None:
 
 
 @pytest.mark.parametrize(
-    "bad", [{"challenge": "c"}, {"signature": "ab"}, {"challenge": "", "signature": "ab"}, "str", 7]
+    "bad",
+    [
+        {"challenge": "c"},
+        {"signature": "ab"},
+        {"challenge": "", "signature": "ab"},
+        {"challenge": "c", "signature": "ab"},
+        {"challenge": "c", "signature": "A" * 128},
+        {"challenge": "c", "signature": " " + "a" * 128},
+        {"challenge": "c", "signature": "a" * 128, "extra": 1},
+        "str",
+        7,
+    ],
 )
 def test_malformed_proof_is_rejected(bad: object) -> None:
     with pytest.raises(HolderProofInvalid):
         HolderProof.from_dict(bad)
 
 
+def test_safe_integer_bounds_and_possession_hex_validation() -> None:
+    from ca2a_runtime.attestation import (
+        CALLER_OFFER_VERSION,
+        CallerPossession,
+        verify_caller_possession,
+    )
+    from ca2a_runtime.canonical import _MAX_SAFE_INTEGER
+    from ca2a_runtime.delegation.credential import verify_chain
+    from ca2a_runtime.delegation.revocation import RevocationSnapshot, RevocationStatement
+    from ca2a_runtime.errors import AttestationFailed, InvalidCredential, InvalidRevocation
+
+    chain, _keys = _chain()
+    raw_cred = {**chain[0].body(), "signature": chain[0].signature}
+    with pytest.raises(InvalidCredential):
+        DelegationCredential.from_dict({**raw_cred, "depth": _MAX_SAFE_INTEGER + 1})
+    with pytest.raises(InvalidCredential):
+        DelegationCredential.from_dict({**raw_cred, "not_after": _MAX_SAFE_INTEGER + 1})
+    with pytest.raises(ValueError, match="at_time"):
+        verify_chain(chain, trusted_root_issuers={chain[0].issuer}, at_time=_MAX_SAFE_INTEGER + 1)
+    with pytest.raises(InvalidRevocation):
+        RevocationStatement(
+            revoked_digest="sha256:" + "a" * 64,
+            revoker="b" * 64,
+            issued_at=_MAX_SAFE_INTEGER + 1,
+        )
+    with pytest.raises(InvalidRevocation):
+        RevocationSnapshot.from_dict({"as_of": _MAX_SAFE_INTEGER + 1, "revocations": []})
+
+    for bad_poss in [
+        {"version": CALLER_OFFER_VERSION, "mac": "A" * 64},
+        {"version": CALLER_OFFER_VERSION, "mac": "a" * 64, "extra": 1},
+        {"version": CALLER_OFFER_VERSION, "mac": "\ud800" + "a" * 63},
+    ]:
+        with pytest.raises(AttestationFailed):
+            CallerPossession.from_dict(bad_poss)
+    with pytest.raises(AttestationFailed):
+        verify_caller_possession(
+            CallerPossession(CALLER_OFFER_VERSION, "\ud800"),
+            callee_private_key=TEST_CALLEE_KEY,
+            transcript={
+                "callee_channel_key": TEST_CALLEE_PUB,
+                "caller_channel_key": TEST_CALLEE_PUB,
+            },
+        )
+
+
 def test_unauthenticated_caller_never_reaches_policy_evaluation() -> None:
     """A caller that proved nothing must not get back a signed denial record."""
     chain, _keys = _chain()