#!/usr/bin/env python3 """Tests for the bounded external MCP replay controller without mocks.""" from __future__ import annotations import hashlib import importlib.util import io import json import sys import tarfile import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[3] CONTROLLER_PATH = ROOT / "scripts/security/external_mcp_replay_controller.py" POLICY_PATH = ROOT / "config/mcp/playwright-mcp-replay-policy.json" ARTIFACT_POLICY_PATH = ROOT / "config/mcp/playwright-mcp-artifact-policy.json" WORKFLOW_PATH = ROOT / ".gitea/workflows/mcp-external-replay.yaml" AUDIT_WRITEBACK_PATH = ROOT / "scripts/ci/write-gitea-audit-receipts.sh" SPEC = importlib.util.spec_from_file_location("external_mcp_replay_controller", CONTROLLER_PATH) assert SPEC and SPEC.loader controller = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = controller SPEC.loader.exec_module(controller) def _canonical(payload: object) -> bytes: return json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def _write_json(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 _artifact_receipts(policy_payload: dict) -> tuple[dict, dict]: artifact = policy_payload["artifact_source"] artifact_run_id = "11111111-2222-4333-8444-555555555555" artifact_trace_id = f"mcp-artifact-{artifact_run_id}" packages = [] for key, digest in artifact["package_integrities"].items(): name, version = key.rsplit("@", 1) packages.append( { "name": name, "version": version, "tarball_sha512_hex": digest, } ) artifact_controller = { "schema_version": "awoooi_external_mcp_artifact_receipt_v1", "status": "completed_internal_mirror_pending_independent_verifier", "run_id": artifact_run_id, "trace_id": artifact_trace_id, "promotion_allowed": False, "replay_allowed": False, "shadow_allowed": False, "canary_allowed": False, "internal_mirror_ref": artifact["artifact_ref"], "runtime_mirror_ref": artifact["artifact_runtime_ref"], "policy_checksum": artifact["policy_checksum"], "bundle_manifest_sha256": artifact["bundle_manifest_sha256"], "cyclonedx_sbom_sha256": artifact["cyclonedx_sbom_sha256"], "packages": packages, } artifact_verifier = { "schema_version": "awoooi_external_mcp_artifact_verifier_receipt_v1", "status": "completed_internal_mirror_verified", "run_id": artifact_run_id, "trace_id": artifact_trace_id, "promotion_allowed": False, "replay_allowed": False, "shadow_allowed": False, "canary_allowed": False, "internal_mirror_ref": artifact["artifact_ref"], "runtime_mirror_ref": artifact["artifact_runtime_ref"], "policy_checksum": artifact["policy_checksum"], "independent_post_verifier": {"status": "passed"}, } return artifact_controller, artifact_verifier def _source_fixture(directory: Path) -> tuple[Path, Path, Path]: payload = _policy_payload() artifact_controller, artifact_verifier = _artifact_receipts(payload) controller_path = directory / "artifact-controller.json" verifier_path = directory / "artifact-verifier.json" controller_raw = _write_json(controller_path, artifact_controller) artifact_verifier["controller_receipt_sha256"] = hashlib.sha256( controller_raw ).hexdigest() verifier_raw = _write_json(verifier_path, artifact_verifier) payload["artifact_source"]["controller_receipt_sha256"] = hashlib.sha256( controller_raw ).hexdigest() payload["artifact_source"]["verifier_receipt_sha256"] = hashlib.sha256( verifier_raw ).hexdigest() policy_path = directory / "replay-policy.json" _write_json(policy_path, payload) return policy_path, controller_path, verifier_path class ReplayPolicyTests(unittest.TestCase): def _mutated_policy(self, mutate) -> Path: self.temp = tempfile.TemporaryDirectory() path = Path(self.temp.name) / "policy.json" payload = _policy_payload() mutate(payload) _write_json(path, payload) return path def tearDown(self) -> None: if hasattr(self, "temp"): self.temp.cleanup() def test_committed_policy_is_exact_and_minimal(self) -> None: policy = controller.load_policy(POLICY_PATH) self.assertEqual(policy.work_item_id, "MCP-REPLAY-PLAYWRIGHT-001") self.assertEqual(policy.risk_class, "high") self.assertEqual(policy.tool_count, 24) self.assertEqual(len(policy.adapter_allowed_tools), 4) self.assertEqual(len(policy.adapter_denied_tools), 20) self.assertRegex(policy.artifact_ref, r"@sha256:[0-9a-f]{64}$") self.assertRegex(policy.runtime_image_ref, r"@sha256:[0-9a-f]{64}$") def test_expired_policy_fails_closed(self) -> None: path = self._mutated_policy( lambda payload: payload.__setitem__( "expires_at", "2026-07-01T00:00:00+08:00" ) ) with self.assertRaisesRegex(controller.ReplayError, "replay_policy_expired"): controller.load_policy(path) def test_network_must_remain_none(self) -> None: path = self._mutated_policy( lambda payload: payload["runtime"].__setitem__("network_mode", "bridge") ) with self.assertRaisesRegex(controller.ReplayError, "runtime_isolation_invalid"): controller.load_policy(path) def test_runtime_must_remain_non_root(self) -> None: path = self._mutated_policy( lambda payload: payload["runtime"].__setitem__("user", "0:0") ) with self.assertRaisesRegex(controller.ReplayError, "runtime_isolation_invalid"): controller.load_policy(path) def test_runtime_memory_limit_cannot_be_weakened(self) -> None: path = self._mutated_policy( lambda payload: payload["runtime"].__setitem__("memory_limit", "2g") ) with self.assertRaisesRegex(controller.ReplayError, "runtime_isolation_invalid"): controller.load_policy(path) def test_runtime_argument_order_is_part_of_allowlist(self) -> None: def mutate(payload: dict) -> None: arguments = payload["runtime"]["arguments"] arguments[2], arguments[4] = arguments[4], arguments[2] path = self._mutated_policy(mutate) with self.assertRaisesRegex(controller.ReplayError, "arguments_not_allowlisted"): controller.load_policy(path) def test_floating_runtime_reference_is_rejected(self) -> None: path = self._mutated_policy( lambda payload: payload["runtime"].__setitem__( "image_ref", "registry.wooo.work/awoooi/ci-runner:current" ) ) with self.assertRaisesRegex(controller.ReplayError, "runtime_image_ref_invalid"): controller.load_policy(path) def test_direct_registration_cannot_be_enabled(self) -> None: path = self._mutated_policy( lambda payload: payload["execution_boundaries"].__setitem__( "direct_gateway_registration_allowed", True ) ) with self.assertRaisesRegex(controller.ReplayError, "execution_boundary_invalid"): controller.load_policy(path) def test_candidate_identity_cannot_be_retargeted(self) -> None: path = self._mutated_policy( lambda payload: payload.__setitem__("candidate_id", "external.other") ) with self.assertRaisesRegex(controller.ReplayError, "replay_asset_identity_invalid"): controller.load_policy(path) def test_dangerous_tool_cannot_leave_denied_partition(self) -> None: def mutate(payload: dict) -> None: payload["protocol_contract"]["adapter_denied_tools"].remove( "browser_run_code_unsafe" ) path = self._mutated_policy(mutate) with self.assertRaises(controller.ReplayError): controller.load_policy(path) class SourceEvidenceTests(unittest.TestCase): def test_verified_source_receipt_chain_passes(self) -> None: with tempfile.TemporaryDirectory() as temp: policy_path, controller_path, verifier_path = _source_fixture(Path(temp)) policy = controller.load_policy(policy_path) evidence = controller.validate_source_evidence( policy, artifact_policy_path=ARTIFACT_POLICY_PATH, controller_receipt_path=controller_path, verifier_receipt_path=verifier_path, ) self.assertEqual( evidence.artifact_run_id, "11111111-2222-4333-8444-555555555555" ) self.assertRegex(evidence.artifact_verifier_receipt_sha256, r"^[0-9a-f]{64}$") def test_tampered_controller_receipt_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as temp: policy_path, controller_path, verifier_path = _source_fixture(Path(temp)) policy = controller.load_policy(policy_path) controller_path.write_bytes(controller_path.read_bytes() + b" ") with self.assertRaisesRegex( controller.ReplayError, "artifact_controller_receipt_checksum_mismatch", ): controller.validate_source_evidence( policy, artifact_policy_path=ARTIFACT_POLICY_PATH, controller_receipt_path=controller_path, verifier_receipt_path=verifier_path, ) def test_artifact_policy_checksum_drift_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) policy_path, controller_path, verifier_path = _source_fixture(root) artifact_policy = json.loads(ARTIFACT_POLICY_PATH.read_text()) artifact_policy["risk_level"] = "medium" drift_path = root / "artifact-policy.json" _write_json(drift_path, artifact_policy) with self.assertRaisesRegex( controller.ReplayError, "artifact_policy_checksum_mismatch" ): controller.validate_source_evidence( controller.load_policy(policy_path), artifact_policy_path=drift_path, controller_receipt_path=controller_path, verifier_receipt_path=verifier_path, ) class ArchiveAndProtocolTests(unittest.TestCase): def test_archive_traversal_is_rejected(self) -> None: with self.assertRaisesRegex(controller.ReplayError, "artifact_archive_path_invalid"): controller._safe_relative_member("package/../../escape") def test_archive_symlink_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) tarball = root / "bad.tgz" with tarfile.open(tarball, "w:gz") as archive: member = tarfile.TarInfo("package/link") member.type = tarfile.SYMTYPE member.linkname = "/etc/passwd" archive.addfile(member) with self.assertRaisesRegex( controller.ReplayError, "artifact_archive_member_type_invalid" ): controller._extract_package(tarball, root / "out") def test_regular_archive_extracts_with_exact_size(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) tarball = root / "good.tgz" content = b'{"name":"fixture"}' with tarfile.open(tarball, "w:gz") as archive: member = tarfile.TarInfo("package/package.json") member.size = len(content) archive.addfile(member, io.BytesIO(content)) files, size = controller._extract_package(tarball, root / "out") self.assertEqual((files, size), (1, len(content))) self.assertEqual((root / "out/package.json").read_bytes(), content) def test_tool_hashes_are_order_independent(self) -> None: tools = [ { "name": "beta", "inputSchema": {"type": "object"}, "annotations": {"readOnlyHint": True}, }, {"name": "alpha", "inputSchema": {"type": "object"}}, ] first = controller._tool_hashes(tools) second = controller._tool_hashes(list(reversed(tools))) self.assertEqual(first, second) self.assertEqual(first[0], 2) self.assertEqual(first[3], ("alpha", "beta")) def test_duplicate_tool_names_are_rejected(self) -> None: tools = [ {"name": "same", "inputSchema": {}}, {"name": "same", "inputSchema": {}}, ] with self.assertRaisesRegex(controller.ReplayError, "mcp_tool_names_not_unique"): controller._tool_hashes(tools) class ReceiptAndWorkflowTests(unittest.TestCase): def test_success_receipt_preserves_no_browser_no_write_boundary(self) -> None: policy = controller.load_policy(POLICY_PATH) source = controller.SourceEvidence("a", "b", "c", "d") observation = controller.ReplayObservation( policy.server_name, policy.server_version, policy.protocol_version, policy.tool_count, policy.tool_name_set_sha256, policy.tool_schema_manifest_sha256, 0, "stdin_eof_clean_exit", 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=10, package_unpacked_bytes=20, ) self.assertEqual( receipt["status"], "completed_protocol_schema_replay_pending_independent_verifier", ) 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_keeps_candidate_disabled(self) -> None: policy = controller.load_policy(POLICY_PATH) receipt = controller._failure_receipt( policy=policy, run_id="11111111-2222-4333-8444-555555555555", trace_id="mcp-replay-11111111-2222-4333-8444-555555555555", error_class="fixture_failure", ) self.assertEqual(receipt["status"], "blocked_with_safe_next_action") terminal = receipt["stage_receipts"]["rollback_or_no_write_terminal"] self.assertEqual(terminal["candidate_registration_state"], "disabled") self.assertIs(receipt["promotion_allowed"], False) def test_gitea_workflow_has_no_floating_or_external_source_controls(self) -> None: raw = WORKFLOW_PATH.read_text(encoding="utf-8").lower() forbidden = ( "uses" + ":", "actions" + "/checkout", "github" + ".com", "api" + ".github" + ".com", "raw" + ".githubusercontent" + ".com", "codeload" + ".github" + ".com", "ghcr" + ".io", "npx" + " -y", "@" + "latest", ":" + "latest", ) self.assertFalse([value for value in forbidden if value in raw]) self.assertIn("runs-on: awoooi-non110-host", raw) self.assertIn("wait-host-web-build-pressure.sh", raw) self.assertIn("mcp-replay-receipts", raw) self.assertIn("verify_external_mcp_replay_receipt.py", raw) self.assertIn("bash scripts/ci/write-gitea-audit-receipts.sh", raw) writeback = AUDIT_WRITEBACK_PATH.read_text(encoding="utf-8").lower() self.assertIn("git checkout --force --detach fetch_head", writeback) self.assertNotIn("git merge --no-edit fetch_head", writeback) self.assertIn( 'git push "${remote_name}" "head:refs/heads/${gitea_audit_branch}"', writeback, ) if __name__ == "__main__": unittest.main()