From 951c2627ba87f5438e46ff46ee5a8b9e19d885ed Mon Sep 17 00:00:00 2001 From: Louielunz <48041247+lywinged@users.noreply.github.com> Date: Wed, 12 Aug 2026 18:13:27 +0000 Subject: [PATCH] feat(provenance): freshness and revocation, with the semantics sign.py documents Closes #151. issued_at has been required since this format existed, and _check_structure explains why: "A record with no issue time cannot be aged, so a consumer has no way to reject a stale one." Nothing aged it. Walking the module for Compare nodes mentioning issued_at gave two, the type check and build_record's default, and the only time.time() call was build_record stamping it. sign.verify_record took revocation; provenance.verify_record took nothing. That is not "one checks and the other does not" - Trust Records skip revocation by default too - it is that one format gave the consumer the hook and the other did not, and the caller-side substitute is not equivalent. _key_identifiers is private and returns the thumbprint *and* the kid, so a caller writing the check by hand reaches for the thumbprint and misses every entry listed by kid, which is what kid is for. It diverges in the direction of accepting a revoked key. def verify_record(record, trusted_jwk, *, revocation: RevocationStore | None = None, max_age_seconds: int | None = None, max_future_skew_seconds: int = 300) -> None: Nothing new was needed: RevocationStore and _check_not_revoked already exist and are reused rather than reimplemented. max_age_seconds defaults to None, not to the 86400 of a Trust Record. A provenance record describes an artifact by immutable digest, like a package signature, and those are conventionally valid indefinitely; a default bound would be wrong. The docstring names endpoint identity as the case where that reasoning does not hold, since a URL and an SPKI digest decay. max_future_skew_seconds is enforced whether or not an age bound is set. Adding max_age_seconds alone would have shipped the defect #155 had just fixed for Trust Records: a far-future issued_at sits inside any later age window until that time arrives. Revocation is consulted before the signature, because a signature made by a revoked key stays cryptographically valid and the verifier is the only place the fact can be applied. Ten tests. Load-bearing, measured rather than assumed: removing the freshness block fails three, removing only the future-skew check fails one, removing the revocation call fails three, and keying revocation on the thumbprint alone - the caller-side substitute - fails one. That last is the kid case, and it is why the hook has to live here rather than in the caller. 353 pass; ruff and mypy clean, with ruff clean on the same files before this change. Merges with #149 without conflict; that PR makes the revocation half cheaper by bringing jwk_thumbprint into this module, but neither depends on the other. Signed-off-by: Louielunz <48041247+lywinged@users.noreply.github.com> Co-authored-by: Claude Opus 5 (1M context) --- src/agentrust_trace/provenance.py | 63 +++++++++++++++- tests/test_provenance.py | 119 +++++++++++++++++++++++++++++- 2 files changed, 179 insertions(+), 3 deletions(-) diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index 3e00fe06..37af2192 100644 --- a/src/agentrust_trace/provenance.py +++ b/src/agentrust_trace/provenance.py @@ -21,7 +21,13 @@ import time from typing import Any -from agentrust_trace.sign import _canonical_bytes, _pubkey_from_jwk, key_to_jwk +from agentrust_trace.sign import ( + RevocationStore, + _canonical_bytes, + _check_not_revoked, + _pubkey_from_jwk, + key_to_jwk, +) __all__ = [ "FORMAT", @@ -212,13 +218,39 @@ def sign_record(record: dict[str, Any], key: Any) -> dict[str, Any]: return {**payload, "signature": sig} -def verify_record(record: dict[str, Any], trusted_jwk: dict[str, Any]) -> None: +def verify_record( + record: dict[str, Any], + trusted_jwk: dict[str, Any], + *, + revocation: RevocationStore | None = None, + max_age_seconds: int | None = None, + max_future_skew_seconds: int = 300, +) -> None: """Verify structure and signature. Raises :class:`ProvenanceError` on failure. *trusted_jwk* is required and is never taken from the record. Verifying a document against a key it supplies proves only that it is internally consistent, which is what a forged record is. + ``revocation`` is consulted before the signature is checked, with exactly the + semantics :func:`sign.verify_record` documents: a container of revoked + identifiers or a callable, matched against the trusted key's RFC 7638 + thumbprint *and* its ``kid``. Both a revoked key and an unreachable store fail + closed. ``None`` skips the check and keeps verification offline; offline + verification cannot prove non-revocation. + + ``max_age_seconds`` bounds how old ``issued_at`` may be. It defaults to + ``None``, unlike the 86400 of a Trust Record, because a provenance record + describes an artifact by immutable digest and those are conventionally valid + indefinitely. A record carrying ``endpoint`` identity is the case where that + reasoning does not hold — a URL and an SPKI digest decay — so a consumer + relying on ``endpoint`` should pass a bound. + + ``max_future_skew_seconds`` (default 300) is enforced whether or not an age + bound is set. Without it a far-future ``issued_at`` stays inside any later + ``max_age_seconds`` window until that time arrives, which is the defect + #155 fixed for Trust Records. + **This does not check the server.** It checks the paper. Call :func:`check_tool_catalog` with the tools the server actually offered. """ @@ -247,6 +279,33 @@ def verify_record(record: dict[str, Any], trusted_jwk: dict[str, Any]) -> None: if not _DIGEST_RE.match(str(catalog.get("hash", ""))): raise ProvenanceError("tool_catalog.hash is not a sha256: digest") + # Freshness. `issued_at` has been required and type-checked since the format + # existed, with an error message explaining that a record with no issue time + # cannot be aged; this is the step that reads it. `_check_structure` above has + # already established it is a non-negative int. + if max_future_skew_seconds < 0: + raise ProvenanceError("max_future_skew_seconds must be non-negative") + age = time.time() - int(record["issued_at"]) + if age < -max_future_skew_seconds: + raise ProvenanceError( + f"record is dated {int(-age)}s in the future, exceeds " + f"max_future_skew_seconds={max_future_skew_seconds}" + ) + if max_age_seconds is not None and age > max_age_seconds: + raise ProvenanceError( + f"record is stale: issued_at is {int(age)}s old, exceeds " + f"max_age_seconds={max_age_seconds}" + ) + + # Revocation, before the signature rather than after. A signature made by a + # revoked key stays cryptographically valid for ever, so the verifier is the + # only place the fact can be applied. + if revocation is not None: + try: + _check_not_revoked(trusted_jwk, revocation) + except ValueError as exc: + raise ProvenanceError(str(exc)) from exc + signature = record.get("signature") if not signature: raise ProvenanceError("record carries no signature") diff --git a/tests/test_provenance.py b/tests/test_provenance.py index 29c1c12c..648accbe 100644 --- a/tests/test_provenance.py +++ b/tests/test_provenance.py @@ -8,6 +8,9 @@ from __future__ import annotations +import base64 +import time + import pytest from agentrust_trace.provenance import ( @@ -20,7 +23,7 @@ tool_catalog_hash, verify_record, ) -from agentrust_trace.sign import generate_key, key_to_jwk +from agentrust_trace.sign import _canonical_bytes, generate_key, jwk_thumbprint, key_to_jwk DIGEST = "sha256:" + "a" * 64 OTHER_DIGEST = "sha256:" + "b" * 64 @@ -358,3 +361,117 @@ def test_a_trailing_newline_does_not_satisfy_a_pattern(field: str, value: str) - with pytest.raises(ProvenanceError): verify_record(sign_record(record, key), key_to_jwk(key)) + + +# --- freshness and revocation, the affordances sign.verify_record already has ----- +# +# `issued_at` has been required since the format existed, with an error message saying a +# record with no issue time cannot be aged. Nothing aged it. These cover the step that +# reads it, and the revocation hook whose absence had no caller-side substitute. + + +def _signed_at(issued_at: int, key): + return sign_record(_record(issued_at=issued_at), key) + + +def test_an_old_record_is_still_accepted_by_default() -> None: + """The default is deliberate, not an oversight. + + A provenance record describes an artifact by immutable digest, like a package + signature, and those are conventionally valid indefinitely. So `max_age_seconds` + defaults to None here, unlike the 86400 of a Trust Record. + """ + key = generate_key() + verify_record(_signed_at(int(time.time()) - 400 * 86400, key), key_to_jwk(key)) + + +def test_an_old_record_is_refused_when_the_consumer_asks_for_a_bound() -> None: + """The hook the format did not give the consumer.""" + key = generate_key() + signed = _signed_at(int(time.time()) - 400 * 86400, key) + with pytest.raises(ProvenanceError, match="stale"): + verify_record(signed, key_to_jwk(key), max_age_seconds=86400) + + +def test_a_future_dated_record_is_refused_without_asking() -> None: + """Enforced whether or not an age bound is set. + + Without this, a far-future `issued_at` sits inside any later `max_age_seconds` + window until that time arrives. Adding the age bound alone would have shipped + the defect #155 had just fixed for Trust Records. + """ + key = generate_key() + signed = _signed_at(int(time.time()) + 3600, key) + with pytest.raises(ProvenanceError, match="future"): + verify_record(signed, key_to_jwk(key)) + + +def test_clock_skew_inside_the_tolerance_is_accepted() -> None: + key = generate_key() + verify_record(_signed_at(int(time.time()) + 60, key), key_to_jwk(key)) + + +def test_a_negative_skew_bound_is_refused_rather_than_applied() -> None: + key = generate_key() + signed = _signed_at(int(time.time()), key) + with pytest.raises(ProvenanceError, match="non-negative"): + verify_record(signed, key_to_jwk(key), max_future_skew_seconds=-1) + + +def test_a_revoked_key_is_refused() -> None: + """A signature by a revoked key stays valid for ever; the verifier is the only + place the fact can be applied.""" + key = generate_key() + signed = sign_record(_record(), key) + crl = {jwk_thumbprint(key_to_jwk(key))} + with pytest.raises(ProvenanceError, match="revoked"): + verify_record(signed, key_to_jwk(key), revocation=crl) + + +def test_a_key_revoked_under_its_kid_is_refused() -> None: + """The case a hand-written caller-side check misses. + + `_key_identifiers` is private, and it returns the thumbprint *and* the `kid`. A + caller writing the check themselves reaches for the thumbprint and misses every + entry listed by `kid`, which is what `kid` is for. Signed by hand because + `sign_record` builds `cnf` from `key_to_jwk` and cannot emit one carrying `kid`. + """ + key = generate_key() + jwk = {**key_to_jwk(key), "kid": "acme-2026q3"} + payload = {**_record(), "cnf": {"jwk": jwk}} + body = _canonical_bytes({k: v for k, v in payload.items() if k != "signature"}) + signature = base64.urlsafe_b64encode(key.sign(body)).rstrip(b"=").decode() + signed = {**payload, "signature": signature} + + verify_record(signed, jwk) # unrevoked: fine + assert jwk_thumbprint(jwk) not in {"acme-2026q3"} # the substitute misses it + with pytest.raises(ProvenanceError, match="revoked"): + verify_record(signed, jwk, revocation={"acme-2026q3"}) + + +def test_an_unreachable_revocation_source_fails_closed() -> None: + """Not evidence that the key is unrevoked.""" + def unreachable(_identifier: str) -> bool: + raise ConnectionError("CRL endpoint unreachable") + + key = generate_key() + signed = sign_record(_record(), key) + with pytest.raises(ProvenanceError, match="could not be determined"): + verify_record(signed, key_to_jwk(key), revocation=unreachable) + + +def test_an_unrevoked_key_passes_the_check() -> None: + """The check must not reject what it should accept.""" + key = generate_key() + signed = sign_record(_record(), key) + verify_record(signed, key_to_jwk(key), revocation={"some-other-key"}) + + +def test_revocation_is_off_by_default_and_verification_stays_offline() -> None: + """`revocation=None` skips the check, as sign.verify_record documents. + + Offline verification cannot prove non-revocation; the point is that the consumer + now has somewhere to put a store, not that one is imposed. + """ + key = generate_key() + verify_record(sign_record(_record(), key), key_to_jwk(key))