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..6824d352 100644 --- a/src/agentrust_trace/provenance.py +++ b/src/agentrust_trace/provenance.py @@ -20,7 +20,10 @@ import time from typing import Any +import rfc8785 + from agentrust_trace.sign import ( + JCS_SAFE_INTEGER, RevocationStore, _b64url_decode, _canonical_bytes, @@ -188,6 +191,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 " @@ -198,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." ) @@ -225,7 +244,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 +291,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 +445,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_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 f09a41f0..3fdbaea7 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 @@ -77,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, @@ -102,12 +126,127 @@ #: 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. +#: 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]: """Walk the package. Discovered rather than listed: a hardcoded roster is how the @@ -137,13 +276,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 +613,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:]))