diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index 648bacf4..aef0becf 100644 --- a/src/agentrust_trace/provenance.py +++ b/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") diff --git a/tests/test_provenance.py b/tests/test_provenance.py index 81e6f652..bbee335c 100644 --- a/tests/test_provenance.py +++ b/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)