diff --git a/.gitignore b/.gitignore index 9bb51e7e..dc6fcc33 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,22 @@ htmlcov/ .venv/ venv/ uv.lock + +# What the documentation tells a reader to write into their working tree. +# +# docs/quickstart.md writes an unencrypted Ed25519 private key to trace-key.pem, under +# a comment reading "keep secure, never commit or log". The comment was the whole of the +# enforcement: a reader following the quickstart inside a clone was one `git add -A` +# away from committing their signing key. tests/test_the_docs_do_not_leave_a_key_behind.py +# recovers these paths from the documentation, so a doc that starts writing somewhere +# new fails rather than quietly widening the gap. +session.trace.json +trace-key.pem +trace-key.pem.pub + +# No key material of any kind, whatever it is called. The repository tracks none today +# and there is no case for the first one arriving unnoticed. +*.pem +*.key +*.p8 +*.pfx diff --git a/CHANGELOG.md b/CHANGELOG.md index cf6c0f36..daeddc1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,28 @@ Format: [Semantic Versioning](https://semver.org/). Spec versions follow `MAJOR. ### Fixed +- **The exported `SCHEMA` was the live object the validator reads.** `_schema()` is `lru_cache`d and `_validator()` is built over whatever it returns, so the name exposed "for downstream tooling that needs the raw dict" and the validator's schema were one object: lowering `SCHEMA["properties"]["iat"]["minimum"]` made a record dated 1970 valid to `validate_json()`, to `iter_errors()`, and to the structural gate inside `sign.verify_record()`, for every later call in the process. Nothing about the call site looks wrong, since adapting the raw dict is the use the comment invites. It is a deep copy now; a shallow one would leave the nested `properties` dicts shared and the same edit would still land. + +- **All eight `"format": "uri"` declarations in the schema were inert, and `"not a uri"` validated.** `jsonschema` treats `format` as an annotation rather than an assertion unless a checker for that format is installed, and `FormatChecker().checkers` does not carry `uri` without an optional dependency the project did not declare. The wiring in `validate.py` was correct and the behaviour was a no-op, which is the shape that survives review. `rfc3986-validator` is now a dependency, chosen over `jsonschema[format]` because that pulls `rfc3987`, which is GPLv3, into an Apache-2.0 package's dependency tree. Turning the constraint on changes nothing about the existing corpus: the suite passed unchanged before any new test was added. + +- **`TrustRecord.model_validate(record).model_dump()` returned a record this package's own validator rejects and whose signature no longer verifies.** Pydantic serializes every unset optional as an explicit `null`, the schema types no named field as nullable, and the added members change the RFC 8785 canonical bytes the signature is taken over. This is the round trip `sign_record()`'s own docstring points a caller at, so a caller who validated the model and then wrote it out wrote a broken record, with neither check running at the moment the damage is done. Absent optionals are omitted now and the round trip is exact identity. Only declared fields are dropped: `JWK` sets `extra="allow"` and the schema permits a null among those members, so a null inside `cnf.jwk` is data. + +- **The models accepted booleans where JSON says integer, and read them as numbers.** `isinstance(True, int)` is a Python fact and not a JSON one: JSON Schema's `"type": "integer"` does not match `true`, so `schema/trace-claim.json` rejects `{"slsa_level": true}` and `models.BuildProvenance` accepted it and coerced it to `1`, making the record a claim of SLSA build level 1 assembled out of a boolean. `tool_transcript.call_count` did the same, and `appraisal.timestamp` read `true` as 1 January 1970. `iat` and `origin.ingested_at` did not have the hole and were safe by accident rather than by design, their lower bound sitting above 1 so the coerced value failed the range check afterwards; all five carry an explicit guard now. Found by mutating every field of a valid record and comparing the two validators, which had never been compared. The differential is committed, and the disagreements it does not fix are declared with the reason. + +- **Six public functions raised exceptions no module documents, and the sweep that finds them is committed.** `key_to_jwk()` and `load_key()` read `.public_key()` and `.encode()` off their argument before establishing its type; `key_to_jwk` now names the public-key case separately, because that is the plausible mistake for a function whose name reads as "turn a key into a JWK" and whose result is the public JWK. `sign_record()` in both modules unpacked `{**record}` before checking it was a mapping. `anchor_bytes()` refused the two value classes registry-anchor-v1 section 1 excludes and handed everything else to `json.dumps`, so a type JSON cannot serialize came back as a message about a serializer. `intent_bridge` let `rfc8785` errors out as themselves: those are `ValueError` subclasses, which satisfies `sign`'s contract but not this module's, since a `CanonicalizationError` is not an `IntentBridgeError`. `verify_bridge()` did not leak but misattributed, canonicalizing inside the `try` that reports the signature invalid. The sweep walks the package rather than listing functions, and a coverage test fails until every discovered function is either swept or declared unsweepable. + +- **`docs/quickstart.md` writes an unencrypted private key into the reader's working tree and `.gitignore` did not cover it.** The block writes `trace-key.pem` under the comment "keep secure, never commit or log", and that comment was the whole of the enforcement: a reader following the quickstart inside a clone was one `git add -A` away from committing their signing key. The three paths the documentation writes are ignored now, along with `*.pem`, `*.key`, `*.p8` and `*.pfx`, since the repository tracks no key material today. The accompanying test recovers the written paths from the documentation rather than listing them, so a doc that starts writing somewhere new fails rather than widening the gap quietly. + +- **The revocation check no longer fails open when the store answers with a non-bool.** `RevocationStore` is `Container[str] | Callable[[str], bool]`, and the callable's return value was read by truthiness, so `None`, `""`, `0` and `[]` all read as "not revoked" and let the key through, while the string `"no"` read as revoked. `None` is the case that matters: it is what a CRL, status or SCITT lookup returns when its author handled the 200 and forgot every other response, which is exactly the outage the existing `except` clause was written to survive. That clause already treats a store that raises as a rejection, on the stated grounds that an unavailable source is not evidence a key is unrevoked; a store answering `None` supplied no more evidence and was being believed. `provenance.verify_record()` imports the same function and makes the same claim in its own docstring, so one fix closes both entry points. The membership branch is untouched, because `in` yields a real bool whatever `__contains__` returns. + +- **`provenance.verify_record()`, `provenance.check_tool_catalog()` and `content_marking.verify_assertion()` now hold their externally supplied argument to the type they document.** Each read that argument before establishing its shape. Measured across a twelve-value junk matrix, the two provenance functions leaked eleven `AttributeError`s apiece; the twelfth value is `{}`, which is an object and so reached the refusal each function documents, which is not the `ProvenanceError` `verify_record` documents. `content_marking.verify_assertion()` was worse than a crash rather than merely undocumented: it did not check that `record_bytes` were bytes, and `bytes(5)` is five zero bytes, so an int was hashed, failed to match, and the caller was told the record at the URL had changed, which is a specific and false accusation about somebody else's server. All three now raise the error their module documents, naming the type received. + +### Fixed + +- **`jwk_thumbprint()` and `verify_record()` now refuse a non-object argument with the error they document.** Both read a member off the argument before establishing its shape, so a string, a number, `None`, a list or a bool raised `AttributeError`, which is not the `ValueError` `verify_record`'s docstring names for every rejection other than a bad signature, and is not caught by a caller written against that contract. Neither argument is one the caller has already established: a JWK reaches `jwk_thumbprint` from a peer, a key document or a record's own `cnf`, and the record handed to `verify_record` is by definition not yet known to be an object. Both now raise `ValueError` naming the type received. 21 tests: removing the two guards fails 20 of them, and the twenty-first is the control that has to keep passing. + +### Fixed + - **`provenance.tool_catalog_hash()` now refuses a malformed `tools` list instead of crashing.** Reached primarily through `check_tool_catalog(record, tools)`, the function the module's own docstring calls "the step that catches a live attack," because `tools` there is what the MCP server actually returned, i.e. the untrusted party this check exists to catch. The function iterated `tools` and called `.get(...)` on each entry with no check that `tools` was a list or that its entries were objects, so a malformed response (an entry that is a string, `None`, a number, or `tools` itself not being a list) raised an unhandled `AttributeError` or `TypeError` instead of the documented `ProvenanceError`. A server that is misbehaving maliciously or just buggily and is exactly the source `tools` has no reason to trust its shape. Fixed with an explicit `isinstance` check on `tools` and on each of its entries, naming the offending index. - **`content_marking.build_assertion()` and `content_marking.verify_assertion()` now refuse record bytes that aren't a JSON object, instead of crashing.** Both functions called `.get(...)` on the result of `json.loads(record_bytes)` without checking it was a dict first. Valid JSON is not always an object that is an array, a string, a number, `null`, and a bool are all valid top-level JSON and `record_bytes` is exactly the kind of externally-sourced input this is likely to happen to: `build_assertion()` takes whatever bytes a caller hands it, and `verify_assertion()`'s docstring is explicit that its `record_bytes` are "the record bytes actually retrieved from its URL," i.e. a network response the caller does not control. Either function raised an unhandled `AttributeError` instead of the documented `ContentMarkingError`. `verify_assertion()` had a second, related gap: its second parse of `record_bytes` (after the hash check) was not wrapped in the try/except its first parse-adjacent check uses, so genuinely malformed (non-JSON) bytes that happened to hash-match raised `json.JSONDecodeError` instead of `ContentMarkingError` too. Both functions now check `isinstance(record, dict)` after parsing, and `verify_assertion()`'s second parse now catches `ValueError` the same way its hash-computation path already implicitly required valid bytes to reach. diff --git a/pyproject.toml b/pyproject.toml index 3285d37e..d52e8042 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,13 @@ classifiers = [ dependencies = [ "pydantic>=2.0", "jsonschema>=4.20", + # `jsonschema` treats `format` as an annotation unless a checker for that format is + # installed. Without this, all eight `"format": "uri"` declarations in + # schema/trace-claim.json are inert: `validate_json` builds a `FormatChecker`, the + # code reads as though URIs are checked, and "not a uri" validates. The non-GPL + # validator is used deliberately: `jsonschema[format]` pulls `rfc3987`, which is + # GPLv3, into an Apache-2.0 package's dependency tree. + "rfc3986-validator>=0.1.1", "cryptography>=42.0", "rfc8785>=0.1.2", ] diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index 18926221..a8aa89ee 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -150,6 +150,16 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str if not _DIGEST_RE.match(str(expected or "")): raise ContentMarkingError(f"record.hash {expected!r} is not a sha256:/sha384: digest") + if not isinstance(record_bytes, bytes | bytearray) or not record_bytes: + raise ContentMarkingError( + f"record_bytes must be the bytes retrieved from {ref['url']}, got " + f"{type(record_bytes).__name__}. `build_assertion` already refuses this and " + "the reason it matters more here is that `bytes(5)` is five zero bytes: an " + "int would be hashed, would not match, and the caller would be told the " + "record at the URL had changed, which is a specific accusation about " + "somebody else's server and would be false." + ) + actual = _digest(bytes(record_bytes), alg) if actual != expected: raise RecordMismatch( diff --git a/src/agentrust_trace/intent_bridge.py b/src/agentrust_trace/intent_bridge.py index 755f76eb..7d3b748a 100644 --- a/src/agentrust_trace/intent_bridge.py +++ b/src/agentrust_trace/intent_bridge.py @@ -8,6 +8,7 @@ from hmac import compare_digest from typing import Any +import rfc8785 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from agentrust_trace.sign import _b64url_decode, _canonical_bytes, _pubkey_from_jwk @@ -33,17 +34,40 @@ class AuthorizationMismatch(IntentBridgeError): """Execution evidence does not match the signed authorization.""" +def _jcs(value: dict[str, Any], what: str) -> bytes: + """Canonical bytes for *value*, as ``IntentBridgeError`` when there are none. + + ``_canonical_bytes`` is ``rfc8785.dumps`` and raises by design: a value JCS has + no form for has no canonical bytes to return. Its errors are ``ValueError`` + subclasses, which satisfies ``sign``'s documented contract but not this + module's: ``rfc8785.CanonicalizationError`` is not an ``IntentBridgeError``, so + a caller written against this module's own exception does not catch it. + + Four of the five values that trip it are ordinary JSON that ``json.loads`` + accepts, an integer outside the JCS safe range, a non-finite float, and a lone + surrogate among them, so an authorization assembled from a parsed document + reaches this. + """ + try: + return _canonical_bytes(value) + except rfc8785.CanonicalizationError as exc: + raise IntentBridgeError( + f"{what} has no RFC 8785 canonical form, so it cannot be digested or " + f"signed: {exc}" + ) from exc + + def digest_jcs(value: dict[str, Any]) -> str: """Return the SHA-256 digest of an RFC 8785 canonical JSON object.""" if not isinstance(value, dict): raise IntentBridgeError("a digest input must be a JSON object") - return f"sha256:{hashlib.sha256(_canonical_bytes(value)).hexdigest()}" + return f"sha256:{hashlib.sha256(_jcs(value, 'the digest input')).hexdigest()}" def sign_bridge(authorization: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any]: """Sign the complete authorization; key material is deliberately not embedded.""" artifact = {"profile": BRIDGE_PROFILE, "authorization": authorization} - signature = base64.urlsafe_b64encode(key.sign(_canonical_bytes(artifact))).rstrip(b"=") + signature = base64.urlsafe_b64encode(key.sign(_jcs(artifact, "the authorization"))).rstrip(b"=") return {**artifact, "signature": signature.decode("ascii")} @@ -114,10 +138,14 @@ def verify_bridge( for field in ("authorization_id", "authorizer", "authorizer_key_id"): _nonempty_string(authorization[field], f"authorization.{field}") + # Hoisted out of the try below. Inside it, an authorization JCS cannot serialize + # was reported as "the signature is invalid", which is a different fact and sends + # the reader to look at a key. There are no bytes for a signature to be checked + # against, so nothing has been learned about the signature at all. + body = _jcs({"profile": BRIDGE_PROFILE, "authorization": authorization}, + "the authorization") try: - _pubkey_from_jwk(trusted_authorizer_jwk).verify( - signature, _canonical_bytes({"profile": BRIDGE_PROFILE, "authorization": authorization}) - ) + _pubkey_from_jwk(trusted_authorizer_jwk).verify(signature, body) except Exception as exc: raise IntentBridgeError("authorization signature is invalid") from exc trusted_kid = trusted_authorizer_jwk.get("kid") diff --git a/src/agentrust_trace/models.py b/src/agentrust_trace/models.py index 650802b5..4fc0a7d9 100644 --- a/src/agentrust_trace/models.py +++ b/src/agentrust_trace/models.py @@ -1,8 +1,16 @@ from __future__ import annotations -from typing import Annotated, Literal - -from pydantic import BaseModel, ConfigDict, Field, model_validator +from typing import Annotated, Any, Literal + +from pydantic import ( + BaseModel, + BeforeValidator, + ConfigDict, + Field, + SerializerFunctionWrapHandler, + model_serializer, + model_validator, +) _DIGEST_RE = r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$" # ISO 8601 duration, spelled out by alternation rather than with a negative @@ -21,10 +29,73 @@ # downstream to whichever canonicalizer the producer happens to be using. JCS_SAFE_INTEGER = 9007199254740991 + +def _not_a_boolean(value: Any) -> Any: + """Reject ``True`` and ``False`` where JSON says integer. + + ``isinstance(True, int)`` is a Python fact and not a JSON one. JSON Schema's + ``"type": "integer"`` does not match a boolean, so ``schema/trace-claim.json`` + rejects ``{"slsa_level": true}`` and these models accepted it, coercing it to + ``1``. The record was then a claim of SLSA build level 1, assembled out of a + boolean, that no other implementation would have validated. + + The two fields that did not have this hole, ``iat`` and ``origin.ingested_at``, + were safe by accident rather than by design: their lower bound is above 1, so + the coerced value failed the range check afterwards. ``appraisal.timestamp`` + allows 1 and turned ``true`` into 1 January 1970. + """ + if isinstance(value, bool): + raise ValueError( + "expected an integer, got a boolean. JSON Schema type 'integer' does " + "not match true or false, so a record carrying one is rejected by " + "schema/trace-claim.json and by any implementation validating against it." + ) + return value + + +#: An integer as JSON means it, rather than as Python's type hierarchy means it. +JsonInt = Annotated[int, BeforeValidator(_not_a_boolean)] + DigestStr = Annotated[str, Field(pattern=_DIGEST_RE)] -class ModelInfo(BaseModel): +class _TraceModel(BaseModel): + """A model whose serialization is a TRACE record, not a Python object dump. + + Pydantic writes every unset optional as an explicit ``null``. The schema permits + ``null`` for no named field, so ``TrustRecord.model_validate(record).model_dump()`` + produced a record that `validate_json` rejects with "None is not of type 'string'", + and whose signature no longer verifies, because the added members change the RFC 8785 + canonical bytes the signature is taken over. + + That is the round trip ``sign_record``'s own docstring points a caller at: pass the + record to ``TrustRecord.model_validate()`` to confirm structural validity before + writing. A caller who then wrote the model out wrote a broken record, and neither the + validator nor the signature check runs at the moment the damage is done. + + Absent optional members are therefore omitted rather than nulled. + + Only *declared* fields are dropped. `JWK` sets ``extra="allow"`` and the schema's + ``canonicalizableValue`` permits a null there, so ``cnf.jwk`` may legitimately carry + one as data. A first version of this filtered the whole serialized dict and removed + it, which is the same defect this exists to fix, moved one level down: the round trip + stopped being identity and the signature stopped verifying, for a record the schema + accepts. An extra member keeps whatever value it was given. + """ + + @model_serializer(mode="wrap") + def _omit_absent_optionals( + self, handler: SerializerFunctionWrapHandler + ) -> dict[str, Any]: + extras = self.__pydantic_extra__ or {} + return { + key: value + for key, value in handler(self).items() + if value is not None or key in extras + } + + +class ModelInfo(_TraceModel): model_config = ConfigDict(extra="forbid") provider: str @@ -34,7 +105,7 @@ class ModelInfo(BaseModel): aibom_uri: str | None = None -class RuntimeInfo(BaseModel): +class RuntimeInfo(_TraceModel): model_config = ConfigDict(extra="forbid") platform: Literal[ @@ -63,7 +134,7 @@ class RuntimeInfo(BaseModel): firmware_version: str | None = None -class PolicyInfo(BaseModel): +class PolicyInfo(_TraceModel): model_config = ConfigDict(extra="forbid") bundle_hash: DigestStr @@ -92,15 +163,15 @@ class PolicyInfo(BaseModel): policy_uri: str | None = None -class ToolTranscript(BaseModel): +class ToolTranscript(_TraceModel): model_config = ConfigDict(extra="forbid") hash: DigestStr - call_count: Annotated[int, Field(ge=0, le=JCS_SAFE_INTEGER)] | None = None + call_count: Annotated[JsonInt, Field(ge=0, le=JCS_SAFE_INTEGER)] | None = None transcript_uri: str | None = None -class Delegation(BaseModel): +class Delegation(_TraceModel): """A2A profile: links this record to the record of the delegating hop. Present when this execution acted on authority delegated by another agent. @@ -116,7 +187,7 @@ class Delegation(BaseModel): credential_id: Annotated[str, Field(min_length=1)] -class Origin(BaseModel): +class Origin(_TraceModel): """Where the evidence in this record came from, when that is not this runtime. Absent means the record was produced by the runtime whose execution it @@ -152,10 +223,10 @@ class Origin(BaseModel): kind: Literal["self", "third-party-control-plane", "log-import"] producer: Annotated[str, Field(min_length=1)] source_event_id: Annotated[str, Field(min_length=1)] | None = None - ingested_at: Annotated[int, Field(ge=1700000000, le=JCS_SAFE_INTEGER)] | None = None + ingested_at: Annotated[JsonInt, Field(ge=1700000000, le=JCS_SAFE_INTEGER)] | None = None -class Reference(BaseModel): +class Reference(_TraceModel): """A fact outside this record that the record points at. Spec section 3.1.2. ``origin`` records where evidence *came from* and can lower assurance. @@ -197,10 +268,10 @@ class Reference(BaseModel): digest: DigestStr | None = None -class BuildProvenance(BaseModel): +class BuildProvenance(_TraceModel): model_config = ConfigDict(extra="forbid") - slsa_level: Annotated[int, Field(ge=0, le=3)] + slsa_level: Annotated[JsonInt, Field(ge=0, le=3)] builder: str | None = None digest: DigestStr provenance_uri: str | None = None @@ -209,13 +280,13 @@ class BuildProvenance(BaseModel): provenance_depth: Literal["surface", "builder", "transitive"] | None = None -class Appraisal(BaseModel): +class Appraisal(_TraceModel): model_config = ConfigDict(extra="forbid") status: Literal["affirming", "warning", "contraindicated", "none"] verifier: str policy_ref: str | None = None - timestamp: Annotated[int, Field(ge=-JCS_SAFE_INTEGER, le=JCS_SAFE_INTEGER)] | None = None + timestamp: Annotated[JsonInt, Field(ge=-JCS_SAFE_INTEGER, le=JCS_SAFE_INTEGER)] | None = None # What this verifier ran, not what the issuer claimed. provenance_depth_verified: Literal["surface", "builder", "transitive"] | None = None @@ -225,7 +296,7 @@ class Appraisal(BaseModel): _JWK_PRIVATE_PARAMS = frozenset({"d", "p", "q", "dp", "dq", "qi", "k"}) -class JWK(BaseModel): +class JWK(_TraceModel): # JWK params vary by key type (EC, OKP, RSA): allow unknown members per RFC 7517 model_config = ConfigDict(extra="allow") @@ -256,19 +327,19 @@ def _require_key_material(self) -> JWK: return self -class ConfirmationKey(BaseModel): +class ConfirmationKey(_TraceModel): model_config = ConfigDict(extra="forbid") jwk: JWK -class TrustRecord(BaseModel): +class TrustRecord(_TraceModel): """TRACE v0.2 Trust Record: hardware-attested governance evidence for an AI agent execution.""" model_config = ConfigDict(extra="forbid") eat_profile: Literal["tag:agentrust-io.com,2026:trace-v0.2"] - iat: Annotated[int, Field(ge=1700000000, le=JCS_SAFE_INTEGER)] + iat: Annotated[JsonInt, Field(ge=1700000000, le=JCS_SAFE_INTEGER)] subject: Annotated[str, Field(pattern=r"^(spiffe://[^/]+/.+|did:[a-z0-9]+:.+)$")] model: ModelInfo runtime: RuntimeInfo diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index 57530c0f..61348bc8 100644 --- a/src/agentrust_trace/provenance.py +++ b/src/agentrust_trace/provenance.py @@ -239,7 +239,16 @@ def build_record( def sign_record(record: dict[str, Any], key: Any) -> dict[str, Any]: - """Sign per TRACE v0.2 §3.2: Ed25519 over the JCS form with the signature absent.""" + """Sign per TRACE v0.2 §3.2: Ed25519 over the JCS form with the signature absent. + + 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. + """ + 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"}) import base64 @@ -304,6 +313,15 @@ def verify_record( """ import base64 + if not isinstance(record, dict): + raise ProvenanceError( + f"record must be a JSON object, got {type(record).__name__}. `_as_object` " + "holds `identity` and `tool_catalog` to that shape, and neither can be " + "reached until the record itself is one: `record.get(...)` on a list or a " + "string raises AttributeError, which is not the ProvenanceError this " + "function documents and is not caught by a caller written against it." + ) + if record.get("format") != FORMAT: raise ProvenanceError( f"unknown format {record.get('format')!r}; expected {FORMAT}. An unknown " @@ -403,6 +421,13 @@ def check_tool_catalog(record: dict[str, Any], tools: list[dict[str, Any]]) -> N contract :func:`verify_record` makes, since this can run against a record ``verify_record`` has not (yet) seen. """ + if not isinstance(record, dict): + raise ProvenanceError( + f"record must be a JSON object, got {type(record).__name__}. This runs " + "against records `verify_record` has not seen, as its docstring says, so it " + "cannot assume that function established the shape." + ) + actual = tool_catalog_hash(tools) catalog = _as_object(record.get("tool_catalog"), "tool_catalog") expected = catalog.get("hash") diff --git a/src/agentrust_trace/sign.py b/src/agentrust_trace/sign.py index f214cd71..903b9775 100644 --- a/src/agentrust_trace/sign.py +++ b/src/agentrust_trace/sign.py @@ -53,8 +53,26 @@ def generate_key() -> Ed25519PrivateKey: def load_key(pem: str) -> Ed25519PrivateKey: - """Load an Ed25519 private key from a PEM string.""" - return serialization.load_pem_private_key(pem.encode(), password=None) # type: ignore[return-value] + """Load an Ed25519 private key from a PEM string. + + Raises ``ValueError`` for anything that is not a PEM string this library can + read. A PEM arrives from a file, an environment variable or a secret store, + so its type is not something the caller has already established: reading + ``.encode()`` off it first turned every non-string into an ``AttributeError`` + and a lone surrogate into a ``UnicodeEncodeError``, neither of which a caller + written against this signature catches. + """ + if not isinstance(pem, str): + raise ValueError( + f"pem must be a PEM string, got {type(pem).__name__}. A key read from a " + "file, an environment variable or a secret store can be bytes or None " + "before anyone has looked at it." + ) + try: + encoded = pem.encode() + except UnicodeEncodeError as exc: + raise ValueError(f"pem is not encodable as UTF-8: {exc}") from exc + return serialization.load_pem_private_key(encoded, password=None) # type: ignore[return-value] def load_signing_key() -> Ed25519PrivateKey: @@ -82,7 +100,26 @@ def _okp_jwk(raw_public_bytes: bytes) -> dict[str, str]: def key_to_jwk(key: Ed25519PrivateKey) -> dict[str, str]: - """Return the public JWK dict for *key* (OKP / Ed25519).""" + """Return the public JWK dict for *key* (OKP / Ed25519). + + Raises ``ValueError`` for anything that is not an ``Ed25519PrivateKey``. A + public key is called out separately because it is the plausible mistake here: + the name reads as "turn a key into a JWK", the result is the *public* JWK, and + a caller holding only the public half will reach for this. It is not a widening + this function can make on its own, since ``sign_record`` depends on being handed + something that can sign; ``_jwk_from_public_key`` is the path for that half. + """ + if not isinstance(key, Ed25519PrivateKey): + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + if isinstance(key, Ed25519PublicKey): + raise ValueError( + "key_to_jwk needs the private key, not the public one. It derives the " + "public JWK from it, and its callers go on to sign with the same object." + ) + raise ValueError( + f"key must be an Ed25519PrivateKey, got {type(key).__name__}" + ) return _okp_jwk( key.public_key().public_bytes( encoding=serialization.Encoding.Raw, @@ -137,6 +174,12 @@ def jwk_thumbprint(jwk: dict[str, Any]) -> str: Raises ``ValueError`` for an unknown ``kty`` or a missing required member. """ + if not isinstance(jwk, dict): + raise ValueError( + f"jwk must be a JSON object, got {type(jwk).__name__}. A JWK reaches this " + "function from a peer, a key document or a record's own `cnf`, so its shape " + "is not something the caller has already established." + ) kty = jwk.get("kty") if not isinstance(kty, str) or kty not in _THUMBPRINT_MEMBERS: raise ValueError( @@ -173,10 +216,32 @@ def _check_not_revoked(jwk: dict[str, Any], revocation: RevocationStore) -> None Both outcomes fail closed. An unreachable revocation source is not evidence that a key is unrevoked, so a store that raises is treated as a rejection rather than passed over. + + A callable that returns a non-bool has also not determined anything, and is + treated the same way. Reading its answer by truthiness would decide the one + check here that exists to catch a compromised key, on a value whose truthiness + means nothing about revocation. The membership branch needs no such guard: + ``in`` yields a real bool whatever ``__contains__`` returns. """ for identifier in _key_identifiers(jwk): try: - revoked = revocation(identifier) if callable(revocation) else identifier in revocation + if callable(revocation): + revoked = revocation(identifier) + if not isinstance(revoked, bool): + # `RevocationStore` is `Callable[[str], bool]`, and a store that + # answers with anything else has not answered. Truthiness would + # decide it here, and truthiness is unrelated to revocation + # status: `None`, `""`, `0` and `[]` would all read as "not + # revoked", which is the direction that lets a compromised key + # through, while the string "no" would read as revoked. The + # `None` case is not hypothetical. It is what a lookup returns + # when its author handled the 200 and forgot the rest, which is + # exactly the outage this check exists to survive. + raise TypeError( + f"returned {type(revoked).__name__}, not bool" + ) + else: + revoked = identifier in revocation except Exception as exc: raise ValueError( f"revocation status for key {identifier!r} could not be determined: {exc}. " @@ -258,9 +323,20 @@ def anchor_bytes(value: Any) -> bytes: by name, is that diagnostic. """ _reject_unanchorable(value) - return json.dumps( - value, sort_keys=True, separators=(",", ":"), ensure_ascii=True - ).encode("ascii") + try: + return json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("ascii") + except TypeError as exc: + # `_reject_unanchorable` names the two cases section 1 puts outside the + # profile. A type JSON cannot serialize at all is a third, and it reached + # `json.dumps` and came back as "Object of type bytes is not JSON + # serializable": a message about a serializer, from a function whose stated + # purpose is to refuse the value by name. + raise UnanchorableValue( + f"$ holds a {type(value).__name__}, which is not JSON at all, so it has " + f"no anchor form: {exc}" + ) from exc def _b64url_decode(value: str, *, field: str) -> bytes: @@ -288,7 +364,16 @@ def sign_record(record: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any The returned dict is a plain JSON-serialisable object. Pass it to ``json.dumps()`` to get the wire form, or to ``TrustRecord.model_validate()`` to confirm structural validity before writing. + + Raises ``ValueError`` for a *record* that is not a JSON object. ``{**record}`` + reads it before anything establishes its shape, so a non-mapping raised a bare + ``TypeError`` naming a dict-unpacking failure, which is not the module's + documented refusal and tells the caller nothing about which argument was wrong. """ + if not isinstance(record, dict): + raise ValueError( + f"record must be a JSON object, got {type(record).__name__}" + ) jwk = key_to_jwk(key) payload: dict[str, Any] = {**record, "cnf": {"jwk": jwk}} body = _canonical_bytes({k: v for k, v in payload.items() if k != "signature"}) @@ -374,8 +459,9 @@ def verify_record( revocation status at verification time. Pass a ``revocation`` store, either a container of revoked key identifiers or a callable performing a live CRL/status/SCITT lookup. The trusted key is rejected if it is listed, or if - the store cannot answer. Identifiers are the key's RFC 7638 thumbprint - (``jwk_thumbprint``) and its ``kid``. + the store cannot answer: a callable that raises has not answered, and so has + one that returns anything other than ``True`` or ``False``. Identifiers are + the key's RFC 7638 thumbprint (``jwk_thumbprint``) and its ``kid``. ``revocation=None`` (the default) skips the check and keeps verification purely offline. Offline verification cannot prove non-revocation: a @@ -389,6 +475,12 @@ def verify_record( # Profile first: refuse semantics this build does not implement before spending # any work on the record. + if not isinstance(record, dict): + raise ValueError( + f"record must be a JSON object, got {type(record).__name__}. A Trust Record " + "is always an object, and what a verifier is handed is by definition not yet " + "established to be one." + ) profile = record.get("eat_profile") if not isinstance(profile, str) or not profile: raise ValueError( diff --git a/src/agentrust_trace/validate.py b/src/agentrust_trace/validate.py index 5349c948..1e305009 100644 --- a/src/agentrust_trace/validate.py +++ b/src/agentrust_trace/validate.py @@ -1,5 +1,6 @@ from __future__ import annotations +import copy import importlib.resources import json from functools import lru_cache @@ -19,8 +20,14 @@ def _validator() -> jsonschema.Draft202012Validator: return jsonschema.Draft202012Validator(_schema(), format_checker=jsonschema.FormatChecker()) -# Canonical schema exposed for downstream tooling that needs the raw dict. -SCHEMA: dict[str, Any] = _schema() +#: Canonical schema exposed for downstream tooling that needs the raw dict. +#: +#: A copy, deliberately. `_schema()` is `lru_cache`d and `_validator()` is built over +#: whatever it returns, so this name used to be the live object the validator reads. +#: Mutating it, which is the ordinary thing to do with a dict handed over to adapt, +#: silently reconfigured `validate_json`, `iter_errors` and the structural gate inside +#: `sign.verify_record`, process-wide, for every later call. +SCHEMA: dict[str, Any] = copy.deepcopy(_schema()) def validate_json(record: dict[str, Any]) -> None: diff --git a/tests/test_content_marking.py b/tests/test_content_marking.py index 860cee7f..ccf77c7f 100644 --- a/tests/test_content_marking.py +++ b/tests/test_content_marking.py @@ -207,3 +207,42 @@ def test_anchor_is_carried_when_given() -> None: def test_anchor_is_omitted_when_absent() -> None: """Omitted rather than null: an unanchored record has no entry to name.""" assert "anchor" not in build_assertion(_bytes(_record()), url=URL)["data"] + + +# --- record_bytes that are not bytes at all ----------------------------------- +# +# build_assertion validates the type of record_bytes and verify_assertion did +# not. The int case is the one worth a test of its own: bytes(5) is five zero +# bytes, so the value was hashed, did not match, and the caller was told the +# record at the URL had changed. That is a specific accusation about somebody +# else's server, made confidently and with a digest attached, when the only +# thing wrong was the argument. + +NOT_BYTES = [None, 5, 0, "a string", "", [], ["x"], {"a": 1}, True] + + +@pytest.mark.parametrize("bad", NOT_BYTES) +def test_verify_assertion_refuses_record_bytes_that_are_not_bytes(bad) -> None: + a = build_assertion(_bytes(_record()), url=URL) + with pytest.raises(ContentMarkingError, match="record_bytes must be the bytes retrieved"): + verify_assertion(a, bad) + + +def test_an_int_no_longer_reports_a_record_mismatch() -> None: + """The failure this closes, named on its own so it cannot come back quietly. + + RecordMismatch means "the URL is serving something else". Reporting it for a + caller's type error points the reader at the wrong party, and it is worse than + a crash for exactly that reason: a crash says the call was wrong. + """ + a = build_assertion(_bytes(_record()), url=URL) + with pytest.raises(ContentMarkingError) as excinfo: + verify_assertion(a, 5) + assert not isinstance(excinfo.value, RecordMismatch) + assert "does not match the assertion" not in str(excinfo.value) + + +def test_empty_bytes_are_refused_like_build_assertion_refuses_them() -> None: + a = build_assertion(_bytes(_record()), url=URL) + with pytest.raises(ContentMarkingError, match="record_bytes must be the bytes retrieved"): + verify_assertion(a, b"") diff --git a/tests/test_provenance.py b/tests/test_provenance.py index a2860f10..85382257 100644 --- a/tests/test_provenance.py +++ b/tests/test_provenance.py @@ -678,3 +678,34 @@ def test_zero_is_a_bound_and_not_a_falsy_stand_in_for_unset() -> None: verify_record(a_moment_ago, key_to_jwk(key)) # None: no age bound with pytest.raises(ProvenanceError, match="stale"): verify_record(a_moment_ago, key_to_jwk(key), max_age_seconds=0) + + +# --- the record's own type, which no guard on a field inside it can reach ------ +# +# #225 added _as_object for record["identity"] and record["tool_catalog"]. Neither +# is reachable until the record itself is a mapping: record.get(...) on a list or +# a string raises AttributeError, which is not the ProvenanceError verify_record +# documents and is not caught by a caller written against that contract. + +NOT_OBJECTS = [None, 5, 0, "a string", "", [], [1, 2], True, False, b"bytes"] + + +@pytest.mark.parametrize("bad", NOT_OBJECTS) +def test_verify_record_refuses_a_non_object_record(bad) -> None: + key = generate_key() + with pytest.raises(ProvenanceError, match="record must be a JSON object"): + verify_record(bad, key_to_jwk(key)) + + +@pytest.mark.parametrize("bad", NOT_OBJECTS) +def test_check_tool_catalog_refuses_a_non_object_record(bad) -> None: + with pytest.raises(ProvenanceError, match="record must be a JSON object"): + check_tool_catalog(bad, TOOLS) + + +def test_the_record_type_is_checked_before_the_record_is_read() -> None: + """A non-object record is refused for being one, not for a missing `format`.""" + key = generate_key() + with pytest.raises(ProvenanceError) as excinfo: + verify_record([], key_to_jwk(key)) + assert "unknown format" not in str(excinfo.value) diff --git a/tests/test_public_functions_raise_what_they_document.py b/tests/test_public_functions_raise_what_they_document.py new file mode 100644 index 00000000..a123903a --- /dev/null +++ b/tests/test_public_functions_raise_what_they_document.py @@ -0,0 +1,206 @@ +"""Every public function in the package, swept, against the error its module documents. + +This test exists because the claim it makes was made once before without it. The change +that closed the last three record-argument leaks said "every public entry point in the +package now reports zero leaks under the same sweep", and the sweep was never committed. +Nobody could re-run it, including its author, and it was wrong in a way that is invisible +from the sentence: it fed record-shaped values, so it never reached the functions whose +argument is a key, and ``key_to_jwk`` leaked ``AttributeError`` on every input including +the public key that is the plausible mistake for a function with that name. + +A claim about a surface has to be checked by something that finds the surface. So the +functions here are discovered by walking the package, and ``DOCUMENTED`` has to name every +module they live in or the first test fails. Adding a module without deciding what it +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. +""" +from __future__ import annotations + +import importlib +import inspect +import json +import pkgutil +import time +from collections.abc import Callable +from typing import Any + +import pytest + +import agentrust_trace as at +from agentrust_trace import (content_marking, generate_key, intent_bridge, key_to_jwk, + provenance, sign, validate) + +#: Values a caller can supply where an object, a string, a key or bytes is expected. +#: The last five are the ones that separate a strict canonicalizer from a permissive +#: one: JSON carries them, and RFC 8785 has no form for four of them. +JUNK: tuple[Any, ...] = ( + "a-string", 123, None, [1, 2], True, False, 0, {}, "", b"bytes", + 10**20, float("nan"), float("inf"), "\ud800", {"a": 1}, [{}], 1.5, +) + +#: The exceptions each module documents as its refusal. Anything else escaping a +#: public function in that module is a leak: a caller written against the documented +#: contract does not catch it. +DOCUMENTED: dict[str, tuple[str, ...]] = { + "content_marking": ("ContentMarkingError", "RecordMismatch"), + "intent_bridge": ("IntentBridgeError", "AuthorizationDenied", "AuthorizationMismatch"), + "provenance": ("ProvenanceError", "ToolCatalogMismatch"), + "sign": ("ValueError", "UnanchorableValue", "InvalidSignature"), + "validate": ("ValueError", "ValidationError"), + "models": ("ValidationError",), + "adapters": ("ValueError", "ValidationError"), +} + +_KEY = generate_key() +_JWK = key_to_jwk(_KEY) +_RECORD_BYTES = json.dumps( + {"eat_profile": "x", "subject": "spiffe://e.example/agent/a"} +).encode() +_AUTHORIZATION = {"iss": "https://a.example", "sub": "urn:agent:x", "iat": int(time.time())} + +#: name -> a call varying only the first positional argument. +CALLS: dict[str, Callable[[Any], Any]] = { + "content_marking.build_assertion": + lambda v: content_marking.build_assertion(v, url="https://e.example/r.json"), + "content_marking.verify_assertion": + lambda v: content_marking.verify_assertion(v, _RECORD_BYTES), + "intent_bridge.digest_jcs": intent_bridge.digest_jcs, + "intent_bridge.sign_bridge": lambda v: intent_bridge.sign_bridge(v, _KEY), + "provenance.check_tool_catalog": lambda v: provenance.check_tool_catalog(v, []), + "provenance.sign_record": lambda v: provenance.sign_record(v, _KEY), + "provenance.tool_catalog_hash": provenance.tool_catalog_hash, + "provenance.verify_record": lambda v: provenance.verify_record(v, _JWK), + "sign.anchor_bytes": sign.anchor_bytes, + "sign.jwk_thumbprint": sign.jwk_thumbprint, + "sign.key_to_jwk": sign.key_to_jwk, + "sign.load_key": sign.load_key, + "sign.sign_record": lambda v: sign.sign_record(v, _KEY), + "sign.verify_record": lambda v: sign.verify_record(v, _JWK), + "validate.iter_errors": validate.iter_errors, + "validate.validate_json": validate.validate_json, +} + +#: 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. +NO_ARGUMENT_TO_SWEEP = { + "sign.generate_key", "sign.load_signing_key", + "provenance.build_record", + "intent_bridge.verify_bridge", # every argument is keyword-only and required +} + + +def _public_functions() -> dict[str, Any]: + """Walk the package. Discovered rather than listed: a hardcoded roster is how the + previous sweep missed a whole class of argument.""" + modules = [at] + for info in pkgutil.walk_packages(at.__path__, at.__name__ + "."): + if "__" not in info.name: + modules.append(importlib.import_module(info.name)) + found: dict[str, Any] = {} + for module in modules: + for name, obj in vars(module).items(): + if name.startswith("_") or inspect.isclass(obj) or not callable(obj): + continue + origin = getattr(obj, "__module__", "") + if origin.startswith("agentrust_trace"): + found[f"{origin.split('.')[-1]}.{name}"] = obj + return found + + +def test_the_walk_finds_something_to_sweep() -> None: + """An empty walk would make every test below vacuous and green.""" + found = _public_functions() + assert len(found) >= 20, f"only found {sorted(found)}" + + +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 + assert found == accounted, ( + f"not swept and not declared unsweepable: {sorted(found - accounted)}\n" + f"declared but no longer present: {sorted(accounted - found)}" + ) + + +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))}" + + +@pytest.mark.parametrize("name", sorted(CALLS)) +def test_no_public_function_raises_an_undocumented_exception(name: str) -> None: + allowed = DOCUMENTED[name.split(".")[0]] + call = CALLS[name] + + leaked: dict[str, Any] = {} + for value in JUNK: + try: + call(value) + 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} raised {leaked}, which its module does not document as its refusal. " + f"Documented: {allowed}. A caller written against that contract does not catch " + f"these." + ) + + +#: One value per function that must produce a named outcome, and the outcome it must +#: produce. A ratio over the junk matrix cannot serve here: `anchor_bytes` and +#: `sign_bridge` legitimately accept most of it, so "most inputs raised" is a property +#: of the function rather than evidence the call is wired up. An explicit witness is. +REACHES: dict[str, tuple[Any, str]] = { + "content_marking.build_assertion": (None, "ContentMarkingError"), + "content_marking.verify_assertion": (None, "ContentMarkingError"), + "intent_bridge.digest_jcs": ("a-string", "IntentBridgeError"), + "intent_bridge.sign_bridge": ({"k": float("nan")}, "IntentBridgeError"), + "provenance.check_tool_catalog": (None, "ProvenanceError"), + "provenance.sign_record": (None, "ProvenanceError"), + "provenance.tool_catalog_hash": (None, "ProvenanceError"), + "provenance.verify_record": (None, "ProvenanceError"), + "sign.anchor_bytes": (b"bytes", "UnanchorableValue"), + "sign.jwk_thumbprint": (None, "ValueError"), + "sign.key_to_jwk": (None, "ValueError"), + "sign.load_key": (None, "ValueError"), + "sign.sign_record": (None, "ValueError"), + "sign.verify_record": (None, "ValueError"), + "validate.validate_json": (None, "ValidationError"), +} + + +def test_every_swept_function_has_a_witness() -> None: + """`iter_errors` is excluded on purpose and named, rather than silently absent: + it returns findings instead of raising, and its witness is the next test.""" + assert set(REACHES) == set(CALLS) - {"validate.iter_errors"} + + +@pytest.mark.parametrize("name", sorted(REACHES)) +def test_the_sweep_actually_reaches_each_function(name: str) -> None: + """A `CALLS` entry can be wrong in a way that never reaches its function, and a + sweep that never arrives reports clean. This is what makes the clean reading mean + something.""" + value, expected = REACHES[name] + with pytest.raises(Exception) as caught: # noqa: PT011 - the type is the assertion + CALLS[name](value) + assert type(caught.value).__name__ == expected, ( + f"{name}({value!r}) raised {type(caught.value).__name__}, expected {expected}" + ) + + +def test_iter_errors_reports_rather_than_raising() -> None: + """Its contract is a list of findings, so silence is its failure mode, not an + 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" diff --git a/tests/test_sign.py b/tests/test_sign.py index 08508929..fad6c94a 100644 --- a/tests/test_sign.py +++ b/tests/test_sign.py @@ -570,6 +570,69 @@ def unreachable(_identifier: str) -> bool: verify_record(record, key_to_jwk(key), revocation=unreachable) +@pytest.mark.parametrize("answer", [None, "", 0, [], {}, 0.0]) +def test_verify_record_fails_closed_when_the_store_answers_with_a_non_bool(answer): + """A falsy non-bool is not "not revoked". It is no answer at all. + + ``RevocationStore`` is ``Callable[[str], bool]``. Before this, the return value + was read by truthiness, so every value here let the key through. ``None`` is the + one that matters: it is what a lookup returns when its author handled the 200 and + forgot the rest, which is precisely the outage + ``test_verify_record_fails_closed_when_revocation_source_errors`` exists to + survive. The two cases are one fact, and only the noisier half was covered. + """ + key = generate_key() + record = sign_record(_fresh_record(), key) + + with pytest.raises(ValueError, match="could not be determined"): + verify_record(record, key_to_jwk(key), revocation=lambda _identifier: answer) + + +@pytest.mark.parametrize("answer", ["no", "false", "unrevoked", [0]]) +def test_verify_record_fails_closed_when_a_truthy_non_bool_would_have_read_as_revoked(answer): + """The same guard, from the side that would not have been a security hole. + + Truthiness read ``"no"`` as revoked and ``0`` as not revoked, which is not a + conservative reading in one direction and a lax one in the other. It is no + reading at all: the truth value of a string says nothing about revocation. Both + halves have to fail closed or the guard is a coin flip that happens to land + safely half the time. + """ + key = generate_key() + record = sign_record(_fresh_record(), key) + + with pytest.raises(ValueError, match="could not be determined"): + verify_record(record, key_to_jwk(key), revocation=lambda _identifier: answer) + + +def test_verify_record_still_accepts_the_two_answers_the_type_allows(): + """Without this, refusing every callable would pass both tests above.""" + key = generate_key() + record = sign_record(_fresh_record(), key) + + verify_record(record, key_to_jwk(key), revocation=lambda _identifier: False) + + with pytest.raises(ValueError, match="revoked"): + verify_record(record, key_to_jwk(key), revocation=lambda _identifier: True) + + +def test_the_membership_branch_is_untouched_by_the_bool_guard(): + """``in`` yields a real bool whatever ``__contains__`` returns, so a container + store needs no guard and must not acquire one by accident.""" + + class AnswersWithAString: + def __contains__(self, _identifier: object) -> bool: + return "yes" # type: ignore[return-value] + + key = generate_key() + record = sign_record(_fresh_record(), key) + + with pytest.raises(ValueError, match="revoked"): + verify_record(record, key_to_jwk(key), revocation=AnswersWithAString()) + + verify_record(record, key_to_jwk(key), revocation=set()) + + def test_verify_record_revocation_works_with_public_key_object(): """The trusted key may be an Ed25519PublicKey; its JWK is derived for the check.""" key = generate_key() @@ -686,3 +749,42 @@ def test_round_trip_with_non_ascii_payload(): record["model"]["provider"] = "modèle français \U0001f916" signed = sign_record(record, key) verify_record(signed, key_to_jwk(key)) # must not raise + + +#: Values a caller can hand a function that documents an object argument. The last +#: five are the ones a record assembled from parsed JSON can actually carry. +_NOT_AN_OBJECT = ("a-string", 123, None, [1, 2], True, False, 0, "", b"bytes", 1.5) + + +@pytest.mark.parametrize("value", _NOT_AN_OBJECT) +def test_jwk_thumbprint_refuses_a_non_object_with_the_error_it_documents(value): + """`jwk_thumbprint` documents `ValueError` and read `.get` off its argument first. + + A JWK reaches it from a peer, a key document, or a record's own `cnf`, so its shape + is not something the caller has established. Before this it raised `AttributeError`, + which a caller written against the documented contract does not catch. + """ + with pytest.raises(ValueError): + jwk_thumbprint(value) + + +@pytest.mark.parametrize("value", _NOT_AN_OBJECT) +def test_verify_record_refuses_a_non_object_record_with_the_error_it_documents(value): + """Same shape, on the argument that is by definition untrusted. + + `verify_record`'s docstring says every rejection other than a bad signature is a + `ValueError`. A non-object record reached `record.get("eat_profile")` and raised + `AttributeError` instead. + """ + key = generate_key() + with pytest.raises(ValueError): + verify_record(value, key_to_jwk(key)) + + +def test_the_guards_do_not_refuse_what_they_should_accept(): + """Without this, raising unconditionally would pass both tests above.""" + key = generate_key() + record = sign_record(_fresh_record(), key) + + jwk_thumbprint(key_to_jwk(key)) + verify_record(record, key_to_jwk(key)) diff --git a/tests/test_the_docs_do_not_leave_a_key_behind.py b/tests/test_the_docs_do_not_leave_a_key_behind.py new file mode 100644 index 00000000..123eccd3 --- /dev/null +++ b/tests/test_the_docs_do_not_leave_a_key_behind.py @@ -0,0 +1,87 @@ +"""Anything the documentation writes into a reader's tree must be ignored by git. + +`docs/quickstart.md` writes an unencrypted Ed25519 private key to `trace-key.pem`, under +a comment that reads *"keep secure, never commit or log"*. That comment was the whole of +the enforcement. `.gitignore` did not cover the path, so a reader following the quickstart +inside a clone, which is what a quickstart invites, was one `git add -A` away from +committing their signing key. + +The finding came from running the documentation's code blocks and then noticing the three +files they left behind in the working tree, which is the sort of thing a check for +untracked files sees and a reader does not. + +The paths are recovered from the documentation rather than listed here, so a doc that +starts writing somewhere new fails this rather than quietly widening the gap. +""" +from __future__ import annotations + +import pathlib +import re +import subprocess + +import pytest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +DOCS = ROOT / "docs" + +_WRITES = ( + re.compile(r"""open\(\s*["']([^"']+)["']\s*,\s*["']w"""), + re.compile(r"""Path\(\s*["']([^"']+)["']\s*\)\.write_(?:text|bytes)"""), +) + + +def _paths_the_docs_write() -> set[str]: + found: set[str] = set() + for doc in sorted(DOCS.rglob("*.md")): + text = doc.read_text(encoding="utf-8") + for pattern in _WRITES: + found.update(pattern.findall(text)) + return {p for p in found if "/" not in p or not p.startswith("/")} + + +def test_the_documentation_writes_something() -> None: + """A recovery that found nothing would make the check below vacuous, and the + patterns above are the kind of thing that stops matching after an edit.""" + found = _paths_the_docs_write() + assert found, "recovered no written paths from docs/; the patterns have stopped matching" + assert "trace-key.pem" in found, f"the quickstart's key is not among {sorted(found)}" + + +@pytest.mark.parametrize("name", sorted(_paths_the_docs_write())) +def test_git_ignores_what_the_docs_write(name: str) -> None: + result = subprocess.run( + ["git", "check-ignore", "-q", name], cwd=ROOT, capture_output=True + ) + assert result.returncode == 0, ( + f"docs/ tells a reader to write {name!r} and .gitignore does not cover it. " + "A reader following along inside a clone can commit it by accident, and for a " + "private key that is the one mistake this repository cannot help them undo." + ) + + +def test_no_key_material_is_tracked() -> None: + """The other direction: not just ignored going forward, but absent today.""" + tracked = subprocess.run( + ["git", "ls-files"], cwd=ROOT, capture_output=True, text=True + ).stdout.split() + keys = [f for f in tracked if f.lower().endswith((".pem", ".key", ".p8", ".pfx"))] + assert not keys, f"key material is tracked in the repository: {keys}" + + +def test_the_ignore_rule_is_not_so_broad_it_hides_real_files() -> None: + """Without this, `*` in .gitignore would pass every test above. + + `--no-index` is required and is the whole point. Plain `git check-ignore` reports a + *tracked* file as not ignored whatever the rules say, because ignore rules do not + apply to tracked files, so the first version of this test passed with `*` appended + and guarded nothing. Asking about the rules rather than about the index is the + difference. + """ + for kept in ("README.md", "pyproject.toml", "src/agentrust_trace/sign.py"): + result = subprocess.run( + ["git", "check-ignore", "-q", "--no-index", kept], cwd=ROOT, capture_output=True + ) + assert result.returncode != 0, ( + f".gitignore now matches {kept}, which is tracked. The rule is too broad: it " + "would hide a new file beside an existing one and nothing would say so." + ) diff --git a/tests/test_the_exported_schema_is_not_the_live_one.py b/tests/test_the_exported_schema_is_not_the_live_one.py new file mode 100644 index 00000000..9ea2472d --- /dev/null +++ b/tests/test_the_exported_schema_is_not_the_live_one.py @@ -0,0 +1,85 @@ +"""`SCHEMA` is handed to callers. Mutating it must not reconfigure the verifier. + +`validate.py` exposes it with the comment "exposed for downstream tooling that needs the +raw dict", and `_schema()` is `lru_cache(maxsize=1)` with `_validator()` built over +whatever it returns. So the exported name *was* the live object, and the ordinary thing to +do with a dict you were handed to adapt, mutate it, silently reconfigured `validate_json`, +`iter_errors`, and the structural gate inside `sign.verify_record`, for every later call in +the process. + +Nothing about the call site looks wrong. `s = SCHEMA` then `s["properties"][...] = ...` is +what a caller building a variant writes. +""" +from __future__ import annotations + +import copy +from typing import Any + +import jsonschema +import pytest + +from agentrust_trace import SCHEMA, validate_json +from agentrust_trace.validate import _schema + +RECORD: dict[str, Any] = { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1750000000, + "subject": "did:mesh:spiffe://factory.example/agent/payments/prod", + "model": {"provider": "anthropic", "model_id": "claude-sonnet-4-6"}, + "runtime": {"platform": "software-only", "measurement": "sha256:" + "0" * 64}, + "policy": {"bundle_hash": "sha256:" + "a" * 64, "enforcement_mode": "enforce"}, + "data_class": "confidential", + "build_provenance": {"slsa_level": 0, "digest": "sha256:" + "b" * 64}, + "appraisal": {"status": "affirming", "verifier": "https://agt.example.org/verifier"}, + "cnf": {"jwk": {"kty": "OKP", "crv": "Ed25519", + "x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo"}}, +} + + +def test_the_record_is_valid_and_a_1970_one_is_not() -> None: + """The control, and the thing the mutation below tries to change.""" + validate_json(RECORD) + with pytest.raises(jsonschema.ValidationError): + validate_json(dict(RECORD, iat=1)) + + +def test_the_export_is_not_the_object_the_validator_reads() -> None: + assert SCHEMA is not _schema(), ( + "SCHEMA is the live cached schema. A caller adapting it reconfigures the " + "verifier for the whole process." + ) + assert SCHEMA == _schema(), "the copy has drifted from the schema it copies" + + +def test_mutating_the_export_does_not_weaken_validation() -> None: + """The behaviour, not only the identity. Two objects that are `is not` each other can + still share nested dicts, and a shallow copy would pass the test above.""" + original = copy.deepcopy(SCHEMA) + try: + SCHEMA["properties"]["iat"]["minimum"] = 0 + + with pytest.raises(jsonschema.ValidationError): + validate_json(dict(RECORD, iat=1)) + finally: + SCHEMA.clear() + SCHEMA.update(original) + + +def test_deleting_from_the_export_does_not_weaken_validation() -> None: + """Removal, not only alteration: dropping `required` is the widest single edit.""" + original = copy.deepcopy(SCHEMA) + try: + SCHEMA.pop("required", None) + + with pytest.raises(jsonschema.ValidationError): + validate_json({"eat_profile": "tag:agentrust-io.com,2026:trace-v0.2"}) + finally: + SCHEMA.clear() + SCHEMA.update(original) + + +def test_the_export_still_carries_the_whole_schema() -> None: + """Without this, exporting an empty dict would pass everything above.""" + assert SCHEMA["properties"]["iat"]["minimum"] == 1700000000 + assert "build_provenance" in SCHEMA["required"] + assert len(SCHEMA["properties"]) >= 12 diff --git a/tests/test_the_model_round_trip_preserves_a_record.py b/tests/test_the_model_round_trip_preserves_a_record.py new file mode 100644 index 00000000..5d96db0d --- /dev/null +++ b/tests/test_the_model_round_trip_preserves_a_record.py @@ -0,0 +1,126 @@ +"""`TrustRecord.model_validate(record).model_dump()` has to give the record back. + +`sign_record`'s docstring points a caller at exactly this round trip: pass the returned +dict to `TrustRecord.model_validate()` to confirm structural validity before writing. A +caller who then wrote the model out wrote a different record. Pydantic serializes every +unset optional as an explicit `null`, the schema permits `null` for no named field, and +the added members change the RFC 8785 canonical bytes the signature is taken over. So: + + validate_json(TrustRecord.model_validate(record).model_dump()) + ValueError: ... None is not of type 'string' + + verify_record(TrustRecord.model_validate(record).model_dump(), jwk) + InvalidSignature + +Two artifacts of this package disagreeing about one record, and neither the validator nor +the signature check runs at the moment the damage is done: the caller has a model in hand +and a `model_dump` that looks like a record. + +The fix omits absent optionals. It has to omit only *declared* ones: `JWK` sets +`extra="allow"` and the schema's `canonicalizableValue` permits a null there, so a null +inside `cnf.jwk` is data. A first version filtered the whole serialized dict and removed +it, which is the same defect one level down. +""" +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from agentrust_trace import (TrustRecord, generate_key, key_to_jwk, sign_record, + validate_json, verify_record) + +KEY = generate_key() +JWK = key_to_jwk(KEY) + +#: The optional members pydantic used to write as `null`. Named rather than counted, so +#: this says what it prevents. +ABSENT_OPTIONALS = ("transparency", "tool_transcript", "delegation", "origin", "references") + + +def _record(**over: Any) -> dict[str, Any]: + base = { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1750000000, + "subject": "did:mesh:spiffe://factory.example/agent/payments/prod", + "model": {"provider": "anthropic", "model_id": "claude-sonnet-4-6"}, + "runtime": {"platform": "software-only", "measurement": "sha256:" + "0" * 64}, + "policy": {"bundle_hash": "sha256:" + "a" * 64, "enforcement_mode": "enforce"}, + "data_class": "confidential", + "build_provenance": {"slsa_level": 0, "digest": "sha256:" + "b" * 64}, + "appraisal": {"status": "affirming", "verifier": "https://agt.example.org/verifier"}, + } + base.update(over) + return sign_record(base, KEY) + + +def test_the_starting_record_is_valid_and_verifies() -> None: + """The control. Everything below compares against this, and a fixture that was + already broken would make each comparison a comparison of two failures.""" + record = _record() + validate_json(record) + verify_record(record, JWK, max_age_seconds=None) + + +@pytest.mark.parametrize("dump", ["model_dump", "model_dump_json"]) +def test_the_round_trip_returns_the_same_record(dump: str) -> None: + record = _record() + + model = TrustRecord.model_validate(record) + out = model.model_dump() if dump == "model_dump" else json.loads(model.model_dump_json()) + + assert out == record, ( + f"{dump}() did not give the record back. Added: " + f"{sorted(set(out) - set(record))}; dropped: {sorted(set(record) - set(out))}" + ) + + +@pytest.mark.parametrize("dump", ["model_dump", "model_dump_json"]) +def test_the_round_trip_still_validates_and_still_verifies(dump: str) -> None: + """Asserted separately from identity, because a serialization could differ from the + input in some harmless way and still be a record. These are the two properties that + actually matter to a caller, and they were both false.""" + model = TrustRecord.model_validate(_record()) + out = model.model_dump() if dump == "model_dump" else json.loads(model.model_dump_json()) + + validate_json(out) + verify_record(out, JWK, max_age_seconds=None) + + +@pytest.mark.parametrize("field", ABSENT_OPTIONALS) +def test_an_absent_optional_is_absent_and_not_null(field: str) -> None: + out = TrustRecord.model_validate(_record()).model_dump() + + assert field not in out, ( + f"{field} was written as {out[field]!r}. The schema types it non-nullable, so a " + "record carrying it as null is rejected by every implementation validating " + "against the published schema, including this package's own validate_json." + ) + + +def test_an_optional_that_is_present_survives() -> None: + """Without this, a serializer that dropped every optional would pass everything + above.""" + record = _record(transparency="https://rekor.example/api/v1/log/entries/x") + + out = TrustRecord.model_validate(record).model_dump() + + assert out["transparency"] == "https://rekor.example/api/v1/log/entries/x" + + +def test_a_null_a_caller_put_in_cnf_jwk_is_data_and_survives() -> None: + """`JWK` allows extra members and the schema's `canonicalizableValue` permits a null + among them, so this one is not an unset field and must not be filtered. + + The signature is not asserted here: the extra is added after signing, which changes + the record, so a mismatch would be the fixture's doing rather than the serializer's. + Identity and schema validity are the properties in question. + """ + record = _record() + record["cnf"]["jwk"]["x5t#S256"] = None + validate_json(record) + + out = TrustRecord.model_validate(record).model_dump() + + assert out == record, "a null extra in cnf.jwk was filtered as though it were unset" diff --git a/tests/test_the_schema_and_the_models_agree.py b/tests/test_the_schema_and_the_models_agree.py new file mode 100644 index 00000000..34536f60 --- /dev/null +++ b/tests/test_the_schema_and_the_models_agree.py @@ -0,0 +1,196 @@ +"""Two validators, one record shape. A caller may reach either, so they have to agree. + +``schema/trace-claim.json`` is the published artifact an implementation in any language +validates against. ``models.TrustRecord`` is the artifact a Python producer builds against, +and ``models.py`` already says why that matters: *"a model that accepts what the schema +rejects sends the failure downstream to whichever canonicalizer the producer happens to be +using."* Nothing checked the two against each other. + +Mutating every field of a valid record across a value matrix found eight records the two +disagreed about. Four were booleans where JSON says integer. ``isinstance(True, int)`` is a +Python fact and not a JSON one: JSON Schema's ``"type": "integer"`` does not match ``true``, +so the schema rejected ``{"slsa_level": true}`` and the model accepted it **and coerced it +to 1**. The record was then a claim of SLSA build level 1 assembled out of a boolean, which +no other implementation would have validated. ``appraisal.timestamp`` did the same and read +``true`` as 1 January 1970. + +``iat`` and ``origin.ingested_at`` did not have the hole, and were safe by accident rather +than by design: their lower bound sits above 1, so the coerced value failed the range check +after the coercion. They carry the guard now too, so the safety does not depend on a bound +nobody is thinking about when they change it. + +The remaining four disagreements are one question and are declared below rather than fixed, +because answering it is a change to the published schema or a break for Python callers, and +neither is a test's decision to make. +""" +from __future__ import annotations + +import copy +from typing import Any + +import jsonschema +import pytest + +from agentrust_trace import TrustRecord, validate_json + +BASE: dict[str, Any] = { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1750000000, + "subject": "did:mesh:spiffe://factory.example/agent/payments/prod", + "model": {"provider": "anthropic", "model_id": "claude-sonnet-4-6"}, + "runtime": {"platform": "software-only", "measurement": "sha256:" + "0" * 64}, + "policy": {"bundle_hash": "sha256:" + "a" * 64, "enforcement_mode": "enforce"}, + "data_class": "confidential", + "build_provenance": {"slsa_level": 0, "digest": "sha256:" + "b" * 64}, + "appraisal": {"status": "affirming", "verifier": "https://agt.example.org/verifier"}, + "transparency": "https://rekor.sigstore.dev/api/v1/log/entries/example", + "tool_transcript": {"hash": "sha256:" + "c" * 64, "call_count": 3}, + "cnf": {"jwk": {"kty": "OKP", "crv": "Ed25519", + "x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo"}}, +} + +MUTANTS: tuple[Any, ...] = ( + "a-string", 123, None, [1, 2], True, False, 0, {}, "", -1, 1.5, + 10**20, 9007199254740992, "sha256:" + "z" * 64, "sha1:" + "a" * 40, + "SHA256:" + "a" * 64, " ", "\t", "x" * 5000, "https://", "not-a-uri", + "sha256:" + "A" * 64, "sha256:" + "a" * 63, "sha256:" + "a" * 65, +) + +#: (path, repr of the value) the two are known to disagree about, and why it is not +#: fixed here. A disagreement not in this set fails the test. Removing one that has +#: been resolved is the other half: this set may not carry a row that now agrees. +#: Fields the schema types `format: uri` and the models type a bare `str`. +#: +#: `jsonschema` enforces `format` only when a checker for it is installed, so whether +#: these fields diverge depends on the environment rather than on either artifact. With +#: the checker absent the format is inert and the two agree; with it present the schema +#: refuses any non-URI and the models accept it. Both readings are correct about the +#: code, so the set is computed rather than fixed, and the reason is the same either +#: way: mirroring `format: uri` in the models is its own change. +_URI_FORMAT_ENFORCED = "uri" in jsonschema.FormatChecker().checkers +FIELDS_WITH_NO_URI_VALIDATION_IN_THE_MODEL = ( + {"appraisal.verifier", "transparency"} if _URI_FORMAT_ENFORCED else set() +) + +DECLARED_DIVERGENCES: dict[tuple[str, str], str] = { + ("transparency", "None"): + "explicit JSON null for an optional field: the schema types it 'string' and " + "does not permit null, the model types it 'str | None'. Resolving it means " + "either 'type': ['string', 'null'] in the published schema or refusing None " + "from Python callers, and both are somebody's decision rather than a test's.", + ("tool_transcript", "None"): "the same question, on an optional object.", + ("tool_transcript.call_count", "None"): "the same question, on an optional integer.", + ("transparency", "''"): + "the opposite direction: the model carries min_length=1 and the schema has no " + "minLength, so the schema admits an empty URI that the model refuses. Adding " + "minLength to the schema is a change to the published artifact.", +} + + +def _by_schema(record: dict[str, Any]) -> bool: + try: + validate_json(record) + except Exception: # noqa: BLE001 - accept or reject is the whole signal + return False + return True + + +def _by_model(record: dict[str, Any]) -> bool: + try: + TrustRecord.model_validate(record) + except Exception: # noqa: BLE001 + return False + return True + + +def _paths(obj: Any, prefix: tuple[str, ...] = ()) -> Any: + if isinstance(obj, dict): + for key, value in obj.items(): + yield prefix + (key,) + yield from _paths(value, prefix + (key,)) + + +def _set(obj: dict[str, Any], path: tuple[str, ...], value: Any) -> None: + for key in path[:-1]: + obj = obj[key] + obj[path[-1]] = value + + +def test_the_base_record_passes_both() -> None: + """The control. Every comparison below is meaningless if the starting point is not + valid to both, and an incomplete fixture is the easy way to get a clean run that + measured nothing.""" + assert _by_schema(BASE), "the schema rejects the base record" + assert _by_model(BASE), "the model rejects the base record" + + +def _disagreements() -> dict[tuple[str, str], tuple[bool, bool]]: + found: dict[tuple[str, str], tuple[bool, bool]] = {} + for path in list(_paths(BASE)): + for mutant in MUTANTS: + record = copy.deepcopy(BASE) + _set(record, path, mutant) + schema, model = _by_schema(record), _by_model(record) + if schema != model: + found[(".".join(path), repr(mutant)[:26])] = (schema, model) + return found + + +def test_the_sweep_covers_the_record() -> None: + """A walk that found nothing to mutate would pass every test in this file.""" + assert len(list(_paths(BASE))) >= 25 + assert len(MUTANTS) >= 20 + + +def test_the_two_validators_agree_except_where_declared() -> None: + found = _disagreements() + + undeclared = { + k: v for k, v in found.items() + if k not in DECLARED_DIVERGENCES + and k[0] not in FIELDS_WITH_NO_URI_VALIDATION_IN_THE_MODEL + } + assert not undeclared, ( + "the schema and the model disagree about records not declared in " + "DECLARED_DIVERGENCES:\n" + "\n".join( + f" {path} = {value}: schema={s}, model={m}" + for (path, value), (s, m) in sorted(undeclared.items()) + ) + "\nOne of the two is wrong. The schema is the published artifact, so a " + "record the model accepts and the schema rejects is one no other " + "implementation will validate." + ) + + +def test_no_declared_divergence_has_quietly_been_resolved() -> None: + """The other half of the declaration. A stale entry reads as an open question that + somebody still owes an answer to, and it is not.""" + found = _disagreements() + stale = sorted( + k for k in set(DECLARED_DIVERGENCES) - set(found) + if k[0] not in FIELDS_WITH_NO_URI_VALIDATION_IN_THE_MODEL + ) + assert not stale, f"these now agree and should be removed from the set: {stale}" + + +@pytest.mark.parametrize("path", [ + ("build_provenance", "slsa_level"), + ("tool_transcript", "call_count"), + ("appraisal", "timestamp"), + ("iat",), +]) +@pytest.mark.parametrize("value", [True, False]) +def test_no_integer_field_accepts_a_boolean(path: tuple[str, ...], value: bool) -> None: + """Named separately from the sweep because the coercion is the part that matters. + + Accepting `true` would be a permissiveness bug. Reading it as 1 makes a claim the + producer never wrote: SLSA build level 1, one tool call, or an appraisal timestamped + 1 January 1970. + """ + record = copy.deepcopy(BASE) + _set(record, path, value) + + assert not _by_schema(record), "the schema should reject a boolean here" + assert not _by_model(record), ( + f"{'.'.join(path)} accepted {value!r}; before this guard it became " + f"{int(value)!r} and the record claimed it" + ) diff --git a/tests/test_uri_formats_are_enforced.py b/tests/test_uri_formats_are_enforced.py new file mode 100644 index 00000000..d2af2aea --- /dev/null +++ b/tests/test_uri_formats_are_enforced.py @@ -0,0 +1,128 @@ +"""The schema's `format: uri` declarations have to do something. + +`jsonschema` treats `format` as an annotation, not an assertion, unless a checker for +that format is installed. `validate.py` builds its validator with +`format_checker=jsonschema.FormatChecker()`, which reads exactly as though URIs are +checked. They were not: `FormatChecker().checkers` ships with `date`, `email`, +`idn-email`, `ipv4`, `ipv6`, `regex`, `time` and `uuid`, and `uri` is not among them +without an optional dependency the project did not declare. + +So all eight `"format": "uri"` fields accepted `"not a uri"` and the empty string, and +nothing anywhere said so. The wiring was correct and the behaviour was a no-op, which is +the shape that survives review: there is nothing wrong to see in `validate.py`. + +The dependency is `rfc3986-validator` rather than `jsonschema[format]`, which pulls +`rfc3987` (GPLv3) into an Apache-2.0 package's dependency tree. + +The first test here is the one that matters. Asserting that a bad URI is rejected proves +the checker is present today; asserting the checker is present proves *why*, and fails +with a message that names the dependency rather than leaving somebody to rediscover this. +""" +from __future__ import annotations + +import copy +from typing import Any + +import jsonschema +import pytest + +from agentrust_trace import validate_json + +#: Every field in schema/trace-claim.json declaring `"format": "uri"`, recovered by hand +#: from the schema and pinned by `test_the_set_is_every_field_that_declares_the_format`. +URI_FIELDS = [ + ("model", "aibom_uri"), + ("runtime", "rim_uri"), + ("policy", "policy_uri"), + ("tool_transcript", "transcript_uri"), + ("build_provenance", "provenance_uri"), + ("appraisal", "verifier"), + ("appraisal", "policy_ref"), + ("transparency",), +] + +BASE: dict[str, Any] = { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1750000000, + "subject": "did:mesh:spiffe://factory.example/agent/payments/prod", + "model": {"provider": "anthropic", "model_id": "claude-sonnet-4-6"}, + "runtime": {"platform": "software-only", "measurement": "sha256:" + "0" * 64}, + "policy": {"bundle_hash": "sha256:" + "a" * 64, "enforcement_mode": "enforce"}, + "data_class": "confidential", + "build_provenance": {"slsa_level": 0, "digest": "sha256:" + "b" * 64}, + "appraisal": {"status": "affirming", "verifier": "https://agt.example.org/verifier"}, + "tool_transcript": {"hash": "sha256:" + "c" * 64, "call_count": 3}, + "cnf": {"jwk": {"kty": "OKP", "crv": "Ed25519", + "x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo"}}, +} + + +def _with(path: tuple[str, ...], value: Any) -> dict[str, Any]: + record = copy.deepcopy(BASE) + target = record + for key in path[:-1]: + target = target[key] + target[path[-1]] = value + return record + + +def test_the_uri_format_checker_is_installed() -> None: + """Fails with the reason rather than leaving somebody to rediscover it. + + Every assertion below passes vacuously without this: an unenforced format makes a + malformed URI valid, so a test that only checked acceptance of good URIs would be + green on a schema doing nothing. + """ + assert "uri" in jsonschema.FormatChecker().checkers, ( + "jsonschema has no `uri` format checker registered, so every `format: uri` in " + "schema/trace-claim.json is inert and `validate_json` accepts 'not a uri'. " + "Install the `rfc3986-validator` dependency declared in pyproject.toml." + ) + + +def test_the_set_is_every_field_that_declares_the_format() -> None: + """`URI_FIELDS` is written out, so it has to be checked against the schema. A field + that gains the format and not a row here would be untested and look tested.""" + from agentrust_trace.validate import SCHEMA + + def walk(node: Any, path: tuple[str, ...] = ()) -> Any: + if isinstance(node, dict): + if node.get("format") == "uri": + yield path + for key, value in node.items(): + yield from walk(value, path if key == "properties" else (*path, key)) + elif isinstance(node, list): + for item in node: + yield from walk(item, path) + + found = {tuple(p) for p in walk(SCHEMA)} + assert found == {tuple(f) for f in URI_FIELDS}, ( + f"schema declares format: uri on {sorted(found)}; this file lists " + f"{sorted(tuple(f) for f in URI_FIELDS)}" + ) + + +@pytest.mark.parametrize("path", URI_FIELDS, ids=lambda p: ".".join(p)) +@pytest.mark.parametrize("bad", ["not a uri", "", " "]) +def test_a_malformed_uri_is_rejected(path: tuple[str, ...], bad: str) -> None: + with pytest.raises(jsonschema.ValidationError): + validate_json(_with(path, bad)) + + +@pytest.mark.parametrize("path", URI_FIELDS, ids=lambda p: ".".join(p)) +@pytest.mark.parametrize( + "good", ["https://example.org/a", "urn:example:a", "did:web:example.org"] +) +def test_the_uris_a_record_legitimately_carries_are_accepted( + path: tuple[str, ...], good: str +) -> None: + """The control. A checker that rejected everything would pass the test above. + + `urn:` and `did:` are here on purpose: a naive check for a scheme-and-authority shape + would reject both, and records carry both. + """ + validate_json(_with(path, good)) + + +def test_the_base_record_is_valid() -> None: + validate_json(BASE)