346 lines
14 KiB
Python
346 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for the standalone external MCP replay verifier without mocks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
VERIFIER_PATH = ROOT / "scripts/security/verify_external_mcp_replay_receipt.py"
|
|
CONTROLLER_PATH = ROOT / "scripts/security/external_mcp_replay_controller.py"
|
|
POLICY_PATH = ROOT / "config/mcp/playwright-mcp-replay-policy.json"
|
|
|
|
|
|
def _load(name: str, path: Path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
verifier = _load("verify_external_mcp_replay_receipt", VERIFIER_PATH)
|
|
controller = _load("external_mcp_replay_controller_for_verifier_test", CONTROLLER_PATH)
|
|
|
|
|
|
def _canonical(payload: object) -> bytes:
|
|
return json.dumps(
|
|
payload,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
|
|
|
|
def _write(path: Path, payload: object) -> bytes:
|
|
raw = _canonical(payload) + b"\n"
|
|
path.write_bytes(raw)
|
|
return raw
|
|
|
|
|
|
def _policy_payload() -> dict:
|
|
return json.loads(POLICY_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _valid_controller_receipt() -> tuple[dict, dict, str]:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
source = controller.SourceEvidence(
|
|
artifact_run_id="aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
|
|
artifact_trace_id="mcp-artifact-aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
|
|
artifact_controller_receipt_sha256="a" * 64,
|
|
artifact_verifier_receipt_sha256="b" * 64,
|
|
)
|
|
observation = controller.ReplayObservation(
|
|
server_name=policy.server_name,
|
|
server_version=policy.server_version,
|
|
protocol_version=policy.protocol_version,
|
|
tool_count=policy.tool_count,
|
|
tool_name_set_sha256=policy.tool_name_set_sha256,
|
|
tool_schema_manifest_sha256=policy.tool_schema_manifest_sha256,
|
|
process_exit_code=0,
|
|
shutdown_mode="stdin_eof_clean_exit",
|
|
ephemeral_container_removed=True,
|
|
)
|
|
receipt = controller.build_receipt(
|
|
policy=policy,
|
|
source=source,
|
|
run_id="11111111-2222-4333-8444-555555555555",
|
|
trace_id="mcp-replay-11111111-2222-4333-8444-555555555555",
|
|
mode="apply",
|
|
observation=observation,
|
|
package_file_count=173,
|
|
package_unpacked_bytes=1_000,
|
|
)
|
|
payload, checksum = verifier.load_and_validate_policy(POLICY_PATH)
|
|
return payload, receipt, checksum
|
|
|
|
|
|
def _artifact_chain(policy: dict) -> tuple[dict, dict]:
|
|
artifact = policy["artifact_source"]
|
|
run_id = "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"
|
|
controller_receipt = {
|
|
"schema_version": "awoooi_external_mcp_artifact_receipt_v1",
|
|
"status": "completed_internal_mirror_pending_independent_verifier",
|
|
"run_id": run_id,
|
|
"trace_id": f"mcp-artifact-{run_id}",
|
|
"internal_mirror_ref": artifact["artifact_ref"],
|
|
"policy_checksum": artifact["policy_checksum"],
|
|
"promotion_allowed": False,
|
|
"replay_allowed": False,
|
|
"shadow_allowed": False,
|
|
"canary_allowed": False,
|
|
}
|
|
verifier_receipt = {
|
|
"schema_version": "awoooi_external_mcp_artifact_verifier_receipt_v1",
|
|
"status": "completed_internal_mirror_verified",
|
|
"run_id": run_id,
|
|
"trace_id": f"mcp-artifact-{run_id}",
|
|
"internal_mirror_ref": artifact["artifact_ref"],
|
|
"policy_checksum": artifact["policy_checksum"],
|
|
"promotion_allowed": False,
|
|
"replay_allowed": False,
|
|
"shadow_allowed": False,
|
|
"canary_allowed": False,
|
|
"independent_post_verifier": {
|
|
"status": "passed",
|
|
"container_or_mcp_started": False,
|
|
},
|
|
}
|
|
return controller_receipt, verifier_receipt
|
|
|
|
|
|
class StandalonePolicyTests(unittest.TestCase):
|
|
def test_committed_policy_passes_independent_validation(self) -> None:
|
|
policy, checksum = verifier.load_and_validate_policy(POLICY_PATH)
|
|
self.assertEqual(policy["work_item_id"], "MCP-REPLAY-PLAYWRIGHT-001")
|
|
self.assertRegex(checksum, r"^sha256:[0-9a-f]{64}$")
|
|
|
|
def test_any_enabled_execution_boundary_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
payload = _policy_payload()
|
|
payload["execution_boundaries"]["browser_start_allowed"] = True
|
|
path = Path(temp) / "policy.json"
|
|
_write(path, payload)
|
|
with self.assertRaisesRegex(verifier.VerifierError, "boundary_must_remain_false"):
|
|
verifier.load_and_validate_policy(path)
|
|
|
|
def test_adapter_tool_partition_overlap_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
payload = _policy_payload()
|
|
payload["protocol_contract"]["adapter_denied_tools"].append(
|
|
"browser_snapshot"
|
|
)
|
|
path = Path(temp) / "policy.json"
|
|
_write(path, payload)
|
|
with self.assertRaisesRegex(verifier.VerifierError, "adapter_tool_partition_invalid"):
|
|
verifier.load_and_validate_policy(path)
|
|
|
|
def test_tagged_artifact_reference_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
payload = _policy_payload()
|
|
payload["artifact_source"]["artifact_ref"] = (
|
|
"registry.wooo.work/awoooi/mcp-artifacts/playwright-mcp:0.0.78"
|
|
)
|
|
path = Path(temp) / "policy.json"
|
|
_write(path, payload)
|
|
with self.assertRaisesRegex(verifier.VerifierError, "artifact_ref_invalid"):
|
|
verifier.load_and_validate_policy(path)
|
|
|
|
def test_runtime_resource_limits_cannot_be_weakened(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
payload = _policy_payload()
|
|
payload["runtime"]["pids_limit"] = 512
|
|
path = Path(temp) / "policy.json"
|
|
_write(path, payload)
|
|
with self.assertRaisesRegex(verifier.VerifierError, "runtime_isolation_invalid"):
|
|
verifier.load_and_validate_policy(path)
|
|
|
|
def test_runtime_argument_order_drift_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
payload = _policy_payload()
|
|
payload["runtime"]["arguments"] = list(
|
|
reversed(payload["runtime"]["arguments"])
|
|
)
|
|
path = Path(temp) / "policy.json"
|
|
_write(path, payload)
|
|
with self.assertRaisesRegex(verifier.VerifierError, "runtime_isolation_invalid"):
|
|
verifier.load_and_validate_policy(path)
|
|
|
|
|
|
class ControllerReceiptTests(unittest.TestCase):
|
|
def test_valid_controller_receipt_passes_standalone_validation(self) -> None:
|
|
policy, receipt, checksum = _valid_controller_receipt()
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "controller.json"
|
|
raw = _write(path, receipt)
|
|
observed, observed_sha = verifier.validate_controller_receipt(
|
|
policy, checksum, path
|
|
)
|
|
self.assertEqual(observed["run_id"], receipt["run_id"])
|
|
self.assertEqual(observed_sha, hashlib.sha256(raw).hexdigest())
|
|
|
|
def test_controller_browser_start_is_rejected(self) -> None:
|
|
policy, receipt, checksum = _valid_controller_receipt()
|
|
receipt["browser_started"] = True
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "controller.json"
|
|
_write(path, receipt)
|
|
with self.assertRaisesRegex(verifier.VerifierError, "controller_boundary_invalid"):
|
|
verifier.validate_controller_receipt(policy, checksum, path)
|
|
|
|
def test_controller_missing_cleanup_terminal_is_rejected(self) -> None:
|
|
policy, receipt, checksum = _valid_controller_receipt()
|
|
receipt["stage_receipts"]["rollback_or_no_write_terminal"][
|
|
"ephemeral_container_removed"
|
|
] = False
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "controller.json"
|
|
_write(path, receipt)
|
|
with self.assertRaisesRegex(verifier.VerifierError, "controller_rollback_invalid"):
|
|
verifier.validate_controller_receipt(policy, checksum, path)
|
|
|
|
def test_controller_protocol_hash_drift_is_rejected(self) -> None:
|
|
policy, receipt, checksum = _valid_controller_receipt()
|
|
receipt["protocol_observation"]["tool_name_set_sha256"] = "0" * 64
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "controller.json"
|
|
_write(path, receipt)
|
|
with self.assertRaisesRegex(
|
|
verifier.VerifierError, "controller_protocol_observation_mismatch"
|
|
):
|
|
verifier.validate_controller_receipt(policy, checksum, path)
|
|
|
|
|
|
class ArtifactChainAndToolTests(unittest.TestCase):
|
|
def test_valid_artifact_chain_passes(self) -> None:
|
|
policy = _policy_payload()
|
|
artifact = policy["artifact_source"]
|
|
artifact_controller, artifact_verifier = _artifact_chain(policy)
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
controller_path = root / "artifact-controller.json"
|
|
verifier_path = root / "artifact-verifier.json"
|
|
controller_raw = _write(controller_path, artifact_controller)
|
|
artifact_verifier["controller_receipt_sha256"] = hashlib.sha256(
|
|
controller_raw
|
|
).hexdigest()
|
|
verifier_raw = _write(verifier_path, artifact_verifier)
|
|
artifact["controller_receipt_sha256"] = hashlib.sha256(
|
|
controller_raw
|
|
).hexdigest()
|
|
artifact["verifier_receipt_sha256"] = hashlib.sha256(
|
|
verifier_raw
|
|
).hexdigest()
|
|
observed = verifier.validate_artifact_receipt_chain(
|
|
policy, controller_path, verifier_path
|
|
)
|
|
self.assertEqual(observed[0], artifact["controller_receipt_sha256"])
|
|
self.assertEqual(observed[1], artifact["verifier_receipt_sha256"])
|
|
|
|
def test_artifact_chain_tamper_is_rejected(self) -> None:
|
|
policy = _policy_payload()
|
|
artifact = policy["artifact_source"]
|
|
artifact_controller, artifact_verifier = _artifact_chain(policy)
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
controller_path = root / "artifact-controller.json"
|
|
verifier_path = root / "artifact-verifier.json"
|
|
controller_raw = _write(controller_path, artifact_controller)
|
|
artifact_verifier["controller_receipt_sha256"] = hashlib.sha256(
|
|
controller_raw
|
|
).hexdigest()
|
|
verifier_raw = _write(verifier_path, artifact_verifier)
|
|
artifact["controller_receipt_sha256"] = hashlib.sha256(
|
|
controller_raw
|
|
).hexdigest()
|
|
artifact["verifier_receipt_sha256"] = hashlib.sha256(
|
|
verifier_raw
|
|
).hexdigest()
|
|
verifier_path.write_bytes(verifier_raw + b" ")
|
|
with self.assertRaisesRegex(verifier.VerifierError, "artifact_receipt_chain_invalid"):
|
|
verifier.validate_artifact_receipt_chain(
|
|
policy, controller_path, verifier_path
|
|
)
|
|
|
|
def test_independent_tool_hashes_are_order_independent(self) -> None:
|
|
tools = [
|
|
{"name": "zeta", "inputSchema": {"type": "object"}},
|
|
{
|
|
"name": "alpha",
|
|
"inputSchema": {"type": "object"},
|
|
"annotations": {"readOnlyHint": True},
|
|
},
|
|
]
|
|
first = verifier._independent_tool_hashes(tools)
|
|
second = verifier._independent_tool_hashes(list(reversed(tools)))
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(first[3], {"alpha", "zeta"})
|
|
|
|
def test_duplicate_tool_names_are_rejected(self) -> None:
|
|
tools = [
|
|
{"name": "duplicate", "inputSchema": {}},
|
|
{"name": "duplicate", "inputSchema": {}},
|
|
]
|
|
with self.assertRaisesRegex(verifier.VerifierError, "tool_names_not_unique"):
|
|
verifier._independent_tool_hashes(tools)
|
|
|
|
|
|
class VerifierReceiptTests(unittest.TestCase):
|
|
def test_completed_verifier_receipt_keeps_promotion_disabled(self) -> None:
|
|
policy, controller_receipt, policy_checksum = _valid_controller_receipt()
|
|
receipt = verifier.build_verifier_receipt(
|
|
policy=policy,
|
|
policy_checksum=policy_checksum,
|
|
controller=controller_receipt,
|
|
controller_sha256="c" * 64,
|
|
artifact_controller_sha256="a" * 64,
|
|
artifact_verifier_sha256="b" * 64,
|
|
observation={
|
|
"protocol_version": "2025-06-18",
|
|
"server_name": "Playwright",
|
|
"server_version": "1.62.0-alpha-1783623505000",
|
|
"tool_count": 24,
|
|
"tool_name_set_sha256": "d" * 64,
|
|
"tool_schema_manifest_sha256": "e" * 64,
|
|
"process_exit_code": 0,
|
|
"shutdown_mode": "stdin_eof_clean_exit",
|
|
"ephemeral_container_removed": True,
|
|
},
|
|
package_file_count=173,
|
|
package_unpacked_bytes=17_000_000,
|
|
)
|
|
self.assertEqual(receipt["status"], "completed_protocol_schema_replay_verified")
|
|
self.assertEqual(
|
|
receipt["independent_post_verifier"]["implementation"],
|
|
"standalone_no_controller_import",
|
|
)
|
|
for field in (
|
|
"browser_started",
|
|
"tool_call_performed",
|
|
"external_rag_write_performed",
|
|
"production_write_performed",
|
|
"promotion_allowed",
|
|
):
|
|
self.assertIs(receipt[field], False)
|
|
|
|
def test_failure_receipt_has_rollback_terminal(self) -> None:
|
|
policy = _policy_payload()
|
|
receipt = verifier._failure_receipt(policy, {}, "fixture_failure")
|
|
self.assertEqual(receipt["status"], "blocked_with_safe_next_action")
|
|
self.assertEqual(
|
|
receipt["rollback_terminal"]["candidate_registration_state"], "disabled"
|
|
)
|
|
self.assertIs(receipt["promotion_allowed"], False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|