diff --git a/docs/spec/verification-library.md b/docs/spec/verification-library.md index 4ad1626b..d721ec65 100644 --- a/docs/spec/verification-library.md +++ b/docs/spec/verification-library.md @@ -54,4 +54,11 @@ record's free-text `subject` field (a SPIFFE/DID URI) to the credential subject, and does not prove task completion. Those are different namespaces; `cnf.jwk` is already a raw key and is what the hop "acted under." +It also requires each signed hop `iat` to fall within that hop's inclusive +credential validity window and not precede the previous hop's `iat`. The +separate `verify_chain` `at_time` checks the whole chain at one evaluation +instant; for historical checks, choose the action's decision time. Signed +timestamps are assertions by the signer, not independent proof of when a +record was created. + The API name uses DAG terminology, but the input is one ordered path, not an arbitrary branching tree. The unsigned `DelegationRecord` helper has a separate `verify_dag` consistency check; it is not a substitute for signature verification. See [provenance DAG](provenance-dag.md) and [TRACE A2A profile](trace-a2a-profile.md). diff --git a/src/ca2a_verify/dag.py b/src/ca2a_verify/dag.py index fa6a323b..ae23e199 100644 --- a/src/ca2a_verify/dag.py +++ b/src/ca2a_verify/dag.py @@ -240,6 +240,7 @@ def cross_check_trace_dag(records: list[dict[str, Any]], chain: list[DelegationC raise ProvenanceLinkBroken( f"DAG length {len(records)} does not match chain length {len(chain)}" ) + previous_iat: int | None = None for i, record in enumerate(records): if i > 0: delegation = record.get("delegation") or {} @@ -250,3 +251,20 @@ def cross_check_trace_dag(records: list[dict[str, Any]], chain: list[DelegationC raise ProvenanceLinkBroken( f"record {i} was signed by a key that is not chain[{i}].subject" ) + iat = record.get("iat") + if isinstance(iat, bool) or not isinstance(iat, int): + raise ProvenanceLinkBroken(f"record {i} iat must be an integer") + credential = chain[i] + if credential.not_before is not None and iat < credential.not_before: + raise ProvenanceLinkBroken( + f"record {i} iat {iat} precedes credential not_before {credential.not_before}" + ) + if credential.not_after is not None and iat > credential.not_after: + raise ProvenanceLinkBroken( + f"record {i} iat {iat} exceeds credential not_after {credential.not_after}" + ) + if previous_iat is not None and iat < previous_iat: + raise ProvenanceLinkBroken( + f"record {i} iat {iat} precedes parent iat {previous_iat}" + ) + previous_iat = iat diff --git a/tests/unit/test_lineage_cli.py b/tests/unit/test_lineage_cli.py index e8ecd832..3bcfe103 100644 --- a/tests/unit/test_lineage_cli.py +++ b/tests/unit/test_lineage_cli.py @@ -62,6 +62,21 @@ def test_signed_path_authenticates(evidence, capsys): assert out["revocation"] == "not_checked" +def test_backdated_signed_hop_is_not_authenticated(evidence, capsys): + chain, keys = evidence[:2] + context = HopContext.software( + model_provider="test", model_id="test", image_label="test", policy_bundle_hash=digest(b"policy") + ) + records = emit_dag([ + HopSpec(f"did:web:peer{i}", keys[i], context, timestamp, c.credential_id) + for i, (c, timestamp) in enumerate(zip(chain, [1750000000, 1749999999, 1750000000], strict=True)) + ]) + rc, out = run(evidence, capsys, records) + assert rc == 1 + assert out["verified"] is False + assert out["code"] == "PROVENANCE_LINK_BROKEN" + + def test_native_default_is_not_a_verification_success(evidence, capsys): rc, out = run(evidence, capsys, [r.body() for r in evidence[3]], "verify-dag") assert rc == 1 diff --git a/tests/unit/test_trace_binding.py b/tests/unit/test_trace_binding.py index 873e1aa0..ac5ce158 100644 --- a/tests/unit/test_trace_binding.py +++ b/tests/unit/test_trace_binding.py @@ -10,6 +10,7 @@ import hashlib import time +from dataclasses import replace import pytest import rfc8785 @@ -159,6 +160,34 @@ def test_cross_check_ties_dag_to_chain() -> None: cross_check_trace_dag(records, chain) # does not raise +@pytest.mark.parametrize( + ("times", "failure"), + [ + ([1700000999, 1700001000, 1700001000], "not_before"), + ([1700001000, 1700001000, 1700002001], "not_after"), + ([1700001100, 1700001099, 1700001100], "parent iat"), + ], +) +def test_cross_check_rejects_invalid_hop_times(times, failure) -> None: + chain, keys = build_chain_with_keys(_THREE_HOP, not_before=1700001000, not_after=1700002000) + hops = [replace(hop, iat=timestamp) for hop, timestamp in zip(_hops(chain, keys), times, strict=True)] + records = emit_dag(hops) + verify_trace_dag(records, trusted_keys=_trusted(keys)) + with pytest.raises(ProvenanceLinkBroken, match=failure): + cross_check_trace_dag(records, chain) + + +def test_cross_check_accepts_inclusive_bounds_and_equal_times() -> None: + chain, keys = build_chain_with_keys(_THREE_HOP, not_before=1700001000, not_after=1700002000) + hops = [ + replace(hop, iat=timestamp) + for hop, timestamp in zip(_hops(chain, keys), [1700001000, 1700002000, 1700002000], strict=True) + ] + records = emit_dag(hops) + verify_trace_dag(records, trusted_keys=_trusted(keys)) + cross_check_trace_dag(records, chain) + + def test_cross_check_rejects_credential_mismatch() -> None: chain, keys = build_chain_with_keys(_TWO_HOP) hops = _hops(chain, keys)