fix(provenance): apply the structural rules on the consumer side too

src/agentrust_trace/provenance.py

@@ -91,6 +91,60 @@ def tool_catalog_hash(tools: list[dict[str, Any]]) -> str:
     return "sha256:" + hashlib.sha256(canonical).hexdigest()
 
 
+def _check_structure(
+    *,
+    kind: str,
+    artifact: dict[str, Any] | None,
+    endpoint: dict[str, Any] | None,
+    attestation: dict[str, Any] | None,
+    issued_at: Any,
+) -> None:
+    """The structural rules both the producer and the consumer must apply.
+
+    Shared rather than duplicated because these rules were enforced only in
+    :func:`build_record` for a time, and a record does not have to come from
+    ``build_record``: anyone can write the JSON and sign it. A rule that lives
+    only on the producer side is a rule an attacker simply does not run, so a
+    ``tee-attested`` record with ``attestation: null`` verified cleanly.
+
+    Keeping one implementation is the point. Two copies drift, and the copy that
+    matters is the one on the consumer side.
+    """
+    if artifact is not None:
+        if not artifact.get("package"):
+            raise ProvenanceError("artifact.package is required (a Package URL)")
+        if not _DIGEST_RE.match(str(artifact.get("digest", ""))):
+            raise ProvenanceError(
+                "artifact.digest must be a sha256: digest of the entrypoint. For an "
+                "interpreted server that is the script, not the interpreter: every such "
+                "server on a host shares one interpreter digest."
+            )
+    if endpoint is not None:
+        if not endpoint.get("url"):
+            raise ProvenanceError("endpoint.url is required when endpoint is present")
+        if not _DIGEST_RE.match(str(endpoint.get("spki_sha256", ""))):
+            raise ProvenanceError(
+                "endpoint.spki_sha256 must be a sha256: digest of the Subject Public Key "
+                "Info. A URL on its own is not an identity."
+            )
+    if kind == "tee-attested" and not attestation:
+        raise ProvenanceError(
+            "kind='tee-attested' without attestation evidence is the claim without the "
+            "thing that backs it"
+        )
+    if kind != "tee-attested" and attestation:
+        raise ProvenanceError(
+            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:
+        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."
+        )
+
+
 def build_record(
     *,
     kind: str,
@@ -120,33 +174,14 @@ def build_record(
             "a record needs artifact identity, endpoint identity, or both. One with "
             "neither identifies nothing."
         )
-    if artifact is not None:
-        if not artifact.get("package"):
-            raise ProvenanceError("artifact.package is required (a Package URL)")
-        if not _DIGEST_RE.match(artifact.get("digest", "")):
-            raise ProvenanceError(
-                "artifact.digest must be a sha256: digest of the entrypoint. For an "
-                "interpreted server that is the script, not the interpreter: every such "
-                "server on a host shares one interpreter digest."
-            )
-    if endpoint is not None:
-        if not endpoint.get("url"):
-            raise ProvenanceError("endpoint.url is required when endpoint is present")
-        if not _DIGEST_RE.match(endpoint.get("spki_sha256", "")):
-            raise ProvenanceError(
-                "endpoint.spki_sha256 must be a sha256: digest of the Subject Public Key "
-                "Info. A URL on its own is not an identity."
-            )
-    if kind == "tee-attested" and not attestation:
-        raise ProvenanceError(
-            "kind='tee-attested' without attestation evidence is the claim without the "
-            "thing that backs it"
-        )
-    if kind != "tee-attested" and attestation:
-        raise ProvenanceError(
-            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."
-        )
+    stamped_at = int(issued_at if issued_at is not None else time.time())
+    _check_structure(
+        kind=kind,
+        artifact=artifact,
+        endpoint=endpoint,
+        attestation=attestation,
+        issued_at=stamped_at,
+    )
 
     identity: dict[str, Any] = {}
     if artifact is not None:
@@ -157,7 +192,7 @@ def build_record(
     return {
         "format": FORMAT,
         "kind": kind,
-        "issued_at": int(issued_at if issued_at is not None else time.time()),
+        "issued_at": stamped_at,
         "identity": identity,
         "publisher": publisher,
         "tool_catalog": {"hash": tool_catalog_hash(tools), "tool_count": len(tools)},
@@ -199,6 +234,13 @@ def verify_record(record: dict[str, Any], trusted_jwk: dict[str, Any]) -> None:
     identity = record.get("identity") or {}
     if not identity.get("artifact") and not identity.get("endpoint"):
         raise ProvenanceError("identity carries neither an artifact nor an endpoint")
+    _check_structure(
+        kind=str(record.get("kind")),
+        artifact=identity.get("artifact"),
+        endpoint=identity.get("endpoint"),
+        attestation=record.get("attestation"),
+        issued_at=record.get("issued_at"),
+    )
     catalog = record.get("tool_catalog") or {}
     if not _DIGEST_RE.match(str(catalog.get("hash", ""))):
         raise ProvenanceError("tool_catalog.hash is not a sha256: digest")

tests/test_provenance.py

@@ -222,3 +222,97 @@ def test_both_identities_may_be_present() -> None:
     rec = _record(endpoint={"url": "https://mcp.acme.example/", "spki_sha256": OTHER_DIGEST})
     assert "artifact" in rec["identity"]
     assert "endpoint" in rec["identity"]
+
+
+# --- rules the consumer must apply, not only the producer (#142) -------------
+#
+# A record does not have to come from build_record. Anyone can write the JSON and
+# sign it, so a structural rule enforced only on the producer side is a rule an
+# attacker never runs. These assert the consumer applies each of them.
+
+
+def _forged(**over):
+    """A hand-assembled record that never passes through build_record."""
+    record = {
+        "format": FORMAT,
+        "kind": "publisher-asserted",
+        "issued_at": 1_754_000_000,
+        "identity": {"artifact": _artifact()},
+        "publisher": "did:web:acme.example",
+        "tool_catalog": {"hash": tool_catalog_hash(TOOLS), "tool_count": len(TOOLS)},
+        "attestation": None,
+    }
+    record.update(over)
+    return record
+
+
+def test_tee_attested_without_evidence_is_refused_by_the_verifier() -> None:
+    """Top trust tier with attestation: null, everything else well-formed."""
+    key = generate_key()
+    signed = sign_record(_forged(kind="tee-attested", attestation=None), key)
+    with pytest.raises(ProvenanceError, match="tee-attested"):
+        verify_record(signed, key_to_jwk(key))
+
+
+def test_the_reported_forgery_is_refused(capsys) -> None:
+    """The reproduction from #142 verbatim, which violates three rules at once.
+
+    Which rule fires first does not matter; that it verified at all was the bug.
+    """
+    key = generate_key()
+    forged = {
+        "format": FORMAT,
+        "kind": "tee-attested",
+        "identity": {"endpoint": {"url": "https://mcp.acme.example"}},
+        "publisher": "did:web:acme.example",
+        "tool_catalog": {"hash": tool_catalog_hash(TOOLS), "tool_count": len(TOOLS)},
+        "attestation": None,
+    }
+    signed = sign_record(forged, key)
+    with pytest.raises(ProvenanceError):
+        verify_record(signed, key_to_jwk(key))
+
+
+def test_endpoint_without_a_key_digest_is_refused_by_the_verifier() -> None:
+    key = generate_key()
+    signed = sign_record(_forged(identity={"endpoint": {"url": "https://mcp.acme.example"}}), key)
+    with pytest.raises(ProvenanceError, match="spki_sha256"):
+        verify_record(signed, key_to_jwk(key))
+
+
+def test_artifact_without_a_digest_is_refused_by_the_verifier() -> None:
+    key = generate_key()
+    signed = sign_record(
+        _forged(identity={"artifact": {"package": "pkg:npm/%40acme/mcp-search@2.1.0"}}), key
+    )
+    with pytest.raises(ProvenanceError, match="artifact.digest"):
+        verify_record(signed, key_to_jwk(key))
+
+
+@pytest.mark.parametrize("issued_at", [None, "tomorrow", -1, True])
+def test_unusable_issued_at_is_refused_by_the_verifier(issued_at) -> None:
+    """No issue time means a consumer cannot age the record, so it cannot reject a stale one."""
+    key = generate_key()
+    record = _forged()
+    if issued_at is None:
+        del record["issued_at"]
+    else:
+        record["issued_at"] = issued_at
+    signed = sign_record(record, key)
+    with pytest.raises(ProvenanceError, match="issued_at"):
+        verify_record(signed, key_to_jwk(key))
+
+
+def test_evidence_without_the_claim_is_refused_by_the_verifier() -> None:
+    """The mirror case: attestation present on a kind that does not claim it."""
+    key = generate_key()
+    signed = sign_record(_forged(attestation={"format": "sev-snp", "quote": "..."}), key)
+    with pytest.raises(ProvenanceError, match="attestation evidence"):
+        verify_record(signed, key_to_jwk(key))
+
+
+def test_a_well_formed_record_still_verifies() -> None:
+    """The new checks must not reject what build_record produces."""
+    key = generate_key()
+    signed = sign_record(_record(), key)
+    verify_record(signed, key_to_jwk(key))