fix(verify): enforce credential validity windows and hop timestamp ordering in cross_check_trace_dag (#306)

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).

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

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

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)