fix(provenance,validate): remove str() coercion on digest/hash fields in provenance and enforce strict JSON in schema loaders (#627)

src/agentrust_trace/provenance.py

@@ -269,7 +269,8 @@ def _check_structure(
                 f"identity.artifact must be an object, got {type(artifact).__name__}"
             )
         _nonempty_string(artifact.get("package"), "artifact.package")
-        if not _DIGEST_RE.match(str(artifact.get("digest", ""))):
+        artifact_digest = artifact.get("digest")
+        if not isinstance(artifact_digest, str) or not _DIGEST_RE.match(artifact_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 "
@@ -281,7 +282,8 @@ def _check_structure(
                 f"identity.endpoint must be an object, got {type(endpoint).__name__}"
             )
         _nonempty_string(endpoint.get("url"), "endpoint.url")
-        if not _DIGEST_RE.match(str(endpoint.get("spki_sha256", ""))):
+        spki = endpoint.get("spki_sha256")
+        if not isinstance(spki, str) or not _DIGEST_RE.match(spki):
             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."
@@ -513,14 +515,15 @@ def verify_record(
     if not identity.get("artifact") and not identity.get("endpoint"):
         raise ProvenanceError("identity carries neither an artifact nor an endpoint")
     issued_at = _check_structure(
-        kind=str(record.get("kind")),
+        kind=record["kind"],
         artifact=identity.get("artifact"),
         endpoint=identity.get("endpoint"),
         attestation=record.get("attestation"),
         issued_at=record.get("issued_at"),
     )
     catalog = _as_object(record.get("tool_catalog"), "tool_catalog")
-    if not _DIGEST_RE.match(str(catalog.get("hash", ""))):
+    catalog_hash = catalog.get("hash")
+    if not isinstance(catalog_hash, str) or not _DIGEST_RE.match(catalog_hash):
         raise ProvenanceError("tool_catalog.hash is not a sha256: digest")
     _tool_count(catalog)
 
@@ -667,6 +670,10 @@ def check_tool_catalog(
     actual = tool_catalog_hash(tools, format=record["format"])
     catalog = _as_object(record.get("tool_catalog"), "tool_catalog")
     expected = catalog.get("hash")
+    if "hash" in catalog and (
+        not isinstance(expected, str) or not _DIGEST_RE.match(expected)
+    ):
+        raise ProvenanceError("tool_catalog.hash is not a sha256: digest")
     if actual != expected:
         declared_count = catalog.get("tool_count")
         raise ToolCatalogMismatch(

src/agentrust_trace/validate.py

@@ -28,10 +28,30 @@ def _ecma_pattern(
 )
 
 
+def _loads_schema(text: str) -> dict[str, Any]:
+    def _no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+        out: dict[str, Any] = {}
+        for key, value in pairs:
+            if key in out:
+                raise ValueError(f"duplicate JSON object key in schema: {key!r}")
+            out[key] = value
+        return out
+
+    def _no_nonfinite(token: str) -> Any:
+        raise ValueError(f"non-finite JSON constant in schema: {token}")
+
+    parsed = json.loads(
+        text, object_pairs_hook=_no_duplicate_keys, parse_constant=_no_nonfinite
+    )
+    if not isinstance(parsed, dict):
+        raise ValueError("schema must be a JSON object")
+    return parsed
+
+
 @lru_cache(maxsize=1)
 def _schema() -> dict[str, Any]:
     ref = importlib.resources.files("agentrust_trace") / "schema" / "trace-v0.2.json"
-    return cast(dict[str, Any], json.loads(ref.read_text(encoding="utf-8")))
+    return _loads_schema(ref.read_text(encoding="utf-8"))
 
 
 @lru_cache(maxsize=1)
@@ -79,7 +99,7 @@ def profiles_with_schema() -> frozenset[str]:
     for entry in (importlib.resources.files("agentrust_trace") / "schema").iterdir():
         if not entry.name.endswith(".json"):
             continue
-        schema = json.loads(entry.read_text(encoding="utf-8"))
+        schema = _loads_schema(entry.read_text(encoding="utf-8"))
         const = schema.get("properties", {}).get("eat_profile", {}).get("const")
         if isinstance(const, str):
             found.add(const)

tests/test_provenance_locator_types.py

@@ -92,3 +92,76 @@ def test_textual_locator_controls_still_pass() -> None:
     )
     assert artifact["identity"]["artifact"]["package"].startswith("pkg:")
     assert endpoint["identity"]["endpoint"]["url"].startswith("https://")
+
+
+class _StrDigest:
+    def __init__(self, text: str) -> None:
+        self._text = text
+
+    def __str__(self) -> str:
+        return self._text
+
+
+def test_provenance_rejects_str_coercible_non_string_digests_and_hashes() -> None:
+    fake_digest = _StrDigest(DIGEST)
+    with pytest.raises(ProvenanceError, match="artifact.digest"):
+        build_record(
+            kind="publisher-asserted",
+            publisher="did:web:acme.example",
+            tools=TOOLS,
+            artifact={"package": "pkg:npm/x@1.0.0", "digest": fake_digest},  # type: ignore[dict-item]
+        )
+    with pytest.raises(ProvenanceError, match="endpoint.spki_sha256"):
+        build_record(
+            kind="publisher-asserted",
+            publisher="did:web:acme.example",
+            tools=TOOLS,
+            endpoint={"url": "https://mcp.acme.example/", "spki_sha256": _StrDigest(SPKI)},  # type: ignore[dict-item]
+        )
+
+    key = generate_key()
+    rec = _base_record({"artifact": {"package": "pkg:npm/x@1.0.0", "digest": DIGEST}})
+    signed = sign_record(rec, key)
+    for bad_hash in (_StrDigest(signed["tool_catalog"]["hash"]), 123, "not-a-digest"):
+        signed_bad_cat = dict(
+            signed,
+            tool_catalog={"hash": bad_hash, "tool_count": 1},
+        )
+        with pytest.raises(ProvenanceError, match="tool_catalog.hash") as excinfo:
+            verify_record(signed_bad_cat, key_to_jwk(key))
+        assert type(excinfo.value) is ProvenanceError
+        with pytest.raises(ProvenanceError, match="tool_catalog.hash") as excinfo_cat:
+            from agentrust_trace.provenance import check_tool_catalog
+
+            check_tool_catalog(signed_bad_cat, TOOLS)
+        assert type(excinfo_cat.value) is ProvenanceError
+
+
+def test_validate_loads_schema_rejects_duplicate_keys_and_nonfinite_constants(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    from agentrust_trace import validate as val_mod
+
+    for idx, bad_json in enumerate(
+        (
+            '{"properties": {"a": 1}, "properties": {"b": 2}}',
+            '{"properties": {"eat_profile": {"const": "a"}}, "x": NaN}',
+            '{"properties": {"eat_profile": {"const": "a"}}, "x": Infinity}',
+            '{"properties": {"eat_profile": {"const": "a"}}, "x": -Infinity}',
+            "[1, 2, 3]",
+        )
+    ):
+        pkg_dir = tmp_path / f"pkg_{idx}"
+        schema_dir = pkg_dir / "schema"
+        schema_dir.mkdir(parents=True)
+        (schema_dir / "trace-v0.2.json").write_text(bad_json, encoding="utf-8")
+        val_mod._schema.cache_clear()
+        val_mod.profiles_with_schema.cache_clear()
+        monkeypatch.setattr(val_mod.importlib.resources, "files", lambda _pkg, d=pkg_dir: d)
+        with pytest.raises(ValueError):
+            val_mod._schema()
+        val_mod.profiles_with_schema.cache_clear()
+        with pytest.raises(ValueError):
+            val_mod.profiles_with_schema()
+    val_mod._schema.cache_clear()
+    val_mod.profiles_with_schema.cache_clear()