From 610da6e40abc8e1e365109756f2502f1218500f9 Mon Sep 17 00:00:00 2001 From: harshnair75567-cloud Date: Sat, 5 Sep 2026 11:26:26 +0530 Subject: [PATCH] sign.verify_record: validate max_age_seconds/max_future_skew_seconds/max_bundle_age_seconds Signed-off-by: harshnair75567-cloud --- CHANGELOG.md | 1 + src/agentrust_trace/provenance.py | 27 +++--------- src/agentrust_trace/sign.py | 33 ++++++++++++-- tests/test_sign.py | 71 +++++++++++++++++++++++++++++++ 4 files changed, 107 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d4d1d8b..9a15069 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,6 +46,7 @@ Format: [Semantic Versioning](https://semver.org/). Spec versions follow `MAJOR. - **`provenance.verify_record()`, `provenance.check_tool_catalog()` and `content_marking.verify_assertion()` now hold their externally supplied argument to the type they document.** Each read that argument before establishing its shape. Measured across a twelve-value junk matrix, the two provenance functions leaked eleven `AttributeError`s apiece; the twelfth value is `{}`, which is an object and so reached the refusal each function documents, which is not the `ProvenanceError` `verify_record` documents. `content_marking.verify_assertion()` was worse than a crash rather than merely undocumented: it did not check that `record_bytes` were bytes, and `bytes(5)` is five zero bytes, so an int was hashed, failed to match, and the caller was told the record at the URL had changed, which is a specific and false accusation about somebody else's server. All three now raise the error their module documents, naming the type received. - **`jwk_thumbprint()` and `verify_record()` now refuse a non-object argument with the error they document.** Both read a member off the argument before establishing its shape, so a string, a number, `None`, a list or a bool raised `AttributeError`, which is not the `ValueError` `verify_record`'s docstring names for every rejection other than a bad signature, and is not caught by a caller written against that contract. Neither argument is one the caller has already established: a JWK reaches `jwk_thumbprint` from a peer, a key document or a record's own `cnf`, and the record handed to `verify_record` is by definition not yet known to be an object. Both now raise `ValueError` naming the type received. 21 tests: removing the two guards fails 20 of them, and the twenty-first is the control that has to keep passing. +- **`sign.verify_record()` now validates `max_age_seconds` and `max_future_skew_seconds` the same way `provenance.verify_record()` does.** `provenance.verify_record()` rejects a malformed freshness bound via `_check_seconds()`, added per the review on #164. `sign.verify_record()`, the original Trust Record verifier, never received the same hardening: it checked only `max_future_skew_seconds < 0`, and `max_age_seconds` was compared against unvalidated. Passing `max_age_seconds=-1` (a value a caller might use meaning "no bound," since `None` is the documented way to disable the check) rejected every record, including one issued the same second, as `record is stale`, naming the record rather than the misconfigured argument. `_check_seconds()` is now defined once in `sign.py` and shared: `sign.verify_record()` calls it directly, and `provenance.verify_record()` imports it, passing its own `ProvenanceError` via a new `exc` parameter so each keeps its existing public error type. - **`provenance.tool_catalog_hash()` now refuses a malformed `tools` list instead of crashing.** Reached primarily through `check_tool_catalog(record, tools)`, the function the module's own docstring calls "the step that catches a live attack," because `tools` there is what the MCP server actually returned, i.e. the untrusted party this check exists to catch. The function iterated `tools` and called `.get(...)` on each entry with no check that `tools` was a list or that its entries were objects, so a malformed response (an entry that is a string, `None`, a number, or `tools` itself not being a list) raised an unhandled `AttributeError` or `TypeError` instead of the documented `ProvenanceError`. A server that is misbehaving maliciously or just buggily and is exactly the source `tools` has no reason to trust its shape. Fixed with an explicit `isinstance` check on `tools` and on each of its entries, naming the offending index. diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index a548deb..c8a19e6 100644 --- a/src/agentrust_trace/provenance.py +++ b/src/agentrust_trace/provenance.py @@ -23,9 +23,10 @@ from agentrust_trace.sign import ( RevocationStore, _canonical_bytes, - anchor_bytes, _check_not_revoked, + _check_seconds, _pubkey_from_jwk, + anchor_bytes, jwk_thumbprint, key_to_jwk, ) @@ -265,24 +266,6 @@ def sign_record(record: dict[str, Any], key: Any) -> dict[str, Any]: return {**payload, "signature": sig} -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], @@ -358,8 +341,10 @@ def verify_record( # 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) + _check_seconds("max_future_skew_seconds", max_future_skew_seconds, exc=ProvenanceError) + _check_seconds( + "max_age_seconds", max_age_seconds, optional=True, exc=ProvenanceError + ) age = time.time() - int(record["issued_at"]) if age < -max_future_skew_seconds: raise ProvenanceError( diff --git a/src/agentrust_trace/sign.py b/src/agentrust_trace/sign.py index 8be2dd3..02e8819 100644 --- a/src/agentrust_trace/sign.py +++ b/src/agentrust_trace/sign.py @@ -357,6 +357,30 @@ def _b64url_decode(value: str, *, field: str) -> bytes: raise ValueError(f"{field} is not valid base64url: {exc}") from exc +def _check_seconds( + name: str, value: Any, *, optional: bool = False, exc: type[Exception] = ValueError +) -> None: + """Reject a malformed freshness-policy input instead of silently acting on it. + + A verifier's age/skew 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 (``None``) would see a uniform refusal with no 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. + + Shared by :func:`sign.verify_record` and :func:`provenance.verify_record`, + which pass their own exception type via *exc* so each keeps its existing + public error type (``ValueError`` and ``ProvenanceError`` respectively). + """ + if optional and value is None: + return + if isinstance(value, bool) or not isinstance(value, int): + raise exc(f"{name} must be an integer, got {type(value).__name__}") + if value < 0: + raise exc(f"{name} must be non-negative, got {value}") + + def sign_record(record: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any]: """Return a copy of *record* with ``cnf.jwk`` populated and a ``signature`` field added. @@ -526,10 +550,8 @@ def verify_record( raise ValueError("now must be an integer Unix timestamp in seconds, or None") else: verification_time = now - if max_bundle_age_seconds < 0: - raise ValueError("max_bundle_age_seconds must be non-negative") - if max_future_skew_seconds < 0: - raise ValueError("max_future_skew_seconds must be non-negative") + _check_seconds("max_bundle_age_seconds", max_bundle_age_seconds) + _check_seconds("max_future_skew_seconds", max_future_skew_seconds) from cryptography.exceptions import InvalidSignature as _InvalidSignature # noqa: F401 @@ -661,6 +683,9 @@ def verify_record( ) # Freshness: bound the age of the record against its issued-at timestamp. + # Both bounds are verifier configuration, not record data, and a malformed + # one is checked before it is used -- see `_check_seconds`. + _check_seconds("max_age_seconds", max_age_seconds, optional=True) iat = record.get("iat") if not isinstance(iat, int) or isinstance(iat, bool): raise ValueError("record has no valid integer 'iat' for freshness check") diff --git a/tests/test_sign.py b/tests/test_sign.py index fad6c94..2404b76 100644 --- a/tests/test_sign.py +++ b/tests/test_sign.py @@ -434,6 +434,77 @@ def test_verify_record_rejects_negative_future_skew_configuration(): verify_record(record, key_to_jwk(key), max_future_skew_seconds=-1) +# --- freshness policy inputs, shared with provenance.py via `_check_seconds` -- +# +# The age/skew bounds are 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. `provenance.verify_record` already +# guarded against this (per the review on #164); this is that same guard on +# the Trust Record side. + + +@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() + record = sign_record(_fresh_record(), key) + with pytest.raises(ValueError) as exc: + verify_record(record, key_to_jwk(key), max_age_seconds=bad) + assert "max_age_seconds must be" in str(exc.value) + + +@pytest.mark.parametrize("bad", [True, False, 1.5, "300", None]) +def test_a_malformed_skew_is_reported_not_applied(bad) -> None: + key = generate_key() + record = sign_record(_fresh_record(), key) + with pytest.raises(ValueError) as exc: + verify_record(record, key_to_jwk(key), max_future_skew_seconds=bad) + assert "max_future_skew_seconds must be" 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(sign_record(_fresh_record(), key), key_to_jwk(key), max_age_seconds=ok) + verify_record( + sign_record(_fresh_record(), key), key_to_jwk(key), max_future_skew_seconds=ok + ) + + +@pytest.mark.parametrize("bad", [-1, -86400, True, False, 1.5, "300", object(), None]) +def test_a_malformed_bundle_age_is_reported_not_applied(bad) -> None: + key = generate_key() + record = sign_record(_fresh_record(), key) + with pytest.raises(ValueError) as exc: + verify_record(record, key_to_jwk(key), max_bundle_age_seconds=bad) + assert "max_bundle_age_seconds must be" in str(exc.value) + + +@pytest.mark.parametrize("ok", [1, 86400]) +def test_a_well_formed_bundle_age_still_verifies(ok: int) -> None: + key = generate_key() + verify_record( + sign_record(_fresh_record(), key), key_to_jwk(key), max_bundle_age_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() + record = sign_record(_fresh_record(), key) + time.sleep(1.1) + verify_record(record, key_to_jwk(key), max_age_seconds=None) # disabled: passes + with pytest.raises(ValueError, match="stale"): + verify_record(record, key_to_jwk(key), max_age_seconds=0) + + def test_verify_record_rejects_non_okp_jwk(): key = generate_key() record = sign_record(_fresh_record(), key)