diff --git a/docs/tutorials/agt-adapter.md b/docs/tutorials/agt-adapter.md index ffae7ac9..5b2eb623 100644 --- a/docs/tutorials/agt-adapter.md +++ b/docs/tutorials/agt-adapter.md @@ -24,13 +24,11 @@ session = AGTSessionResult( adapter = TraceAGTAdapter( model_provider="example", model_id="synthetic-demo", build_provenance_digest="sha256:" + "e" * 64, - transparency="https://example.test/unused", enforcement_mode="declared", # synthetic input: no policy engine ran ) record = adapter.build_trust_record(session) assert record["policy"]["enforcement_mode"] == "declared" -# The adapter currently requires a URI argument but performs no registration. -record.pop("transparency") +assert "transparency" not in record # `appraisal.status` defaults to "none", which is correct here: synthetic input has not # been appraised. Pass appraisal_status only when an appraisal actually happened. key = generate_key() diff --git a/src/agentrust_trace/adapters/agt.py b/src/agentrust_trace/adapters/agt.py index 8223901b..a36ea089 100644 --- a/src/agentrust_trace/adapters/agt.py +++ b/src/agentrust_trace/adapters/agt.py @@ -11,6 +11,7 @@ from dataclasses import dataclass, field from typing import Any, Literal +from agentrust_trace._patterns import _PYTHON_PATTERNS from agentrust_trace.models import ( JCS_SAFE_INTEGER, Appraisal, @@ -27,6 +28,67 @@ # failure this check exists to stop -- so the floor is the contract's, not zero. TRACE_MIN_IAT = 1700000000 +_SUBJECT_RE = _PYTHON_PATTERNS[r"^(spiffe://[^/]+/.+|did:[a-z0-9]+:.+)$"] + + +def _check_agent_did(value: object) -> None: + if not isinstance(value, str): + raise ValueError(f"agent_did must be a string, got {type(value).__name__}") + if not _SUBJECT_RE.match(value): + raise ValueError( + f"agent_did {value!r} must be a SPIFFE URI " + "('spiffe:///') or a DID ('did::'). " + "It becomes the record subject, which is what a verifier keys on." + ) + + +def _check_policy_bundle_bytes(value: object) -> None: + if not isinstance(value, bytes): + raise ValueError( + f"policy_bundle_bytes must be bytes, got {type(value).__name__}" + ) + + +def _check_audit_entries(value: object) -> None: + if not isinstance(value, list) or any(not isinstance(entry, dict) for entry in value): + raise ValueError( + "audit_entries must be a list of dict objects, got " + f"{type(value).__name__}" + ) + + +def _check_merkle_chain_tip(value: object) -> None: + if not isinstance(value, str) or not value: + raise ValueError( + f"merkle_chain_tip must be a non-empty string, got {value!r}" + ) + try: + value.encode("utf-8") + except UnicodeEncodeError as exc: + raise ValueError( + "merkle_chain_tip must be valid UTF-8 without lone surrogates" + ) from exc + + +def _check_call_count(value: object) -> None: + if value is None: + return + if ( + not isinstance(value, int) + or isinstance(value, bool) + or value < 0 + or value > JCS_SAFE_INTEGER + ): + raise ValueError( + f"call_count must be an integer within [0, {JCS_SAFE_INTEGER}] or None, " + f"got {value!r}" + ) + + +def _check_transparency(value: object) -> None: + if value is not None and (not isinstance(value, str) or not value): + raise ValueError("transparency must be a non-empty URI string or None") + def _check_iat(value: object) -> None: """The bound ``TrustRecord.iat`` carries, applied to whatever is about to be signed. @@ -86,6 +148,11 @@ class AGTSessionResult: """Issuance timestamp. Defaults to now.""" def __post_init__(self) -> None: + _check_agent_did(self.agent_did) + _check_policy_bundle_bytes(self.policy_bundle_bytes) + _check_audit_entries(self.audit_entries) + _check_merkle_chain_tip(self.merkle_chain_tip) + _check_call_count(self.call_count) _check_iat(self.iat) @@ -125,7 +192,7 @@ def __init__( build_provenance_digest: str, build_provenance_builder: str | None = None, build_provenance_uri: str | None = None, - transparency: str, + transparency: str | None = None, appraisal_status: Literal["affirming", "warning", "contraindicated", "none"] = "none", appraisal_verifier: str = "https://agentrust-io.com/verify", appraisal_policy_ref: str | None = None, @@ -144,7 +211,10 @@ def __init__( being built, and stamping ``affirming`` on an unappraised record puts a verdict in the field a consumer reads to find out whether anybody checked. Set this only when an appraisal actually happened. + transparency: SCITT receipt URI. ``None`` means unanchored, which is correct + below Level 2 and leaves the key out of the record. """ + _check_transparency(transparency) self._model = ModelInfo( provider=model_provider, model_id=model_id, @@ -181,14 +251,24 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]: The ``cnf.jwk`` placeholder carries no key material until ``sign_record()`` populates it from the signing key. """ + if not isinstance(session, AGTSessionResult): + raise ValueError( + f"session must be an AGTSessionResult, got {type(session).__name__}" + ) + _check_agent_did(session.agent_did) + _check_policy_bundle_bytes(session.policy_bundle_bytes) + _check_audit_entries(session.audit_entries) + _check_merkle_chain_tip(session.merkle_chain_tip) + _check_call_count(session.call_count) + _check_iat(session.iat) + _check_transparency(self._transparency) + call_count = ( session.call_count if session.call_count is not None else len(session.audit_entries) ) - _check_iat(session.iat) - record: dict[str, Any] = { "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", "iat": session.iat, @@ -214,7 +294,6 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]: policy_ref=self._appraisal_policy_ref, timestamp=session.iat, ).model_dump(exclude_none=True), - "transparency": self._transparency, # cnf is populated by sign_record(); placeholder keeps schema valid "cnf": { "jwk": { @@ -224,6 +303,8 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]: }, }, } + if self._transparency is not None: + record["transparency"] = self._transparency return record # ------------------------------------------------------------------ @@ -232,14 +313,17 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]: @staticmethod def _bundle_hash(policy_bundle_bytes: bytes) -> str: + _check_policy_bundle_bytes(policy_bundle_bytes) return "sha256:" + hashlib.sha256(policy_bundle_bytes).hexdigest() @staticmethod def _transcript_hash(audit_entries: list[dict[str, Any]]) -> str: # RFC 8785 (JCS): same canonicalization TraceSandboxAdapter already uses # for the same field, and what docs/schema.md calls "canonical JSON". + _check_audit_entries(audit_entries) return "sha256:" + hashlib.sha256(_canonical_bytes(audit_entries)).hexdigest() @staticmethod def _measurement(merkle_chain_tip: str) -> str: + _check_merkle_chain_tip(merkle_chain_tip) return "sha256:" + hashlib.sha256(merkle_chain_tip.encode()).hexdigest() diff --git a/src/agentrust_trace/adapters/sandbox.py b/src/agentrust_trace/adapters/sandbox.py index af618f2c..bf0c5e07 100644 --- a/src/agentrust_trace/adapters/sandbox.py +++ b/src/agentrust_trace/adapters/sandbox.py @@ -130,6 +130,71 @@ def _check_iat(value: object) -> None: _DIGEST_RE = _PYTHON_PATTERNS[r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$"] +def _check_sandbox_id(value: object) -> None: + if not isinstance(value, str): + raise ValueError(f"sandbox_id must be a string, got {type(value).__name__}") + if not _SUBJECT_RE.match(value): + raise ValueError( + f"sandbox_id {value!r} must be a SPIFFE URI " + "('spiffe:///') or a DID ('did::'). " + "It becomes the record subject, which is what a verifier keys on." + ) + + +def _check_image_digest(value: object) -> None: + if not isinstance(value, str): + raise ValueError( + f"image_digest must be a string, got {type(value).__name__}" + ) + if not _DIGEST_RE.match(value): + raise ValueError( + f"image_digest {value!r} must be a sha256: or sha384: digest." + ) + + +def _check_policy_bundle_bytes(value: object) -> None: + if not isinstance(value, bytes): + raise ValueError( + f"policy_bundle_bytes must be bytes, got {type(value).__name__}" + ) + + +def _check_decisions(value: object) -> None: + if not isinstance(value, list) or any(not isinstance(d, dict) for d in value): + raise ValueError( + f"decisions must be a list of dict objects, got {type(value).__name__}" + ) + + +def _check_call_count(value: object) -> None: + if value is None: + return + if ( + not isinstance(value, int) + or isinstance(value, bool) + or value < 0 + or value > JCS_SAFE_INTEGER + ): + raise ValueError( + f"call_count must be an integer within [0, {JCS_SAFE_INTEGER}] or None, " + f"got {value!r}" + ) + + +def _check_transparency(value: object) -> None: + if value is not None and (not isinstance(value, str) or not value): + raise ValueError("transparency must be a non-empty URI string or None") + + +def _check_bundle_hash(value: object) -> None: + if not isinstance(value, str): + raise ValueError(f"bundle_hash must be a string, got {type(value).__name__}") + if not _DIGEST_RE.match(value): + raise ValueError( + f"bundle_hash {value!r} must be a sha256: or sha384: digest." + ) + + @dataclass(frozen=True) class SandboxAttestation: """Hardware evidence for the machine a sandbox ran on. @@ -203,6 +268,16 @@ def __post_init__(self) -> None: ) +def _check_attestation(value: object) -> None: + if value is None: + return + if not isinstance(value, SandboxAttestation): + raise ValueError( + f"attestation must be a SandboxAttestation or None, got {type(value).__name__}" + ) + value.__post_init__() + + @dataclass(frozen=True) class SandboxSessionResult: """One sandbox session, as the runtime has it at close. @@ -238,22 +313,12 @@ class SandboxSessionResult: """Issuance timestamp. Defaults to now.""" def __post_init__(self) -> None: - if not isinstance(self.sandbox_id, str): - raise ValueError(f"sandbox_id must be a string, got {type(self.sandbox_id).__name__}") - if not _SUBJECT_RE.match(self.sandbox_id): - raise ValueError( - f"sandbox_id {self.sandbox_id!r} must be a SPIFFE URI " - "('spiffe:///') or a DID ('did::'). " - "It becomes the record subject, which is what a verifier keys on." - ) - if not isinstance(self.image_digest, str): - raise ValueError( - f"image_digest must be a string, got {type(self.image_digest).__name__}" - ) - if not _DIGEST_RE.match(self.image_digest): - raise ValueError( - f"image_digest {self.image_digest!r} must be a sha256: or sha384: digest." - ) + _check_sandbox_id(self.sandbox_id) + _check_image_digest(self.image_digest) + _check_policy_bundle_bytes(self.policy_bundle_bytes) + _check_decisions(self.decisions) + _check_attestation(self.attestation) + _check_call_count(self.call_count) _check_iat(self.iat) @@ -311,6 +376,7 @@ def __init__( build_provenance_slsa_level: Defaults to 0, meaning no provenance claim. Raise it only when a builder genuinely produced the attested level. """ + _check_transparency(transparency) self._model = ModelInfo( provider=model_provider, model_id=model_id, version=model_version ) @@ -346,11 +412,22 @@ def build_trust_record(self, session: SandboxSessionResult) -> dict[str, Any]: passed here. See the module docstring and :class:`SandboxAttestation` for this boundary. """ + if not isinstance(session, SandboxSessionResult): + raise ValueError( + f"session must be a SandboxSessionResult, got {type(session).__name__}" + ) + _check_sandbox_id(session.sandbox_id) + _check_image_digest(session.image_digest) + _check_policy_bundle_bytes(session.policy_bundle_bytes) + _check_decisions(session.decisions) + _check_attestation(session.attestation) + _check_call_count(session.call_count) + _check_iat(session.iat) + _check_transparency(self._transparency) + bundle_hash = self.bundle_hash(session.policy_bundle_bytes) runtime = self._runtime(session, bundle_hash) - _check_iat(session.iat) - record: dict[str, Any] = { "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", "iat": session.iat, @@ -404,6 +481,7 @@ def build_trust_record(self, session: SandboxSessionResult) -> dict[str, Any]: @staticmethod def bundle_hash(policy_bundle_bytes: bytes) -> str: """SHA-256 of the policy bundle bytes, exactly as supplied.""" + _check_policy_bundle_bytes(policy_bundle_bytes) return "sha256:" + hashlib.sha256(policy_bundle_bytes).hexdigest() @staticmethod @@ -416,6 +494,7 @@ def transcript_hash(decisions: list[dict[str, Any]]) -> str: pre-image (see :func:`~agentrust_trace.sign.sign_record`) keeps a record reproducible by an implementation in another language. """ + _check_decisions(decisions) return "sha256:" + hashlib.sha256(_canonical_bytes(decisions)).hexdigest() @staticmethod @@ -431,6 +510,8 @@ def software_measurement(image_digest: str, bundle_hash: str) -> str: ``platform: "software-only"``, which the model documents as never to be mistaken for hardware-backed evidence. """ + _check_image_digest(image_digest) + _check_bundle_hash(bundle_hash) return ( "sha256:" + hashlib.sha256(f"{image_digest}\n{bundle_hash}".encode()).hexdigest() diff --git a/tests/test_agt_adapter.py b/tests/test_agt_adapter.py index 5f1470e1..a666fb06 100644 --- a/tests/test_agt_adapter.py +++ b/tests/test_agt_adapter.py @@ -2,11 +2,13 @@ from __future__ import annotations +import hashlib +import inspect import json from dataclasses import FrozenInstanceError from pathlib import Path +from typing import Any -import hashlib import rfc8785 import pytest @@ -388,3 +390,134 @@ def test_the_adapter_floor_is_the_one_the_record_contract_carries() -> None: m for m in TrustRecord.model_fields["iat"].metadata if getattr(m, "ge", None) is not None ) assert bound.ge == TRACE_MIN_IAT + + +# --------------------------------------------------------------------------- +# Issue #534: optional transparency and session field validation +# --------------------------------------------------------------------------- + +def test_transparency_defaults_to_none_and_is_omitted_when_unanchored() -> None: + record = _make_adapter(transparency=None).build_trust_record(_make_session()) + assert "transparency" not in record + validate_json(record) + assert ( + inspect.signature(TraceAGTAdapter.__init__).parameters["transparency"].default + is None + ) + default_record = TraceAGTAdapter( + model_provider="example-provider", + model_id="example-model", + build_provenance_digest="sha256:" + "a" * 64, + enforcement_mode="enforce", + ).build_trust_record(_make_session()) + assert "transparency" not in default_record + validate_json(default_record) + + +@pytest.mark.parametrize( + "transparency", + ["", 123, True, ["https://example.org"], {"uri": "https://example.org"}], + ids=repr, +) +def test_transparency_rejects_empty_or_non_string(transparency: Any) -> None: + with pytest.raises(ValueError, match="transparency"): + _make_adapter(transparency=transparency) + + +@pytest.mark.parametrize( + "agent_did", + [ + "not-a-uri", + "", + "https://example.org/agent", + "spiffe://example.org", + "did:x", + 123, + None, + True, + ["spiffe://example.org/agent/a"], + ], + ids=repr, +) +def test_agent_did_rejects_invalid_or_non_string_values(agent_did: Any) -> None: + with pytest.raises(ValueError, match="agent_did"): + _make_session(agent_did=agent_did) + + +@pytest.mark.parametrize( + "bundle", + ["permit(principal, action, resource);", 123, None, True, [1, 2], {"a": 1}], + ids=repr, +) +def test_policy_bundle_bytes_rejects_non_bytes(bundle: Any) -> None: + with pytest.raises(ValueError, match="policy_bundle_bytes"): + _make_session(policy_bundle_bytes=bundle) + with pytest.raises(ValueError, match="policy_bundle_bytes"): + TraceAGTAdapter._bundle_hash(bundle) + + +@pytest.mark.parametrize( + "entries", + [ + "a-string", + {"tool": "x", "decision": "permit"}, + [1, 2], + ["entry"], + [None], + 123, + None, + True, + ], + ids=repr, +) +def test_audit_entries_rejects_non_list_or_non_dict_items(entries: Any) -> None: + with pytest.raises(ValueError, match="audit_entries"): + _make_session(audit_entries=entries) + with pytest.raises(ValueError, match="audit_entries"): + TraceAGTAdapter._transcript_hash(entries) + + +@pytest.mark.parametrize( + "tip", + ["", "\ud800", 123, None, True, b"deadbeef", ["tip"]], + ids=repr, +) +def test_merkle_chain_tip_rejects_empty_or_non_string(tip: Any) -> None: + with pytest.raises(ValueError, match="merkle_chain_tip"): + _make_session(merkle_chain_tip=tip) + with pytest.raises(ValueError, match="merkle_chain_tip"): + TraceAGTAdapter._measurement(tip) + + +@pytest.mark.parametrize( + "call_count", + [True, False, -1, "1", 1.5, JCS_SAFE_INTEGER + 1], + ids=repr, +) +def test_call_count_rejects_invalid_values(call_count: Any) -> None: + with pytest.raises(ValueError, match="call_count"): + _make_session(call_count=call_count) + + +def test_build_trust_record_revalidates_session_and_in_place_mutations() -> None: + adapter = _make_adapter() + with pytest.raises(ValueError, match="session must be an AGTSessionResult"): + adapter.build_trust_record(None) # type: ignore[arg-type] + + session = _make_session(audit_entries=[{"entry_id": 1, "decision": "permit"}]) + session.audit_entries.append("not-a-dict") # type: ignore[arg-type] + with pytest.raises(ValueError, match="audit_entries"): + adapter.build_trust_record(session) + + for field_name, bad_value in ( + ("agent_did", "not-a-uri"), + ("policy_bundle_bytes", "not-bytes"), + ("merkle_chain_tip", ""), + ("call_count", True), + ): + tampered = _make_session() + object.__setattr__(tampered, field_name, bad_value) + with pytest.raises(ValueError, match=field_name): + adapter.build_trust_record(tampered) + + diff --git a/tests/test_sandbox_required_string_fields.py b/tests/test_sandbox_required_string_fields.py index ea7d10c6..d0ca65b7 100644 --- a/tests/test_sandbox_required_string_fields.py +++ b/tests/test_sandbox_required_string_fields.py @@ -6,7 +6,12 @@ import pytest -from agentrust_trace.adapters.sandbox import SandboxAttestation, SandboxSessionResult +from agentrust_trace.adapters.sandbox import ( + SandboxAttestation, + SandboxSessionResult, + TraceSandboxAdapter, +) +from agentrust_trace.models import JCS_SAFE_INTEGER DIGEST = "sha256:" + "a" * 64 BAD_VALUES: tuple[Any, ...] = (True, 1, [1], {"x": 1}) @@ -57,3 +62,137 @@ def test_textual_controls_still_construct() -> None: ) assert attestation.platform == "tpm2" assert session.sandbox_id.startswith("spiffe://") + + +@pytest.mark.parametrize( + "policy_bundle_bytes", + ("policy", 123, None, True, [1, 2], {"x": 1}), + ids=repr, +) +def test_session_policy_bundle_bytes_refuses_non_bytes( + policy_bundle_bytes: Any, +) -> None: + with pytest.raises(ValueError, match="policy_bundle_bytes"): + SandboxSessionResult( + sandbox_id="spiffe://runtime.example.org/sandbox/test", + image_digest=DIGEST, + policy_bundle_bytes=policy_bundle_bytes, + decisions=[], + ) + with pytest.raises(ValueError, match="policy_bundle_bytes"): + TraceSandboxAdapter.bundle_hash(policy_bundle_bytes) + + +@pytest.mark.parametrize( + "decisions", + ( + "a-string", + {"tool": "x", "decision": "permit"}, + [1, 2], + ["entry"], + [None], + 123, + None, + True, + ), + ids=repr, +) +def test_session_decisions_refuses_non_list_or_non_dict_entries( + decisions: Any, +) -> None: + with pytest.raises(ValueError, match="decisions"): + SandboxSessionResult( + sandbox_id="spiffe://runtime.example.org/sandbox/test", + image_digest=DIGEST, + policy_bundle_bytes=b"policy", + decisions=decisions, + ) + with pytest.raises(ValueError, match="decisions"): + TraceSandboxAdapter.transcript_hash(decisions) + + +@pytest.mark.parametrize( + "call_count", + (True, False, -1, "1", 1.5, JCS_SAFE_INTEGER + 1), + ids=repr, +) +def test_session_call_count_refuses_invalid_values(call_count: Any) -> None: + with pytest.raises(ValueError, match="call_count"): + SandboxSessionResult( + sandbox_id="spiffe://runtime.example.org/sandbox/test", + image_digest=DIGEST, + policy_bundle_bytes=b"policy", + decisions=[], + call_count=call_count, + ) + + +@pytest.mark.parametrize( + "transparency", + ("", 123, True, [1], {"x": 1}), + ids=repr, +) +def test_adapter_transparency_refuses_empty_or_non_string( + transparency: Any, +) -> None: + with pytest.raises(ValueError, match="transparency"): + TraceSandboxAdapter( + model_provider="example-provider", + model_id="example-model", + enforcement_mode="declared", + transparency=transparency, + ) + + +@pytest.mark.parametrize("attestation", ("tpm2", 123, True, [1], {"platform": "tpm2"}), ids=repr) +def test_session_attestation_refuses_non_attestation_values(attestation: Any) -> None: + with pytest.raises(ValueError, match="attestation"): + SandboxSessionResult( + sandbox_id="spiffe://runtime.example.org/sandbox/test", + image_digest=DIGEST, + policy_bundle_bytes=b"policy", + decisions=[], + attestation=attestation, + ) + + +@pytest.mark.parametrize("bad_digest", ("not-a-digest", "", 123, None, True), ids=repr) +def test_software_measurement_refuses_invalid_digests(bad_digest: Any) -> None: + with pytest.raises(ValueError, match="image_digest"): + TraceSandboxAdapter.software_measurement(bad_digest, DIGEST) + with pytest.raises(ValueError, match="bundle_hash"): + TraceSandboxAdapter.software_measurement(DIGEST, bad_digest) + + +def test_sandbox_build_trust_record_revalidates_session_and_in_place_mutations() -> None: + adapter = TraceSandboxAdapter( + model_provider="example-provider", + model_id="example-model", + enforcement_mode="declared", + ) + with pytest.raises(ValueError, match="session must be a SandboxSessionResult"): + adapter.build_trust_record(None) # type: ignore[arg-type] + + session = SandboxSessionResult( + sandbox_id="spiffe://runtime.example.org/sandbox/test", + image_digest=DIGEST, + policy_bundle_bytes=b"policy", + decisions=[{"tool": "fs.read", "decision": "permit"}], + ) + session.decisions.append("not-a-dict") # type: ignore[arg-type] + with pytest.raises(ValueError, match="decisions"): + adapter.build_trust_record(session) + + attestation = SandboxAttestation(platform="tpm2", measurement=DIGEST) + tampered_session = SandboxSessionResult( + sandbox_id="spiffe://runtime.example.org/sandbox/test", + image_digest=DIGEST, + policy_bundle_bytes=b"policy", + decisions=[], + attestation=attestation, + ) + object.__setattr__(attestation, "platform", "software-only") + with pytest.raises(ValueError, match="SandboxAttestation.platform"): + adapter.build_trust_record(tampered_session) + +