#!/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()