From 64df3cd82aa34cb3ed230aa5eb8d9a8fc71973d8 Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Thu, 29 Oct 2026 10:13:54 -0500 Subject: [PATCH] fix(provenance,validate): remove str() coercion on digest/hash fields and harden _loads_schema (#627) Require isinstance(..., str) and _DIGEST_RE.match(...) on artifact.digest, endpoint.spki_sha256, and tool_catalog.hash in verify_record and check_tool_catalog in src/agentrust_trace/provenance.py, and reject duplicate JSON keys, non-finite constants, and non-object roots in _loads_schema in src/agentrust_trace/validate.py. Closes #627 Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/agentrust_trace/provenance.py | 15 ++++-- src/agentrust_trace/validate.py | 24 ++++++++- tests/test_provenance_locator_types.py | 73 ++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 6 deletions(-) diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index 6b04f3aa..fc66d835 100644 --- a/src/agentrust_trace/provenance.py +++ b/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( diff --git a/src/agentrust_trace/validate.py b/src/agentrust_trace/validate.py index 3bb27beb..0cf122e9 100644 --- a/src/agentrust_trace/validate.py +++ b/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) diff --git a/tests/test_provenance_locator_types.py b/tests/test_provenance_locator_types.py index c8006e3f..78c64e6a 100644 --- a/tests/test_provenance_locator_types.py +++ b/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()