From 30f31d5ed9986da6f2e0e616c82146f144328cb6 Mon Sep 17 00:00:00 2001 From: LouieLuNZ <48041247+lywinged@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:42:37 +0000 Subject: [PATCH 1/2] test: sweep every keyword argument of every public function, and close what it found The documented-exception sweep covered the first positional argument of each public function and declared `provenance.build_record` and `intent_bridge.verify_bridge` unsweepable, with a true reason: their arguments are keyword-only. #320 then arrived on `build_record`'s `issued_at`, a keyword argument that `int()` coerced before the guard saw it. An exemption whose reason is true is still an exemption, and the leaks were on the other kind of argument. The sweep now varies every parameter of every public function from a call that is valid in full, one parameter at a time, against the same junk matrix. Three things hold it honest: every parameter is accounted for by name, so a new one fails until it is swept or declared; every baseline is asserted to succeed before anything is varied, so a leak is from the varied parameter and not from a broken baseline; and one witness per function proves the call arrives. Two further checks come from #320's shape: what a producer accepts its own verifier has to accept, and a bare string or single object where an iterable is expected is refused rather than iterated. On main the sweep found eight parameter sites leaking undocumented exceptions and five behaviours that produced a wrong result with no exception at all. Closed here, each in its module's own pattern: - `build_record(publisher=)` raised `TypeError` from the regex. - `build_record(attestation=False)` emitted `false` where the profile says `null`; truthiness stood where `is None` was meant. - `provenance.sign_record` let `rfc8785`'s errors and `key_to_jwk`'s `ValueError` escape; wrapped as `intent_bridge._jcs` already wraps. - `intent_bridge.sign_bridge(key=)` raised `AttributeError`. The guard is `isinstance(key, Ed25519PrivateKey)`, which the package's two other signers already require through `key_to_jwk`; a duck-typed key no longer signs a bridge. - `content_marking.build_assertion` raised `TypeError` for a non-string `alg` and dropped a non-string or empty `anchor` on the floor. - `revocation.check_bundle(trusted_key_identifiers="sha256:...")` iterated the string as characters, matched no statement, and reported `verified`; a single JWK where a list was meant iterated as field names. Both refused now, and their elements are typed. - `sign.verify_record(trusted_bundle_keys="")` read an empty string as no keys. The eighth site is `issued_at` itself, which is #320's. Its cases are marked as strict expected failures against that issue, so its fix is what turns them green and this file fails the day the marker goes stale. Controls: reverting each fixed module fails the sweep (2, 2, 5, 10 and 4 tests respectively). 1485 passed, 1 skipped, 2 xfailed on 3.11 and 3.12. No wire format, schema or signature semantics change; every input that verified before verifies now. Signed-off-by: Louie Lu <48041247+lywinged@users.noreply.github.com> Co-Authored-By: Claude Fable 5.1 --- src/agentrust_trace/content_marking.py | 16 + src/agentrust_trace/intent_bridge.py | 12 +- src/agentrust_trace/provenance.py | 44 +- src/agentrust_trace/revocation.py | 34 +- src/agentrust_trace/sign.py | 8 +- ...blic_functions_raise_what_they_document.py | 442 +++++++++++++++++- tools/sweep_public_surface.py | 105 +++++ 7 files changed, 642 insertions(+), 19 deletions(-) create mode 100644 tools/sweep_public_surface.py diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index 9e2e4b66..fbd9c990 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -116,6 +116,11 @@ def build_assertion( re-serialized dict is a hash of something nobody will ever fetch: key order, separators and escaping all change the bytes without changing the record, and the verifier hashes what the server sends. + + Raises ``ContentMarkingError`` for a *url* that is empty, an *alg* that is not a + digest algorithm name, and an *anchor* that is not a non-empty string. A non-string + or empty anchor used to be dropped without a word, so the caller got an assertion + with no anchor and no error. """ if not isinstance(record_bytes, bytes | bytearray) or not record_bytes: raise ContentMarkingError( @@ -125,6 +130,17 @@ def build_assertion( if not url: raise ContentMarkingError("url is required: an assertion with no reference binds nothing") url = _record_url(url) + if not isinstance(alg, str): + raise ContentMarkingError( + f"alg must be a digest algorithm name, got {type(alg).__name__}; use sha256 or sha384" + ) + if anchor is not None and (not isinstance(anchor, str) or not anchor): + # `if anchor:` below used to drop a non-string or empty anchor on the floor, so a + # caller who passed one got an assertion with no anchor and no error. Refuse it. + raise ContentMarkingError( + "anchor must be a non-empty registry entry URI string or None, got " + f"{anchor!r}" + ) import json diff --git a/src/agentrust_trace/intent_bridge.py b/src/agentrust_trace/intent_bridge.py index 3a7e8235..2d91c91e 100644 --- a/src/agentrust_trace/intent_bridge.py +++ b/src/agentrust_trace/intent_bridge.py @@ -65,7 +65,17 @@ def digest_jcs(value: dict[str, Any]) -> str: def sign_bridge(authorization: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any]: - """Sign the complete authorization; key material is deliberately not embedded.""" + """Sign the complete authorization; key material is deliberately not embedded. + + Raises ``IntentBridgeError`` for a *key* that is not an ``Ed25519PrivateKey``: the + bridge profile fixes the algorithm, and the package's two other signers hold their + key to the same type through ``key_to_jwk``. + """ + if not isinstance(key, Ed25519PrivateKey): + raise IntentBridgeError( + f"key must be an Ed25519PrivateKey, got {type(key).__name__}. The bridge " + "profile fixes the algorithm, so there is no other key this can sign with." + ) artifact = {"profile": BRIDGE_PROFILE, "authorization": authorization} signature = base64.urlsafe_b64encode(key.sign(_jcs(artifact, "the authorization"))).rstrip(b"=") return {**artifact, "signature": signature.decode("ascii")} diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index fdaff013..6cb1b40a 100644 --- a/src/agentrust_trace/provenance.py +++ b/src/agentrust_trace/provenance.py @@ -20,6 +20,8 @@ import time from typing import Any +import rfc8785 + from agentrust_trace.sign import ( RevocationStore, _b64url_decode, @@ -188,6 +190,12 @@ def _check_structure( "endpoint.spki_sha256 must be a sha256: digest of the Subject Public Key " "Info. A URL on its own is not an identity." ) + if attestation is not None and not isinstance(attestation, dict): + raise ProvenanceError( + f"attestation must be an object or null, got {type(attestation).__name__}. " + "spec/server-provenance-v1.md: the evidence in the shape TRACE v0.2 section 3.1 " + "runtime uses, or null." + ) if kind == "tee-attested" and not attestation: raise ProvenanceError( "kind='tee-attested' without attestation evidence is the claim without the " @@ -225,7 +233,7 @@ def build_record( """ if kind not in KINDS: raise ProvenanceError(f"kind {kind!r} is not one of {', '.join(KINDS)}") - if not _PUBLISHER_RE.match(publisher or ""): + if not isinstance(publisher, str) or not _PUBLISHER_RE.match(publisher): raise ProvenanceError( f"publisher {publisher!r} must be a DID or SPIFFE URI. A display name is not " "resolvable and a verifier cannot check one." @@ -272,13 +280,29 @@ def sign_record(record: dict[str, Any], key: Any) -> dict[str, Any]: Raises ``ProvenanceError`` for a *record* that is not a JSON object. ``{**record}`` reads it before its shape is established, so a non-mapping raised a bare ``TypeError`` about dict unpacking, which is not this module's documented refusal. + Also raises ``ProvenanceError`` for a *key* that is not an Ed25519 private key, and + for a record with no RFC 8785 canonical form, such as an integer outside the JCS + safe range; both used to escape as the underlying library's ``ValueError``. """ if not isinstance(record, dict): raise ProvenanceError( f"record must be a JSON object, got {type(record).__name__}" ) - payload = {**record, "cnf": {"jwk": key_to_jwk(key)}} - body = _canonical_bytes({k: v for k, v in payload.items() if k != "signature"}) + try: + jwk = key_to_jwk(key) + except ValueError as exc: + raise ProvenanceError(f"key must be an Ed25519 private key: {exc}") from exc + payload = {**record, "cnf": {"jwk": jwk}} + try: + body = _canonical_bytes({k: v for k, v in payload.items() if k != "signature"}) + except rfc8785.CanonicalizationError as exc: + # `_canonical_bytes` is `rfc8785.dumps` and raises its own errors for a value JCS + # has no form for, including an integer outside the safe domain. Those are + # `ValueError`s, not this module's, so a caller catching `ProvenanceError` saw a + # crash. Same shape as the wrap `intent_bridge._jcs` already carries. + raise ProvenanceError( + f"record has no RFC 8785 canonical form, so it cannot be signed: {exc}" + ) from exc import base64 sig = base64.urlsafe_b64encode(key.sign(body)).rstrip(b"=").decode() @@ -410,7 +434,19 @@ def verify_record( ) pub = _pubkey_from_jwk(trusted_jwk) - body = _canonical_bytes({k: v for k, v in record.items() if k != "signature"}) + try: + body = _canonical_bytes({k: v for k, v in record.items() if k != "signature"}) + except rfc8785.CanonicalizationError as exc: + # The other half of the wrap `sign_record` carries. The record here is the + # untrusted document, so it can hold a value JCS has no form for wherever the + # structural checks above do not type the field: an integer outside the safe + # range under `tools`, say. `rfc8785`'s errors are its own `ValueError`s, not + # this module's, so a caller written against `ProvenanceError` saw a crash + # where every other malformed record gives a refusal. + raise ProvenanceError( + f"record has no RFC 8785 canonical form, so its signature cannot be " + f"checked: {exc}" + ) from exc # `signature` reaches this function from whatever the caller is verifying, the # same untrusted document nothing above this line has vouched for either: a # non-string here (an int, a list of chars, a nested object) previously hit diff --git a/src/agentrust_trace/revocation.py b/src/agentrust_trace/revocation.py index 8566d1fb..cfa42050 100644 --- a/src/agentrust_trace/revocation.py +++ b/src/agentrust_trace/revocation.py @@ -181,6 +181,30 @@ def _trusted_bundle_key( return None +def _sequence_of(value: Any, name: str, element: type) -> list[Any]: + """Materialise a caller-supplied iterable, refusing the shapes that iterate wrongly. + + A ``str`` iterates as characters, so ``trusted_key_identifiers="sha256:..."`` used to + become a list of one-character identifiers that matched no statement, and the check + reported ``verified`` for a key it never looked up. A ``dict`` iterates as its keys, + so a single JWK passed where a list of them was meant became a list of field names. + Neither is an iterable of *element* and both are refused here, with the documented + error, rather than turned into a result. + """ + if isinstance(value, (str, bytes, bytearray, dict)) or not hasattr(value, "__iter__"): + raise ValueError( + f"{name} must be an iterable of {element.__name__} values, got " + f"{type(value).__name__}" + ) + items = list(value) + bad = [type(v).__name__ for v in items if not isinstance(v, element)] + if bad: + raise ValueError( + f"{name} must contain only {element.__name__} values, found {sorted(set(bad))}" + ) + return items + + def check_bundle( bundle: dict[str, Any], *, @@ -204,13 +228,17 @@ def check_bundle( Raises ``ValueError`` when a statement on the bundle's log names the trusted key. That is evidence failing rather than evidence absent, and it fails closed - like the ``revocation`` store does. + like the ``revocation`` store does. Also raises ``ValueError`` for the caller's + own arguments when they are not what they say: ``trusted_key_identifiers`` must + be an iterable of strings and ``trusted_bundle_keys`` an iterable of JWK + objects, and a bare string or a single object is refused rather than iterated + as characters or field names. """ _check_seconds("now", now) _check_seconds("max_bundle_age_seconds", max_bundle_age_seconds) _check_seconds("max_future_skew_seconds", max_future_skew_seconds) - trusted_ids = list(trusted_key_identifiers) - trusted_bundle_keys = list(trusted_bundle_keys) + trusted_ids = _sequence_of(trusted_key_identifiers, "trusted_key_identifiers", str) + trusted_bundle_keys = _sequence_of(trusted_bundle_keys, "trusted_bundle_keys", dict) # 3a. Shape, against the packaged schema pair. The first error by path, so the # evidence points at one place rather than listing the file. diff --git a/src/agentrust_trace/sign.py b/src/agentrust_trace/sign.py index 02e88197..c6f25151 100644 --- a/src/agentrust_trace/sign.py +++ b/src/agentrust_trace/sign.py @@ -509,7 +509,11 @@ def verify_record( signed by a key not in ``trusted_bundle_keys``, signed with an algorithm this build cannot verify, dated in the future, or expired under either bound yields ``unverified_for_revocation`` with the cause named; it does - not raise, because inability to check is not evidence of a defect. A + not raise, because inability to check is not evidence of a defect. The + argument itself is held to its shape: ``trusted_bundle_keys`` must be an + iterable of JWK objects or ``None``, and a string, a single object or + another non-iterable raises ``ValueError`` rather than being read as no + keys. A statement on the bundle's log naming the trusted key raises ``ValueError``: no inclusion entry ID reaches this function, so 3.2.3's fallback applies and every record the key signed is rejected. @@ -652,7 +656,7 @@ def verify_record( revocation_check = check_bundle( revocation_bundle, trusted_key_identifiers=_key_identifiers(trusted_jwk), - trusted_bundle_keys=trusted_bundle_keys or (), + trusted_bundle_keys=() if trusted_bundle_keys is None else trusted_bundle_keys, now=verification_time, max_bundle_age_seconds=max_bundle_age_seconds, max_future_skew_seconds=max_future_skew_seconds, diff --git a/tests/test_public_functions_raise_what_they_document.py b/tests/test_public_functions_raise_what_they_document.py index f09a41f0..877ad4a2 100644 --- a/tests/test_public_functions_raise_what_they_document.py +++ b/tests/test_public_functions_raise_what_they_document.py @@ -14,19 +14,37 @@ refuses with is the failure this catches; adding one and quietly not sweeping it is the failure that produced this file. -What it does not do: sweep every argument of every function. It sweeps the first -positional argument, holding the rest valid, which is where externally-supplied data -arrives. ``CALLS`` is written out per function rather than generated, because a generated -call passes ``None`` for the arguments it does not vary, and a ``TypeError`` from the -second argument then reads exactly like a leak in the first. That happened while this file -was being written, and produced a finding against ``verify_bridge`` that did not exist. +``CALLS`` sweeps the first positional argument, holding the rest valid, which is where +externally-supplied data arrives. It is written out per function rather than generated, +because a generated call passes ``None`` for the arguments it does not vary, and a +``TypeError`` from the second argument then reads exactly like a leak in the first. That +happened while this file was being written, and produced a finding against +``verify_bridge`` that did not exist. + +``KEYWORD_CALLS`` sweeps every other parameter, one at a time, from a call that is valid +in full. It exists because the first version of this file said "what it does not do: +sweep every argument", listed ``build_record`` and ``verify_bridge`` as having no +positional argument to sweep, and #320 then arrived on ``build_record``'s ``issued_at``, +a keyword argument that ``int()`` coerced before the guard saw it. An exemption with a +true reason is still an exemption: "no positional argument" was correct and the leaks +were on the other kind. Every parameter of every public function is now either swept +here or named in ``UNSWEPT_PARAMETERS`` with the reason, and the coverage test holds the +three sets equal to the signatures, so a new parameter fails until somebody decides. + +The assertions decide what must be refused. What each parameter *accepts* is a judgment, +so ``tools/sweep_public_surface.py`` prints it from these same tables as a report for a +reader; the last two tests hold that the report still runs over every function here and +that its ``--strict`` exit goes red the moment a leak is not filed. """ from __future__ import annotations import importlib +import importlib.util import inspect import json +import pathlib import pkgutil +import sys import time from collections.abc import Callable from typing import Any @@ -102,10 +120,120 @@ #: Functions with no externally-supplied positional argument to sweep. Listed so that #: the coverage test below can account for the whole surface rather than for the part -#: somebody remembered. +#: somebody remembered. `build_record` and `verify_bridge` were here until #320; their +#: arguments are keyword arguments and are swept below. NO_ARGUMENT_TO_SWEEP = { "sign.generate_key", "sign.load_signing_key", - "intent_bridge.verify_bridge", # every argument is keyword-only and required +} + +_RECORD_JSON = json.dumps({ + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", "iat": 1760000000, + "subject": "spiffe://example.org/agent/image-bot", "data_class": "public", +}).encode() +_ASSERTION = content_marking.build_assertion(_RECORD_JSON, url="https://r.example/r.json") +_DECLARATION = {"impact": "external-side-effect", "purpose": "send invoice"} +_TOOL_CALL = {"name": "send_invoice", "arguments": {"approved": 1}} +_BRIDGE_AUTH = { + "authorization_id": "auth-1", "decision": "allow", "authorizer": "finance-policy", + "authorizer_key_id": "key-1", "authorized_at": 100, "expires_at": 200, + "scope": {"tools": ["send_invoice"], "impacts": ["external-side-effect"]}, + "pic": {"profile": "PIC-CJSON/1.0", "intent_digest": "sha256:" + "1" * 64, + "args_digest": "sha256:" + "2" * 64}, + "declaration_digest": intent_bridge.digest_jcs(_DECLARATION), + "tool_call_digest": intent_bridge.digest_jcs(_TOOL_CALL), + "transcript_required": True, +} +_BRIDGE = intent_bridge.sign_bridge(_BRIDGE_AUTH, _KEY) +_TRANSCRIPT = {"before": {"tool_call": dict(_TOOL_CALL)}, "after": {"status": "accepted"}} +_TOOLS = [{"name": "search", "description": "search", "input_schema": {"type": "object"}}] +_ARTIFACT = {"package": "pkg:npm/%40acme/mcp-search@2.1.0", "digest": "sha256:" + "0" * 64} +_PROVENANCE = provenance.build_record( + kind="publisher-asserted", publisher="did:web:acme.example", tools=_TOOLS, + artifact=_ARTIFACT, issued_at=1760000000, +) +_PROVENANCE_SIGNED = provenance.sign_record(_PROVENANCE, _KEY) +_VECTOR = json.loads( + (pathlib.Path(__file__).resolve().parents[1] / "examples" / "revocation-bundle" + / "01-fresh-well-inside-both-bounds.json").read_text(encoding="utf-8") +) +_CTX = _VECTOR["context"] +_TRUST_RECORD = _VECTOR["records"][0] + +#: name -> (a complete valid call as keyword arguments, the parameters to vary). Each +#: baseline is asserted to succeed before anything is varied, so a leak reported here +#: is from the varied parameter and not from a baseline that was already broken. +KEYWORD_CALLS: dict[str, tuple[Callable[[], dict[str, Any]], tuple[str, ...]]] = { + "content_marking.build_assertion": ( + lambda: {"record_bytes": _RECORD_JSON, "url": "https://r.example/r.json", + "alg": "sha256", "anchor": None}, + ("url", "alg", "anchor"), + ), + "content_marking.verify_assertion": ( + lambda: {"assertion": _ASSERTION, "record_bytes": _RECORD_JSON}, ("record_bytes",), + ), + "intent_bridge.sign_bridge": ( + lambda: {"authorization": _BRIDGE_AUTH, "key": _KEY}, ("key",), + ), + "intent_bridge.verify_bridge": ( + lambda: {"bridge": _BRIDGE, "trusted_authorizer_jwk": {**_JWK, "kid": "key-1"}, + "declaration": _DECLARATION, + "pic_intent_digest": _BRIDGE_AUTH["pic"]["intent_digest"], + "pic_args_digest": _BRIDGE_AUTH["pic"]["args_digest"], + "tool_call": _TOOL_CALL, "transcript": _TRANSCRIPT, "now": 150}, + ("bridge", "trusted_authorizer_jwk", "declaration", "pic_intent_digest", + "pic_args_digest", "tool_call", "transcript", "now"), + ), + "provenance.build_record": ( + lambda: {"kind": "publisher-asserted", "publisher": "did:web:acme.example", + "tools": _TOOLS, "artifact": _ARTIFACT, "endpoint": None, "attestation": None, + "issued_at": 1760000000}, + ("kind", "publisher", "tools", "artifact", "endpoint", "attestation", "issued_at"), + ), + "provenance.check_tool_catalog": ( + lambda: {"record": _PROVENANCE, "tools": _TOOLS}, ("tools",), + ), + "provenance.sign_record": (lambda: {"record": _PROVENANCE, "key": _KEY}, ("key",)), + "provenance.verify_record": ( + lambda: {"record": _PROVENANCE_SIGNED, "trusted_jwk": _JWK, "revocation": None, + "max_age_seconds": None, "max_future_skew_seconds": 300}, + ("trusted_jwk", "revocation", "max_age_seconds", "max_future_skew_seconds"), + ), + "revocation.check_bundle": ( + lambda: {"bundle": _CTX["bundle"], + "trusted_key_identifiers": [sign.jwk_thumbprint(_CTX["trusted_key"])], + "trusted_bundle_keys": _CTX["trusted_bundle_keys"], "now": _CTX["now"], + "max_bundle_age_seconds": _CTX["max_bundle_age_seconds"], + "max_future_skew_seconds": _CTX["max_future_skew_seconds"]}, + ("trusted_key_identifiers", "trusted_bundle_keys", "now", + "max_bundle_age_seconds", "max_future_skew_seconds"), + ), + "sign.sign_record": ( + lambda: {"record": {k: v for k, v in _TRUST_RECORD.items() if k != "signature"}, + "key": _KEY}, + ("key",), + ), + "sign.verify_record": ( + lambda: {"record": _TRUST_RECORD, "public_key_or_jwk": _CTX["trusted_key"], + "allow_embedded_key": False, "max_age_seconds": None, + "max_future_skew_seconds": 300, "expected_nonce": None, "revocation": None, + "revocation_bundle": _CTX["bundle"], + "trusted_bundle_keys": _CTX["trusted_bundle_keys"], + "max_bundle_age_seconds": _CTX["max_bundle_age_seconds"], + "now": _CTX["now"]}, + ("public_key_or_jwk", "allow_embedded_key", "max_age_seconds", + "max_future_skew_seconds", "expected_nonce", "revocation", "revocation_bundle", + "trusted_bundle_keys", "max_bundle_age_seconds", "now"), + ), +} + +#: Parameters swept by neither table, each with the reason. Empty is the goal; a +#: reason that stops being true is what the coverage test is for. +UNSWEPT_PARAMETERS: dict[str, str] = {} + +#: (function, parameter) pairs whose leaks are known, filed, and owned by someone +#: else's fix. Strict: the day the fix lands, the entry has to go, or this fails. +LEAKS_FILED: dict[tuple[str, str], str] = { + ("provenance.build_record", "issued_at"): "#320", } @@ -137,13 +265,258 @@ def test_every_public_function_is_either_swept_or_declared_unsweepable() -> None """The coverage test. A new public function fails here until somebody decides which it is, which is the step that was skipped last time.""" found = set(_public_functions()) - accounted = set(CALLS) | NO_ARGUMENT_TO_SWEEP + accounted = set(CALLS) | set(KEYWORD_CALLS) | NO_ARGUMENT_TO_SWEEP assert found == accounted, ( f"not swept and not declared unsweepable: {sorted(found - accounted)}\n" f"declared but no longer present: {sorted(accounted - found)}" ) +def _module_of(name: str) -> Any: + return {"content_marking": content_marking, "intent_bridge": intent_bridge, + "provenance": provenance, "revocation": revocation, "sign": sign, + "validate": validate}[name.split(".")[0]] + + +def test_every_parameter_of_every_public_function_is_swept_or_named() -> None: + """The coverage test for the other kind of argument. #320 sat on a keyword argument of + a function this file had declared unsweepable, with a true reason. So the accounting + is now per parameter: the first positional through `CALLS`, the rest through + `KEYWORD_CALLS`, and anything else named in `UNSWEPT_PARAMETERS` with why.""" + missing: dict[str, list[str]] = {} + stale: list[str] = [] + for name, func in _public_functions().items(): + if name in NO_ARGUMENT_TO_SWEEP: + continue + params = list(inspect.signature(func).parameters) + covered = set(params[:1]) if name in CALLS else set() + covered |= set(KEYWORD_CALLS.get(name, (None, ()))[1]) + covered |= {p.split(".")[-1] for p in UNSWEPT_PARAMETERS if p.startswith(name + ".")} + left = [p for p in params if p not in covered] + if left: + missing[name] = left + for p in KEYWORD_CALLS.get(name, (None, ()))[1]: + if p not in params: + stale.append(f"{name}.{p}") + assert not missing, f"parameters swept by nothing and named by nobody: {missing}" + assert not stale, f"KEYWORD_CALLS names parameters that no longer exist: {stale}" + + +@pytest.mark.parametrize("name", sorted(KEYWORD_CALLS)) +def test_every_keyword_baseline_succeeds_before_anything_is_varied(name: str) -> None: + base, _ = KEYWORD_CALLS[name] + func = getattr(_module_of(name), name.split(".")[1]) + func(**base()) + + +def _keyword_cases() -> list[Any]: + cases = [] + for name, (_, params) in sorted(KEYWORD_CALLS.items()): + for param in params: + marks = [] + if (name, param) in LEAKS_FILED: + marks.append(pytest.mark.xfail( + strict=True, reason=f"filed as {LEAKS_FILED[(name, param)]}")) + cases.append(pytest.param(name, param, id=f"{name}.{param}", marks=marks)) + return cases + + +@pytest.mark.parametrize(("name", "param"), _keyword_cases()) +def test_no_keyword_argument_leaks_an_undocumented_exception(name: str, param: str) -> None: + allowed = DOCUMENTED[name.split(".")[0]] + base, _ = KEYWORD_CALLS[name] + func = getattr(_module_of(name), name.split(".")[1]) + leaked: dict[str, Any] = {} + for value in JUNK: + kwargs = base() + kwargs[param] = value + try: + func(**kwargs) + except Exception as exc: # noqa: BLE001 - the whole point is what escapes + if type(exc).__name__ not in allowed: + leaked.setdefault(type(exc).__name__, repr(value)[:20]) + assert not leaked, ( + f"{name}({param}=...) raised {leaked}, which its module does not document as its " + f"refusal. Documented: {allowed}." + ) + + +#: One (parameter, value) per keyword-swept function that must reach the function and +#: come back as the documented refusal, so a clean sweep above means the call arrived. +KEYWORD_REACHES: dict[str, tuple[str, Any, str]] = { + "content_marking.build_assertion": ("alg", 123, "ContentMarkingError"), + "content_marking.verify_assertion": ("record_bytes", None, "ContentMarkingError"), + "intent_bridge.sign_bridge": ("key", None, "IntentBridgeError"), + "intent_bridge.verify_bridge": ("now", "a-string", "IntentBridgeError"), + "provenance.build_record": ("publisher", 123, "ProvenanceError"), + "provenance.check_tool_catalog": ("tools", None, "ProvenanceError"), + "provenance.sign_record": ("key", None, "ProvenanceError"), + "provenance.verify_record": ("max_age_seconds", "a-string", "ProvenanceError"), + "revocation.check_bundle": ("now", "a-string", "ValueError"), + "sign.sign_record": ("key", None, "ValueError"), + "sign.verify_record": ("max_age_seconds", "a-string", "ValueError"), +} + + +def test_every_keyword_swept_function_has_a_witness() -> None: + assert set(KEYWORD_REACHES) == set(KEYWORD_CALLS) + + +@pytest.mark.parametrize("name", sorted(KEYWORD_REACHES)) +def test_the_keyword_sweep_actually_reaches_each_function(name: str) -> None: + param, value, expected = KEYWORD_REACHES[name] + base, _ = KEYWORD_CALLS[name] + kwargs = base() + kwargs[param] = value + func = getattr(_module_of(name), name.split(".")[1]) + with pytest.raises(Exception) as caught: # noqa: PT011 - the type is the assertion + func(**kwargs) + assert type(caught.value).__name__ == expected + + +#: A producer that accepts a value has to emit something its own verifier accepts. This +#: is the half of #320 the leak test cannot see: `build_record(issued_at=10**20)` raised +#: nothing, and `sign_record` then refused the record it built. +PRODUCERS: dict[str, tuple[str, Callable[[dict[str, Any]], Any]]] = { + "provenance.build_record": ( + "provenance.build_record", + lambda built: provenance.verify_record( + provenance.sign_record(built, _KEY), _JWK, max_age_seconds=None), + ), + "content_marking.build_assertion": ( + "content_marking.build_assertion", + lambda built: content_marking.verify_assertion(built, _RECORD_JSON), + ), +} + + +def _producer_cases() -> list[Any]: + cases = [] + for name in sorted(PRODUCERS): + for param in KEYWORD_CALLS[name][1]: + marks = [] + if (name, param) in LEAKS_FILED: + marks.append(pytest.mark.xfail( + strict=True, reason=f"filed as {LEAKS_FILED[(name, param)]}")) + cases.append(pytest.param(name, param, id=f"{name}.{param}", marks=marks)) + return cases + + +@pytest.mark.parametrize(("name", "param"), _producer_cases()) +def test_what_a_producer_accepts_its_own_verifier_accepts(name: str, param: str) -> None: + base, _ = KEYWORD_CALLS[name] + func = getattr(_module_of(name), name.split(".")[1]) + _, verify = PRODUCERS[name] + orphaned: dict[str, str] = {} + for value in JUNK: + kwargs = base() + kwargs[param] = value + try: + built = func(**kwargs) + except Exception: # noqa: BLE001 - refusing is the leak test's business + continue + try: + verify(built) + except Exception as exc: # noqa: BLE001 + orphaned[repr(value)[:20]] = type(exc).__name__ + assert not orphaned, ( + f"{name}({param}=...) accepted values whose output its own verifier refuses: " + f"{orphaned}. A producer emitting what its signer or verifier will not take is " + f"the producer reporting success for a record nobody can use." + ) + + +@pytest.mark.parametrize("anchor", ["", 123, [1, 2]], ids=["empty", "int", "list"]) +def test_a_wrong_anchor_is_refused_rather_than_dropped(anchor: Any) -> None: + """`if anchor:` used to be the only gate, so a non-string was written out as-is and an + empty string was dropped on the floor: the caller got an assertion with no anchor and + no error. None of these is a registry entry URI and each is refused, with the error + naming the parameter. The empty string matters separately because it passes an + `isinstance` check and would still fall to `if anchor:` without its own clause.""" + with pytest.raises(content_marking.ContentMarkingError, match="anchor"): + content_marking.build_assertion(_RECORD_JSON, url="https://r.example/r.json", anchor=anchor) + + +@pytest.mark.parametrize("attestation", [False, 0, ""], ids=["False", "0", "empty"]) +def test_a_falsy_attestation_is_refused_rather_than_emitted(attestation: Any) -> None: + """`kind != "tee-attested" and attestation` is a truthiness test, so on `main` a + falsy non-`None` value passed it and `build_record` emitted `"attestation": false` + where spec/server-provenance-v1.md gives `null`. The guard is `is not None` plus the + type, and it raises the module's own error.""" + kwargs = KEYWORD_CALLS["provenance.build_record"][0]() + kwargs["attestation"] = attestation + with pytest.raises(provenance.ProvenanceError, match="attestation"): + provenance.build_record(**kwargs) + + +def test_a_record_carrying_attestation_false_is_refused_by_the_verifier() -> None: + """The one verification-side change in the sweep. The same shape check runs on both + sides, so a record already carrying `"attestation": false`, which `main`'s + `build_record` could emit and `main`'s `verify_record` accepted, is refused now. The + control is the same record with `null`, which verifies.""" + record = dict(_PROVENANCE) + record["attestation"] = False + with pytest.raises(provenance.ProvenanceError, match="attestation"): + provenance.verify_record(provenance.sign_record(record, _KEY), _JWK, max_age_seconds=None) + record["attestation"] = None + provenance.verify_record(provenance.sign_record(record, _KEY), _JWK, max_age_seconds=None) + + +def test_sign_record_refuses_a_record_it_cannot_canonicalise_with_its_own_error() -> None: + """`_canonical_bytes` is `rfc8785.dumps`, whose errors are its own `ValueError`s. An + integer past the safe domain reached it through a record `build_record` had accepted, + and left `sign_record` as `IntegerDomainError`. Same wrap `intent_bridge._jcs` has.""" + with pytest.raises(provenance.ProvenanceError, match="canonical form"): + provenance.sign_record({**_PROVENANCE, "issued_at": 10**20}, _KEY) + + +def test_verify_record_refuses_a_record_it_cannot_canonicalise_with_its_own_error() -> None: + """The verifier half of the same wrap. `sign_record` is a producer and can refuse + early, but the record reaching `verify_record` is the untrusted document, so a value + JCS has no form for arrives wherever the structural checks do not type the field. + An out-of-range integer under `tools` left this function as `rfc8785`'s own + `IntegerDomainError`. The control is the same edit with a safe integer: it reaches + the signature check and is refused there, which is what says the first refusal came + from canonicalisation rather than from the edit itself.""" + signed = provenance.sign_record(_PROVENANCE, _KEY) + unsignable = {**signed, "tools": [{"name": "t", "version": 10**20}]} + with pytest.raises(provenance.ProvenanceError, match="canonical form"): + provenance.verify_record(unsignable, _JWK, max_age_seconds=None) + safe = {**signed, "tools": [{"name": "t", "version": 1}]} + with pytest.raises(provenance.ProvenanceError, match="signature does not verify"): + provenance.verify_record(safe, _JWK, max_age_seconds=None) + + +@pytest.mark.parametrize("bare", ["sha256:" + "a" * 64, b"bytes", {"kty": "OKP"}], + ids=["str", "bytes", "dict"]) +def test_a_bare_value_where_an_iterable_is_expected_is_refused(bare: Any) -> None: + """A `str` iterates as characters and a `dict` as its keys. `check_bundle` with + `trusted_key_identifiers="sha256:..."` used to look up one-character identifiers, + match nothing, and report `verified`. That is a result, and it is the wrong one.""" + base, _ = KEYWORD_CALLS["revocation.check_bundle"] + for param in ("trusted_key_identifiers", "trusted_bundle_keys"): + kwargs = base() + kwargs[param] = bare + with pytest.raises(ValueError, match=param): + revocation.check_bundle(**kwargs) + kwargs = KEYWORD_CALLS["sign.verify_record"][0]() + kwargs["trusted_bundle_keys"] = bare + with pytest.raises(ValueError, match="trusted_bundle_keys"): + sign.verify_record(**kwargs) + + +@pytest.mark.parametrize("falsy", ["", {}, False, 0], ids=["str", "dict", "False", "0"]) +def test_a_falsy_non_none_trusted_bundle_keys_is_not_read_as_no_keys(falsy: Any) -> None: + """`verify_record` used to pass `trusted_bundle_keys or ()` down, so an empty string, + an empty object or `False` became "no trusted keys" and the bundle check reported + the bundle key untrusted. Only `None` means no keys; the rest reach the check and + are refused as the wrong shape.""" + kwargs = KEYWORD_CALLS["sign.verify_record"][0]() + kwargs["trusted_bundle_keys"] = falsy + with pytest.raises(ValueError, match="trusted_bundle_keys"): + sign.verify_record(**kwargs) + + def test_every_swept_module_declares_what_it_refuses_with() -> None: modules = {name.split(".")[0] for name in CALLS} assert modules <= set(DOCUMENTED), f"undeclared: {sorted(modules - set(DOCUMENTED))}" @@ -229,3 +602,54 @@ def test_iter_errors_reports_rather_than_raising() -> None: exception. A non-record returning no findings would be a caller accepting junk.""" assert validate.iter_errors("a-string"), "iter_errors reported nothing for a string" assert validate.iter_errors({}), "iter_errors reported nothing for an empty object" + + +# --- The report tool ----------------------------------------------------------------- + +_TOOL = pathlib.Path(__file__).resolve().parents[1] / "tools" / "sweep_public_surface.py" + + +def _load_tool() -> Any: + spec = importlib.util.spec_from_file_location("sweep_public_surface", _TOOL) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_the_report_tool_runs_over_the_same_tables( + capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + """The tool reads this file's tables rather than keeping its own, so it cannot drift + from the assertions. This holds that it still loads them, reaches every function in + both tables, and exits 0 while every known leak is filed.""" + tool = _load_tool() + monkeypatch.setattr(tool, "_load_tables", lambda: sys.modules[__name__]) + assert tool.main(["--strict"]) == 0 + out = capsys.readouterr().out + for name in list(CALLS) + list(KEYWORD_CALLS): + assert name in out, f"{name} is missing from the report" + assert ", 0 unfiled leak(s)" in out + + +def test_the_report_tool_goes_red_when_a_leak_is_not_filed( + capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + """The control for the exit code. Plant a function that raises what its module does + not document, with nothing filed for it, and ``--strict`` has to fail, or a green run + says nothing. The leak is planted rather than borrowed from ``LEAKS_FILED``, so this + control still fires on the day the last real leak is fixed and that table is empty. + Without ``--strict`` the report stays informational.""" + + def leaks(value: Any) -> None: + raise RuntimeError("planted: not a ProvenanceError") + + tool = _load_tool() + monkeypatch.setattr(tool, "_load_tables", lambda: sys.modules[__name__]) + monkeypatch.setitem(CALLS, "provenance.planted_control", leaks) + assert tool.main(["--strict"]) == 1 + out = capsys.readouterr().out + assert "provenance.planted_control" in out + assert "RuntimeError" in out + assert ", 0 unfiled leak(s)" not in out + assert tool.main([]) == 0 diff --git a/tools/sweep_public_surface.py b/tools/sweep_public_surface.py new file mode 100644 index 00000000..17eef08e --- /dev/null +++ b/tools/sweep_public_surface.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Report what every public function does with junk, one parameter at a time. + + python tools/sweep_public_surface.py # the report + python tools/sweep_public_surface.py --strict # exit 1 on any unfiled leak + +The assertions live in ``tests/test_public_functions_raise_what_they_document.py`` and +run in CI: no undocumented exception from any parameter, what a producer accepts its own +verifier accepts, no bare value where an iterable is expected. This script reuses that +file's tables (``JUNK``, ``CALLS``, ``KEYWORD_CALLS``, ``LEAKS_FILED``) and prints the +part the assertions cannot decide for you: which junk each parameter *accepts*. A +``now`` of ``0`` or a ``max_age_seconds`` of ``10**20`` is accepted on purpose; an +``issued_at`` of ``True`` is accepted by accident, and that is #320. The list is for a +reader to look down, not for CI to pass, which is why it is a script and not a test. + +Run it from a checkout with ``requirements/dev.txt`` installed before a release, or when +a public function gains a parameter, and read the ACCEPTED column with the function's +contract open. +""" +from __future__ import annotations + +import importlib.util +import pathlib +import sys +from collections import defaultdict +from typing import Any + +ROOT = pathlib.Path(__file__).resolve().parents[1] +TEST_FILE = ROOT / "tests" / "test_public_functions_raise_what_they_document.py" +# The tree first, so the report is about this checkout and not an installed release. +sys.path.insert(0, str(ROOT / "src")) + + +def _load_tables() -> Any: + spec = importlib.util.spec_from_file_location("sweep_tables", TEST_FILE) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules["sweep_tables"] = module + spec.loader.exec_module(module) + return module + + +def _outcome(call: Any, allowed: tuple[str, ...]) -> tuple[str, str]: + try: + call() + except Exception as exc: # noqa: BLE001 - classifying what escapes is the job + kind = type(exc).__name__ + return ("documented" if kind in allowed else "LEAK", kind) + return ("accepted", "") + + +def main(argv: list[str]) -> int: + strict = "--strict" in argv + t = _load_tables() + rows: list[tuple[str, str, str, list[str], dict[str, str]]] = [] + unfiled_leaks = 0 + for name in sorted(t.CALLS): + allowed = t.DOCUMENTED[name.split(".")[0]] + accepted: list[str] = [] + leaks: dict[str, str] = {} + for value in t.JUNK: + call = t.CALLS[name] + state, kind = _outcome(lambda v=value, c=call: c(v), allowed) + if state == "accepted": + accepted.append(repr(value)[:14]) + elif state == "LEAK": + leaks.setdefault(kind, repr(value)[:14]) + rows.append((name, "", "", accepted, leaks)) + for name, (base, params) in sorted(t.KEYWORD_CALLS.items()): + allowed = t.DOCUMENTED[name.split(".")[0]] + func = getattr(t._module_of(name), name.split(".")[1]) + for param in params: + accepted = [] + leaks = {} + for value in t.JUNK: + kwargs = base() + kwargs[param] = value + state, kind = _outcome(lambda k=kwargs, f=func: f(**k), allowed) + if state == "accepted": + accepted.append(repr(value)[:14]) + elif state == "LEAK": + leaks.setdefault(kind, repr(value)[:14]) + filed = t.LEAKS_FILED.get((name, param), "") + rows.append((name, param, filed, accepted, leaks)) + width = max(len(r[0]) + len(r[1]) + 1 for r in rows) + print(f"{'function.parameter':<{width}} {'LEAKS':<28} ACCEPTED") + print("-" * (width + 40)) + by_function: dict[str, int] = defaultdict(int) + for name, param, filed, accepted, leaks in rows: + label = f"{name}.{param}" if param != "" else f"{name} {param}" + leak_text = ", ".join(f"{k} on {v}" for k, v in leaks.items()) or "none" + if leaks and filed: + leak_text += f" (filed {filed})" + elif leaks: + unfiled_leaks += len(leaks) + by_function[name] += len(accepted) + print(f"{label:<{width}} {leak_text:<28} {', '.join(accepted) or 'none'}") + print() + print(f"{len(rows)} parameters swept, {sum(len(r[3]) for r in rows)} accepted junk values, " + f"{unfiled_leaks} unfiled leak(s)") + return 1 if strict and unfiled_leaks else 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) From 5ec62d53b692e314fa7026af50ed3587a9858731 Mon Sep 17 00:00:00 2001 From: LouieLuNZ <48041247+lywinged@users.noreply.github.com> Date: Sat, 12 Sep 2026 06:37:20 +0000 Subject: [PATCH 2/2] Close the other half of #320: the issued_at bound the sweep was still reporting Rebased onto `c04d938`. #334 merged the reordering half of #320 and closed the issue, and the strict marker here did exactly what it is for: on the rebase, `test_no_keyword_argument_leaks_an_undocumented_exception[provenance.build_record.issued_at]` went `XPASS(strict)` and the file failed rather than going quietly green. The second case filed under the same entry did not flip. `test_what_a_producer_accepts_its_own_verifier_accepts` asks whether a producer emits anything its own verifier refuses, and for `issued_at` it still reported one value, `10000000000000000000`. That is the half #334 did not carry: `_check_structure` had no upper bound, so `build_record` accepted a timestamp `sign_record` cannot canonicalise and the caller met `rfc8785`'s `IntegerDomainError` rather than the `ProvenanceError` this module documents. Proposed on #320 on 09-10 and carried on #323, which is now superseded on everything else. `_check_structure` gains the bound, reading `JCS_SAFE_INTEGER` from `sign.py` rather than writing the number a third time, and `LEAKS_FILED` is empty. The comment above it records what the entry held and why it went, so the marker's two firings are not lost with it. `_check_structure` is shared, so this reaches `verify_record` too. Under the default freshness policy a record carrying such a timestamp was already refused as dated in the future, so no record that verified before is refused now. What changes is the caller who widens `max_future_skew_seconds` past the gap: that returned `rfc8785.IntegerDomainError` out of a function documented to raise `ProvenanceError`, and now returns `ProvenanceError`. A test pins it. The boundary is asserted against the canonicalizer rather than against a literal: each of the four cases checks `build_record` and `rfc8785.dumps` agree about whether the value exists, so the guard cannot drift away from the thing it is guarding. Verified in both directions. With the clause: 1547 passed, 1 skipped, and the sweep at 152 passed with nothing xfailed. With only the clause reverted: 4 fail, being the two out-of-range boundary cases, the verifier case, and the sweep's own producer-and-verifier case, which is the instrument that reported the gap now confirming it is closed. `ruff`, `mypy` and `tools/check_dashes.py` pass. No `CHANGELOG.md` hunk, per the instruction on 09-11. Signed-off-by: LouieLuNZ <48041247+lywinged@users.noreply.github.com> --- src/agentrust_trace/provenance.py | 19 +++++-- tests/test_provenance.py | 55 ++++++++++++++++++- ...blic_functions_raise_what_they_document.py | 17 +++++- 3 files changed, 83 insertions(+), 8 deletions(-) diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index 6cb1b40a..6824d352 100644 --- a/src/agentrust_trace/provenance.py +++ b/src/agentrust_trace/provenance.py @@ -23,6 +23,7 @@ import rfc8785 from agentrust_trace.sign import ( + JCS_SAFE_INTEGER, RevocationStore, _b64url_decode, _canonical_bytes, @@ -206,11 +207,21 @@ def _check_structure( f"kind={kind!r} carries attestation evidence. Evidence that is present but " "not claimed invites a consumer to read it as an attestation that was made." ) - # bool is an int subclass, and True would otherwise pass as a timestamp. - if not isinstance(issued_at, int) or isinstance(issued_at, bool) or issued_at < 0: + # bool is an int subclass, and True would otherwise pass as a timestamp. The upper + # bound is the same JCS safe-integer limit #219 applies to every other signed integer + # in this package: above it there is no portable canonical form, so the producer would + # accept a timestamp it cannot sign and the caller would meet `rfc8785`'s + # `IntegerDomainError` instead of the class this module documents. + if ( + not isinstance(issued_at, int) + or isinstance(issued_at, bool) + or issued_at < 0 + or issued_at > JCS_SAFE_INTEGER + ): raise ProvenanceError( - "issued_at must be a non-negative integer Unix timestamp. A record with no " - "issue time cannot be aged, so a consumer has no way to reject a stale one." + "issued_at must be a non-negative integer Unix timestamp within the JCS " + "safe-integer range. A record with no usable issue time cannot be aged, so " + "a consumer has no way to reject a stale one." ) diff --git a/tests/test_provenance.py b/tests/test_provenance.py index 991fe4f1..b0efb87f 100644 --- a/tests/test_provenance.py +++ b/tests/test_provenance.py @@ -12,6 +12,7 @@ import time import pytest +import rfc8785 from agentrust_trace.provenance import ( FORMAT, @@ -23,7 +24,13 @@ tool_catalog_hash, verify_record, ) -from agentrust_trace.sign import _canonical_bytes, generate_key, jwk_thumbprint, key_to_jwk +from agentrust_trace.sign import ( + JCS_SAFE_INTEGER, + _canonical_bytes, + generate_key, + jwk_thumbprint, + key_to_jwk, +) DIGEST = "sha256:" + "a" * 64 OTHER_DIGEST = "sha256:" + "b" * 64 @@ -797,6 +804,52 @@ def test_an_unconvertible_issued_at_raises_what_the_module_documents( _record(issued_at=supplied) +# The other half of #320, which #334 did not carry: the guard had no upper bound, so a +# producer accepted a timestamp it could not sign. The value is a well-formed non-negative +# integer, so none of the tests above reaches it; `int(2**60)` is `2**60`. +@pytest.mark.parametrize( + ("supplied", "accepted"), + [ + (JCS_SAFE_INTEGER - 1, True), + (JCS_SAFE_INTEGER, True), + (JCS_SAFE_INTEGER + 1, False), + (2**60, False), + ], + ids=["below", "at", "above", "far-above"], +) +def test_the_issued_at_bound_sits_where_the_canonicalizer_stops( + supplied: int, accepted: bool +) -> None: + """The boundary is asserted against `rfc8785` rather than against a number written + twice: the guard and the canonicalizer have to agree about which integers exist, or + `build_record` emits records `sign_record` refuses.""" + if accepted: + assert _record(issued_at=supplied)["issued_at"] == supplied + rfc8785.dumps({"issued_at": supplied}) + return + with pytest.raises(ProvenanceError, match="issued_at"): + _record(issued_at=supplied) + with pytest.raises(rfc8785.IntegerDomainError): + rfc8785.dumps({"issued_at": supplied}) + + +def test_an_out_of_range_issued_at_no_longer_leaves_the_verifier_as_rfc8785s_error() -> None: + """`_check_structure` is shared, so the bound reaches `verify_record` as well. + + Under the default freshness policy such a record was already refused, as dated in the + future, so nothing that verified before is refused now. The case that changes is a + caller who widens `max_future_skew_seconds` past the gap: the structural check ran + with the value in hand, the canonicalizer met it first, and `rfc8785`'s + `IntegerDomainError` left a function documented to raise `ProvenanceError`. + """ + key = generate_key() + record = _record() + record["issued_at"] = 2**60 + signed = dict(record, signature="AA", cnf={"jwk": key_to_jwk(key)}) + with pytest.raises(ProvenanceError, match="issued_at"): + verify_record(signed, key_to_jwk(key), max_future_skew_seconds=10**19) + + def test_a_valid_issued_at_is_still_carried_through_unchanged() -> None: assert _record(issued_at=1_754_000_000)["issued_at"] == 1_754_000_000 diff --git a/tests/test_public_functions_raise_what_they_document.py b/tests/test_public_functions_raise_what_they_document.py index 877ad4a2..3fdbaea7 100644 --- a/tests/test_public_functions_raise_what_they_document.py +++ b/tests/test_public_functions_raise_what_they_document.py @@ -95,6 +95,12 @@ # #320: every argument is keyword-only, which is why this was listed as # unsweepable and why a coercion in front of the validator went unnoticed. # Only issued_at varies; the rest are valid so a refusal can only come from it. + # + # `KEYWORD_CALLS` below also sweeps this function, and over every parameter rather + # than this one. The overlap is deliberate and is left for the maintainer to + # collapse or keep: this entry arrived with #334 and deleting a test that landed + # hours ago, inside a pull request about something else, is not this branch's call. + # Keeping both costs one duplicated `issued_at` sweep and no coverage. "provenance.build_record": lambda v: provenance.build_record( kind="publisher-asserted", publisher="did:web:example.com", tools=[], artifact={"package": "x", "digest": "sha256:" + "a" * 64}, issued_at=v, @@ -232,9 +238,14 @@ #: (function, parameter) pairs whose leaks are known, filed, and owned by someone #: else's fix. Strict: the day the fix lands, the entry has to go, or this fails. -LEAKS_FILED: dict[tuple[str, str], str] = { - ("provenance.build_record", "issued_at"): "#320", -} +#: It held `("provenance.build_record", "issued_at"): "#320"` until 2026-09-12, and the +#: marker did its job twice. #334 landed the reordering half and the leak case went +#: `XPASS(strict)` on the rebase, which is what took the entry off the exception class. +#: The producer-and-verifier case did not flip, because the other half of #320 is the +#: safe-integer bound and #334 did not carry it; the one value that case still reported +#: was `10000000000000000000`. That bound is now in `_check_structure`, so both cases +#: pass and nothing is filed. +LEAKS_FILED: dict[tuple[str, str], str] = {} def _public_functions() -> dict[str, Any]: