Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 48 additions & 3 deletions src/cmcp_verify/verify.py
Original file line number Diff line number Diff line change
Expand Up @@ -662,9 +662,8 @@ def verify_audit_bundle(
transcript_hash = transcript.get("hash")
chain_tip = chain.get("tip")

bundle_has_tool_calls = any(
entry.get("entry_type") == "tool_call" for entry in entries
)
tool_calls = [entry for entry in entries if entry.get("entry_type") == "tool_call"]
bundle_has_tool_calls = bool(tool_calls)
if bundle_has_tool_calls and not isinstance(transcript_hash, str):
failures.append(
"claim trace.tool_transcript.hash is missing for a bundle with tool calls"
Expand All @@ -679,6 +678,52 @@ def verify_audit_bundle(
failures.append(
"claim trace.tool_transcript.hash does not match gateway.audit_chain.tip"
)

call_summary = claim_json.get("gateway", {}).get("call_summary", {})
expected_call_summary = {
"tool_calls_total": len(tool_calls),
"tool_calls_allowed": sum(
1 for entry in tool_calls if entry.get("policy_decision") == "allow"
),
"tool_calls_denied": sum(
1
for entry in tool_calls
if entry.get("policy_decision") in ("deny", "advisory_deny")
),
"tool_calls_faulted": sum(
1 for entry in tool_calls if entry.get("policy_decision") == "fault"
),
"tools_invoked": sorted(
{
entry["tool_name"]
for entry in tool_calls
if entry.get("tool_name") is not None
}
),
}
actual_call_count = transcript.get("call_count")
if (
not isinstance(actual_call_count, int)
or isinstance(actual_call_count, bool)
or actual_call_count != len(tool_calls)
):
failures.append(
"claim trace.tool_transcript.call_count does not match audit bundle tool calls"
)
for field_name, expected_value in expected_call_summary.items():
actual_value = call_summary.get(field_name)
invalid_integer = (
isinstance(expected_value, int)
and (
not isinstance(actual_value, int)
or isinstance(actual_value, bool)
)
)
if invalid_integer or actual_value != expected_value:
failures.append(
f"claim gateway.call_summary.{field_name} does not match audit bundle tool calls"
)

if chain.get("root") != entries[0].get("entry_hash"):
failures.append("bundle root does not match claim gateway.audit_chain.root")
if chain.get("tip") != entries[-1].get("entry_hash"):
Expand Down
19 changes: 12 additions & 7 deletions tests/conformance/test_audit_conformance.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,13 +40,18 @@ def _report():
)


def _summary():
def _summary(chain):
tool_calls = [entry for entry in chain.entries if entry.entry_type == "tool_call"]
return CallSummary(
tool_calls_total=1,
tool_calls_allowed=1,
tool_calls_denied=0,
tool_calls_faulted=0,
tools_invoked=["tool.a"],
tool_calls_total=len(tool_calls),
tool_calls_allowed=sum(1 for entry in tool_calls if entry.policy_decision == "allow"),
tool_calls_denied=sum(
1 for entry in tool_calls if entry.policy_decision in ("deny", "advisory_deny")
),
tool_calls_faulted=sum(1 for entry in tool_calls if entry.policy_decision == "fault"),
tools_invoked=sorted(
{entry.tool_name for entry in tool_calls if entry.tool_name is not None}
),
session_max_sensitivity="public",
call_graph_summary=CallGraphSummary(
compliance_domains_touched=["external"],
Expand All @@ -66,7 +71,7 @@ def _claim_dict(chain, key, seq=1, prev=None):
policy_version="1.0.0",
),
tool_catalog=ToolCatalogInfo(hash="sha256:" + "1" * 64),
call_summary=_summary(),
call_summary=_summary(chain),
audit_chain_root=chain.chain_root,
audit_chain_tip=chain.chain_tip,
audit_chain_length=chain.length,
Expand Down
112 changes: 111 additions & 1 deletion tests/unit/test_verify_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,25 @@ def test_verify_fails_on_tampered_claim(claim_and_bundle, tmp_path):
def test_verify_audit_bundle_passes(claim_and_bundle):
# The audit bundle itself verifies (PASS), but the software-only claim is
# only partially_verified, so the overall CLI result is still FAIL.
claim_file, bundle_file, _, _, _ = claim_and_bundle
claim_file, bundle_file, claim, _, _ = claim_and_bundle
assert claim["trace"]["tool_transcript"]["call_count"] == 1
call_summary = claim["gateway"]["call_summary"]
assert {
field: call_summary[field]
for field in (
"tool_calls_total",
"tool_calls_allowed",
"tool_calls_denied",
"tool_calls_faulted",
"tools_invoked",
)
} == {
"tool_calls_total": 1,
"tool_calls_allowed": 1,
"tool_calls_denied": 0,
"tool_calls_faulted": 0,
"tools_invoked": ["fixture-tool"],
}
result = CliRunner().invoke(main, [
"verify", str(claim_file), "--audit-bundle", str(bundle_file),
])
Expand Down Expand Up @@ -192,6 +210,98 @@ def test_verify_rejects_removed_tool_transcript(claim_and_bundle, tmp_path):
assert "tool_transcript.hash is missing for a bundle with tool calls" in result.output


@pytest.mark.parametrize(
("section", "field_name", "tampered_value", "expected_error"),
[
(
"tool_transcript",
"call_count",
7,
"trace.tool_transcript.call_count does not match audit bundle tool calls",
),
(
"call_summary",
"tool_calls_total",
7,
"gateway.call_summary.tool_calls_total does not match audit bundle tool calls",
),
(
"call_summary",
"tool_calls_allowed",
0,
"gateway.call_summary.tool_calls_allowed does not match audit bundle tool calls",
),
(
"call_summary",
"tool_calls_denied",
1,
"gateway.call_summary.tool_calls_denied does not match audit bundle tool calls",
),
(
"call_summary",
"tool_calls_faulted",
1,
"gateway.call_summary.tool_calls_faulted does not match audit bundle tool calls",
),
(
"call_summary",
"tools_invoked",
["substituted.tool"],
"gateway.call_summary.tools_invoked does not match audit bundle tool calls",
),
(
"call_summary",
"tool_calls_total",
True,
"gateway.call_summary.tool_calls_total does not match audit bundle tool calls",
),
(
"call_summary",
"tool_calls_denied",
False,
"gateway.call_summary.tool_calls_denied does not match audit bundle tool calls",
),
],
)
def test_verify_rejects_call_summary_mismatch(
claim_and_bundle,
tmp_path,
section,
field_name,
tampered_value,
expected_error,
):
"""A re-signed claim must bind audit-derived call metadata to the bundle."""
_, bundle_file, claim, _, signing_key = claim_and_bundle

if section == "tool_transcript":
claim["trace"][section][field_name] = tampered_value
else:
claim["gateway"][section][field_name] = tampered_value

body = {k: v for k, v in claim.items() if k != "signature"}
body_bytes = json.dumps(
body,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
).encode()
raw_sig = signing_key.sign(body_bytes)
claim["signature"] = base64.urlsafe_b64encode(raw_sig).rstrip(b"=").decode()

tampered_claim = tmp_path / f"{field_name}-mismatch.json"
tampered_claim.write_text(json.dumps(claim))

result = CliRunner().invoke(main, [
"verify", str(tampered_claim), "--audit-bundle", str(bundle_file),
])

assert result.exit_code == 1, result.output
assert "signature PASS" in result.output
assert "audit_bundle FAIL" in result.output
assert expected_error in result.output


def test_verify_fails_on_tampered_audit_bundle(claim_and_bundle, tmp_path):
"""Mutating one audit entry breaks the hash chain and the bundle signature."""
claim_file, _, _, bundle, _ = claim_and_bundle
Expand Down