fix(challenge,trace_binding): validate challenge secret/ttl/now/expiry and enforce strict iat/credential_id in trace_binding (#285)

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.<expiry>.<random>.<mac>",
+        )
+    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:

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,

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),
+            ]
+        )