diff --git a/src/ca2a_runtime/attestation.py b/src/ca2a_runtime/attestation.py index 1dbda89a..c0223ba7 100644 --- a/src/ca2a_runtime/attestation.py +++ b/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") diff --git a/src/ca2a_runtime/delegation/credential.py b/src/ca2a_runtime/delegation/credential.py index c4245589..74c7d89b 100644 --- a/src/ca2a_runtime/delegation/credential.py +++ b/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") diff --git a/src/ca2a_runtime/delegation/holder.py b/src/ca2a_runtime/delegation/holder.py index 99a9799f..7470b185 100644 --- a/src/ca2a_runtime/delegation/holder.py +++ b/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)) diff --git a/src/ca2a_runtime/delegation/revocation.py b/src/ca2a_runtime/delegation/revocation.py index a28d9f16..0aaf85bf 100644 --- a/src/ca2a_runtime/delegation/revocation.py +++ b/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) diff --git a/tests/unit/test_holder_binding.py b/tests/unit/test_holder_binding.py index dd979aa8..549f07d1 100644 --- a/tests/unit/test_holder_binding.py +++ b/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()