test(verify): reproduce malformed audit-bundle shape escapes

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"])
 

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=[],
+    )