diff --git a/src/cmcp_verify/verify.py b/src/cmcp_verify/verify.py index 96f6ebe..8f7cdbc 100644 --- a/src/cmcp_verify/verify.py +++ b/src/cmcp_verify/verify.py @@ -539,6 +539,15 @@ def _external_evidence_failure(entry_index: int, reason: str) -> str: return f"entry {entry_index}: {_EXTERNAL_EVIDENCE_ERROR}: {reason}" +def _audit_bundle_shape_failure(path: str, entry_count: int = 0) -> AuditBundleResult: + """Classify malformed external structure without interpreting its contents.""" + return AuditBundleResult( + verified=False, + entry_count=entry_count, + failures=[f"{path} has invalid object or array shape"], + ) + + def verify_audit_bundle( bundle_json: dict[str, Any], claim_json: dict[str, Any] | None = None, @@ -560,8 +569,39 @@ def verify_audit_bundle( signature). This is opt-in: receipt-less entries and callers that do not supply keys are unaffected, so existing evidence keeps verifying. """ - failures: list[str] = [] + if not isinstance(bundle_json, dict): + return _audit_bundle_shape_failure("bundle") + entries = bundle_json.get("entries", []) + if not isinstance(entries, list): + return _audit_bundle_shape_failure("bundle.entries") + for i, entry in enumerate(entries): + if not isinstance(entry, dict): + return _audit_bundle_shape_failure(f"bundle.entries[{i}]", len(entries)) + + if claim_json is not None: + if not isinstance(claim_json, dict): + return _audit_bundle_shape_failure("claim", len(entries)) + for path in ( + ("gateway",), + ("gateway", "audit_chain"), + ("gateway", "call_summary"), + ("trace",), + ("trace", "tool_transcript"), + ("trace", "cnf"), + ("trace", "cnf", "jwk"), + ): + obj: Any = claim_json + for depth, name in enumerate(path): + if name not in obj: + break + obj = obj[name] + if not isinstance(obj, dict): + return _audit_bundle_shape_failure( + "claim." + ".".join(path[: depth + 1]), len(entries) + ) + + failures: list[str] = [] if not entries: return AuditBundleResult(verified=False, entry_count=0, failures=["bundle has no entries"]) diff --git a/tests/unit/test_verify_audit_bundle_malformed_shapes.py b/tests/unit/test_verify_audit_bundle_malformed_shapes.py new file mode 100644 index 0000000..26e4c3b --- /dev/null +++ b/tests/unit/test_verify_audit_bundle_malformed_shapes.py @@ -0,0 +1,114 @@ +"""Malformed audit-bundle boundary vectors for issue #593.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +import pytest + +from cmcp_verify.verify import AuditBundleResult, verify_audit_bundle + + +def _one_entry_bundle() -> dict[str, Any]: + body = { + "entry_type": "session", + "call_id": "call-1", + "prev_entry_hash": "genesis", + } + entry_hash = hashlib.sha256( + json.dumps( + body, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode() + ).hexdigest() + return {"entries": [{**body, "entry_hash": entry_hash}]} + + +@pytest.mark.parametrize( + ("entries", "failure_path"), + [ + ("bad", "bundle.entries"), + (1, "bundle.entries"), + (True, "bundle.entries"), + ({"unexpected": "object"}, "bundle.entries"), + (["bad"], "bundle.entries[0]"), + ([1], "bundle.entries[0]"), + ([True], "bundle.entries[0]"), + ([[]], "bundle.entries[0]"), + ([None], "bundle.entries[0]"), + ], +) +def test_malformed_entries_return_failed_result(entries: Any, failure_path: str) -> None: + result = verify_audit_bundle({"entries": entries}) + + assert result == AuditBundleResult( + verified=False, + entry_count=len(entries) if isinstance(entries, list) else 0, + failures=[f"{failure_path} has invalid object or array shape"], + ) + + +@pytest.mark.parametrize( + ("claim", "failure_path"), + [ + ({"gateway": "bad"}, "claim.gateway"), + ({"gateway": {"audit_chain": "bad"}}, "claim.gateway.audit_chain"), + ({"gateway": {"call_summary": "bad"}}, "claim.gateway.call_summary"), + ({"trace": "bad"}, "claim.trace"), + ({"trace": {"tool_transcript": "bad"}}, "claim.trace.tool_transcript"), + ({"trace": {"cnf": "bad"}}, "claim.trace.cnf"), + ({"trace": {"cnf": {"jwk": "bad"}}}, "claim.trace.cnf.jwk"), + ], +) +def test_malformed_claim_binding_shapes_return_failed_result( + claim: dict[str, Any], failure_path: str +) -> None: + result = verify_audit_bundle(_one_entry_bundle(), claim) + + assert result == AuditBundleResult( + verified=False, + entry_count=1, + failures=[f"{failure_path} has invalid object or array shape"], + ) + + +def test_malformed_bundle_root_returns_failed_result() -> None: + result = verify_audit_bundle(None) # type: ignore[arg-type] + + assert result == AuditBundleResult( + verified=False, + entry_count=0, + failures=["bundle has invalid object or array shape"], + ) + + +def test_malformed_claim_root_returns_failed_result() -> None: + result = verify_audit_bundle(_one_entry_bundle(), "bad") # type: ignore[arg-type] + + assert result == AuditBundleResult( + verified=False, + entry_count=1, + failures=["claim has invalid object or array shape"], + ) + + +def test_missing_entries_preserves_existing_failure() -> None: + result = verify_audit_bundle({}) + + assert result == AuditBundleResult( + verified=False, + entry_count=0, + failures=["bundle has no entries"], + ) + + +def test_valid_entry_preserves_success() -> None: + assert verify_audit_bundle(_one_entry_bundle()) == AuditBundleResult( + verified=True, + entry_count=1, + failures=[], + )