diff --git a/src/agentrust_trace/provenance.py b/src/agentrust_trace/provenance.py index a548deb..55441cd 100644 --- a/src/agentrust_trace/provenance.py +++ b/src/agentrust_trace/provenance.py @@ -66,6 +66,13 @@ def _as_object(value: Any, field: str) -> dict[str, Any]: return value +def _nonempty_string(value: Any, field: str) -> str: + """Return a required textual locator, rejecting truthy non-string JSON values.""" + if not isinstance(value, str) or not value: + raise ProvenanceError(f"{field} must be a non-empty string") + return value + + def _tool_count(catalog: dict[str, Any]) -> int: """Return the required catalog count as a JSON integer.""" value = catalog.get("tool_count") @@ -153,8 +160,7 @@ def _check_structure( raise ProvenanceError( f"identity.artifact must be an object, got {type(artifact).__name__}" ) - if not artifact.get("package"): - raise ProvenanceError("artifact.package is required (a Package URL)") + _nonempty_string(artifact.get("package"), "artifact.package") if not _DIGEST_RE.match(str(artifact.get("digest", ""))): raise ProvenanceError( "artifact.digest must be a sha256: digest of the entrypoint. For an " @@ -166,8 +172,7 @@ def _check_structure( raise ProvenanceError( f"identity.endpoint must be an object, got {type(endpoint).__name__}" ) - if not endpoint.get("url"): - raise ProvenanceError("endpoint.url is required when endpoint is present") + _nonempty_string(endpoint.get("url"), "endpoint.url") 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 " diff --git a/tests/test_provenance_locator_types.py b/tests/test_provenance_locator_types.py new file mode 100644 index 0000000..c8006e3 --- /dev/null +++ b/tests/test_provenance_locator_types.py @@ -0,0 +1,94 @@ +"""Regression tests for MCP provenance identity locator primitive boundaries.""" + +from __future__ import annotations + +import time +from typing import Any + +import pytest + +from agentrust_trace.provenance import ( + FORMAT, + ProvenanceError, + build_record, + sign_record, + tool_catalog_hash, + verify_record, +) +from agentrust_trace.sign import generate_key, key_to_jwk + +DIGEST = "sha256:" + "a" * 64 +SPKI = "sha256:" + "b" * 64 +TOOLS = [{"name": "search", "description": "search", "input_schema": {"type": "object"}}] +BAD_LOCATORS: tuple[Any, ...] = (True, 1, [1], {"x": 1}) + + +def _base_record(identity: dict[str, Any]) -> dict[str, Any]: + return { + "format": FORMAT, + "kind": "publisher-asserted", + "issued_at": int(time.time()), + "identity": identity, + "publisher": "did:web:acme.example", + "tool_catalog": {"hash": tool_catalog_hash(TOOLS), "tool_count": len(TOOLS)}, + "attestation": None, + } + + +@pytest.mark.parametrize("package", BAD_LOCATORS, ids=repr) +def test_builder_refuses_non_string_artifact_package(package: Any) -> None: + with pytest.raises(ProvenanceError, match="artifact.package must be a non-empty string"): + build_record( + kind="publisher-asserted", + publisher="did:web:acme.example", + tools=TOOLS, + artifact={"package": package, "digest": DIGEST}, # type: ignore[dict-item] + ) + + +@pytest.mark.parametrize("url", BAD_LOCATORS, ids=repr) +def test_builder_refuses_non_string_endpoint_url(url: Any) -> None: + with pytest.raises(ProvenanceError, match="endpoint.url must be a non-empty string"): + build_record( + kind="publisher-asserted", + publisher="did:web:acme.example", + tools=TOOLS, + endpoint={"url": url, "spki_sha256": SPKI}, # type: ignore[dict-item] + ) + + +@pytest.mark.parametrize("package", BAD_LOCATORS, ids=repr) +def test_verifier_refuses_signed_non_string_artifact_package(package: Any) -> None: + key = generate_key() + signed = sign_record( + _base_record({"artifact": {"package": package, "digest": DIGEST}}), key + ) + with pytest.raises(ProvenanceError, match="artifact.package must be a non-empty string"): + verify_record(signed, key_to_jwk(key)) + + +@pytest.mark.parametrize("url", BAD_LOCATORS, ids=repr) +def test_verifier_refuses_signed_non_string_endpoint_url(url: Any) -> None: + key = generate_key() + signed = sign_record( + _base_record({"endpoint": {"url": url, "spki_sha256": SPKI}}), key + ) + with pytest.raises(ProvenanceError, match="endpoint.url must be a non-empty string"): + verify_record(signed, key_to_jwk(key)) + + +def test_textual_locator_controls_still_pass() -> None: + artifact = build_record( + kind="publisher-asserted", + publisher="did:web:acme.example", + tools=TOOLS, + artifact={"package": "pkg:npm/%40acme/mcp-search@2.1.0", "digest": DIGEST}, + ) + endpoint = build_record( + kind="publisher-asserted", + publisher="did:web:acme.example", + tools=TOOLS, + endpoint={"url": "https://mcp.acme.example/", "spki_sha256": SPKI}, + ) + assert artifact["identity"]["artifact"]["package"].startswith("pkg:") + assert endpoint["identity"]["endpoint"]["url"].startswith("https://")