From d32fe0d00c90da939e20f239d10d95be8b845fb8 Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Thu, 29 Oct 2026 10:23:00 -0500 Subject: [PATCH] fix(challenge,trace_binding): validate challenge secret/ttl/now/expiry and enforce strict iat/credential_id in trace_binding (#285) Validate secret bytes/length, ttl_seconds, now, and ASCII decimal expiry (preserving AttestationFailed when int(expiry_str) exceeds the integer string conversion limit) in ca2a_runtime.challenge, and enforce strict integer iat and non-empty credential_id in ca2a_runtime.trace_binding. Closes #285 Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/ca2a_runtime/challenge.py | 20 ++++++++- src/ca2a_runtime/trace_binding.py | 15 +++++++ tests/unit/test_challenge.py | 74 +++++++++++++++++++++++++++++++ 3 files changed, 107 insertions(+), 2 deletions(-) diff --git a/src/ca2a_runtime/challenge.py b/src/ca2a_runtime/challenge.py index 9a178e00..b73cef03 100644 --- a/src/ca2a_runtime/challenge.py +++ b/src/ca2a_runtime/challenge.py @@ -49,9 +49,15 @@ def _mac(secret: bytes, expiry: int, rand: str) -> str: return hmac.new(secret, f"{_PREFIX}.{expiry}.{rand}".encode(), sha256).hexdigest() +def _validate_secret(secret: bytes) -> None: + if type(secret) is not bytes or len(secret) < 16: + raise ValueError("secret must be bytes of length at least 16") + + def issue_challenge(secret: bytes, *, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> str: """Issue a challenge that expires ``ttl_seconds`` from now.""" - if ttl_seconds <= 0: + _validate_secret(secret) + if type(ttl_seconds) is not int or ttl_seconds <= 0: raise ValueError( "ttl_seconds must be positive; a challenge that never validates is not a challenge" ) @@ -68,7 +74,15 @@ def verify_challenge(secret: bytes, challenge: str, *, now: int | None = None) - arithmetic, and reporting "expired" for a forged challenge tells the sender their forgery was well-formed. """ - parts = (challenge or "").split(".") + _validate_secret(secret) + if now is not None and type(now) is not int: + raise ValueError("now must be an integer timestamp") + if not isinstance(challenge, str): + raise AttestationFailed( + "challenge is malformed", + detail="expected v1...", + ) + parts = challenge.split(".") if len(parts) != 4 or parts[0] != _PREFIX: raise AttestationFailed( "challenge is malformed", @@ -83,6 +97,8 @@ def verify_challenge(secret: bytes, challenge: str, *, now: int | None = None) - "challenge is malformed", detail="the MAC is not 64 lowercase hex characters", ) + if not expiry_str.isascii() or not expiry_str.isdecimal(): + raise AttestationFailed("challenge expiry is not an integer") try: expiry = int(expiry_str) except ValueError as exc: diff --git a/src/ca2a_runtime/trace_binding.py b/src/ca2a_runtime/trace_binding.py index 802454ae..dd1fcf2a 100644 --- a/src/ca2a_runtime/trace_binding.py +++ b/src/ca2a_runtime/trace_binding.py @@ -134,11 +134,22 @@ def build_trace_record( The returned dict has no ``cnf`` or ``signature`` yet; pass it to :func:`sign_trace_record` to get a schema-valid, signed record. """ + if not isinstance(subject, str) or not subject: + raise ValueError("subject must be a non-empty string") + if type(iat) is not int or not 0 < iat <= 9_007_199_254_740_991: + raise ValueError("iat must be a positive RFC 8785 safe integer") if (parent_record_hash is None) != (credential_id is None): raise ValueError( "the delegation block needs both parent_record_hash and credential_id; " "a root hop supplies neither" ) + if parent_record_hash is not None and ( + not isinstance(parent_record_hash, str) + or not parent_record_hash + or not isinstance(credential_id, str) + or not credential_id + ): + raise ValueError("parent_record_hash and credential_id must be non-empty strings") runtime: dict[str, Any] = { "platform": context.runtime_platform, @@ -238,9 +249,13 @@ def emit_dag(hops: list[HopSpec]) -> list[dict[str, Any]]: ``delegation`` block referencing its parent's ``trace_record_hash`` and its own ``credential_id``. Returns the signed records in root-to-leaf order. """ + if not isinstance(hops, list) or not hops: + raise ValueError("hops must be a non-empty list of HopSpec entries") records: list[dict[str, Any]] = [] parent_hash: str | None = None for i, hop in enumerate(hops): + if i > 0 and (not isinstance(hop.credential_id, str) or not hop.credential_id): + raise ValueError(f"non-root hop {i} must supply a non-empty credential_id") credential_id = None if i == 0 else hop.credential_id record = build_trace_record( subject=hop.subject, diff --git a/tests/unit/test_challenge.py b/tests/unit/test_challenge.py index 0099f1b1..58f02e4c 100644 --- a/tests/unit/test_challenge.py +++ b/tests/unit/test_challenge.py @@ -124,3 +124,77 @@ def test_a_malformed_mac_is_refused_as_attestation_failure(mac: str) -> None: prefix, expiry, rand, _ = issue_challenge(secret).split(".") with pytest.raises(AttestationFailed, match="malformed"): verify_challenge(secret, f"{prefix}.{expiry}.{rand}.{mac}") + + +@pytest.mark.parametrize("bad_ttl", [True, False, -5, 1.5, "60"]) +def test_issue_challenge_rejects_boolean_or_non_int_ttl(bad_ttl) -> None: + with pytest.raises(ValueError): + issue_challenge(generate_secret(), ttl_seconds=bad_ttl) + + +@pytest.mark.parametrize("bad_secret", [b"", b"short", "32-char-string------------------", None]) +def test_challenge_rejects_invalid_secret(bad_secret) -> None: + with pytest.raises(ValueError, match="secret"): + issue_challenge(bad_secret) + with pytest.raises(ValueError, match="secret"): + verify_challenge(bad_secret, "v1.1.aa.bb") + + +@pytest.mark.parametrize("bad_now", [True, False, 1.5, "100"]) +def test_verify_challenge_rejects_boolean_or_non_int_now(bad_now) -> None: + secret = generate_secret() + ch = issue_challenge(secret) + with pytest.raises(ValueError, match="now"): + verify_challenge(secret, ch, now=bad_now) + + +@pytest.mark.parametrize( + "bad_challenge", + [ + 123, + None, + ["v1"], + f"v1.+100.{'a' * 32}.{'0' * 64}", + f"v1. 100 .{'a' * 32}.{'0' * 64}", + f"v1.{'9' * 4301}.{'a' * 32}.{'0' * 64}", + ], +) +def test_verify_challenge_rejects_non_str_and_non_decimal_expiry(bad_challenge) -> None: + with pytest.raises(AttestationFailed): + verify_challenge(generate_secret(), bad_challenge) + + +@pytest.mark.parametrize("bad_iat", [True, False, 0, -1, 1.5, 2**53]) +def test_build_trace_record_and_emit_dag_validate_inputs(bad_iat) -> None: + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + from ca2a_runtime.trace_binding import HopContext, HopSpec, build_trace_record, emit_dag + + ctx = HopContext.software( + model_provider="p", + model_id="m", + image_label="img", + policy_bundle_hash="sha256:" + "a" * 64, + ) + with pytest.raises(ValueError): + build_trace_record(subject="spiffe://example/a", iat=bad_iat, context=ctx) + with pytest.raises(ValueError): + build_trace_record(subject="", iat=1000, context=ctx) + with pytest.raises(ValueError): + build_trace_record( + subject="spiffe://example/a", + iat=1000, + context=ctx, + credential_id="", + parent_record_hash="sha256:" + "b" * 64, + ) + with pytest.raises(ValueError): + emit_dag([]) + key = Ed25519PrivateKey.generate() + with pytest.raises(ValueError, match="non-root hop"): + emit_dag( + [ + HopSpec(subject="spiffe://example/0", signing_key=key, context=ctx, iat=1000), + HopSpec(subject="spiffe://example/1", signing_key=key, context=ctx, iat=1000), + ] + )