feat(provenance): freshness and revocation, with the semantics sign.py documents

src/agentrust_trace/provenance.py

@@ -22,7 +22,9 @@
 from typing import Any
 
 from agentrust_trace.sign import (
+    RevocationStore,
     _canonical_bytes,
+    _check_not_revoked,
     _pubkey_from_jwk,
     jwk_thumbprint,
     key_to_jwk,
@@ -217,13 +219,57 @@ 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 _check_seconds(name: str, value: Any, *, optional: bool = False) -> None:
+    """Reject a malformed policy input instead of silently acting on it.
+
+    A verifier's age policy is configuration, and a wrong one fails in the
+    direction that matters: ``max_age_seconds=-1`` is not a stricter bound, it
+    classifies every record ever issued as stale, and a caller who meant to
+    disable the bound would see a uniform refusal rather than an error naming
+    the cause. ``bool`` is excluded explicitly because it is a subclass of
+    ``int`` in Python, so ``True`` would otherwise pass as one second.
+    """
+    if optional and value is None:
+        return
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ProvenanceError(f"{name} must be an integer, got {type(value).__name__}")
+    if value < 0:
+        raise ProvenanceError(f"{name} must be non-negative, got {value}")
+
+
+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.
     """
@@ -252,6 +298,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.
+    _check_seconds("max_future_skew_seconds", max_future_skew_seconds)
+    _check_seconds("max_age_seconds", max_age_seconds, optional=True)
+    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")

tests/test_provenance.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import base64
+import time
 
 import pytest
 
@@ -22,7 +23,7 @@
     tool_catalog_hash,
     verify_record,
 )
-from agentrust_trace.sign import _canonical_bytes, 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
@@ -410,3 +411,166 @@ def test_an_unusable_embedded_key_is_refused_rather_than_crashing() -> None:
     with pytest.raises(ProvenanceError, match="unusable"):
         verify_record(signed, 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))
+
+
+# --- policy inputs, per the review on #164 ----------------------------------
+#
+# The age bound is verifier configuration, and a malformed one fails in the
+# direction that matters: -1 is not a stricter bound, it calls every record
+# ever issued stale, uniformly, with no error naming the cause. `bool` gets its
+# own case because it is a subclass of `int`, so `True` would otherwise be
+# accepted as one second.
+
+
+@pytest.mark.parametrize("bad", [-1, -86400, True, False, 1.5, "300", object()])
+def test_a_malformed_max_age_is_reported_not_applied(bad) -> None:
+    key = generate_key()
+    signed = _signed_at(int(time.time()), key)
+    with pytest.raises(ProvenanceError) as exc:
+        verify_record(signed, key_to_jwk(key), max_age_seconds=bad)
+    assert "max_age_seconds" in str(exc.value)
+
+
+@pytest.mark.parametrize("bad", [-1, True, False, 1.5, "300", None])
+def test_a_malformed_skew_is_reported_not_applied(bad) -> None:
+    key = generate_key()
+    signed = _signed_at(int(time.time()), key)
+    with pytest.raises(ProvenanceError) as exc:
+        verify_record(signed, key_to_jwk(key), max_future_skew_seconds=bad)
+    assert "max_future_skew_seconds" in str(exc.value)
+
+
+@pytest.mark.parametrize("ok", [1, 86400])
+def test_a_well_formed_bound_still_verifies(ok: int) -> None:
+    key = generate_key()
+    verify_record(_signed_at(int(time.time()), key), key_to_jwk(key), max_age_seconds=ok)
+    verify_record(
+        _signed_at(int(time.time()), key), key_to_jwk(key), max_future_skew_seconds=ok
+    )
+
+
+def test_zero_is_a_bound_and_not_a_falsy_stand_in_for_unset() -> None:
+    """`0` and `None` are different policies and must not be conflated.
+
+    `None` disables the age bound; `0` is the strictest one expressible - the
+    record must be issued at this instant, so anything already in the past is
+    stale. A validator that treated `0` as falsy would silently accept every
+    record under the strictest policy a caller can write.
+    """
+    key = generate_key()
+    a_moment_ago = _signed_at(int(time.time()) - 5, key)
+    verify_record(a_moment_ago, key_to_jwk(key))  # None: no age bound
+    with pytest.raises(ProvenanceError, match="stale"):
+        verify_record(a_moment_ago, key_to_jwk(key), max_age_seconds=0)