#!/usr/bin/env python3 """Run a bounded, offline MCP protocol/schema compatibility replay. The replay consumes only an independently verified immutable artifact bundle and an immutable internal runtime image. It starts the MCP process with no network, performs initialize and tools/list, then terminates it. It never starts a browser, invokes an MCP tool, registers a provider, writes RAG, or mutates production. """ from __future__ import annotations import argparse import hashlib import json import os import re import select import shutil import subprocess import sys import tarfile import tempfile import uuid from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path, PurePosixPath from typing import Any POLICY_SCHEMA_VERSION = "awoooi_external_mcp_replay_policy_v1" RECEIPT_SCHEMA_VERSION = "awoooi_external_mcp_replay_receipt_v1" ARTIFACT_POLICY_SCHEMA_VERSION = "awoooi_external_mcp_artifact_policy_v1" ARTIFACT_CONTROLLER_SCHEMA_VERSION = "awoooi_external_mcp_artifact_receipt_v1" ARTIFACT_VERIFIER_SCHEMA_VERSION = ( "awoooi_external_mcp_artifact_verifier_receipt_v1" ) ARTIFACT_BUNDLE_SCHEMA_VERSION = "awoooi_external_mcp_artifact_bundle_v1" TRACE_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{3,160}$") SHA256_HEX_PATTERN = re.compile(r"^[0-9a-f]{64}$") SHA512_HEX_PATTERN = re.compile(r"^[0-9a-f]{128}$") OCI_REF_PATTERN = re.compile( r"^[a-z0-9.-]+(?::[0-9]+)?/[a-z0-9._/-]+@sha256:[0-9a-f]{64}$" ) GIT_SHA_PATTERN = re.compile(r"^[0-9a-f]{40}$") MAX_JSON_BYTES = 4 * 1024 * 1024 MAX_PROTOCOL_LINE_BYTES = 2 * 1024 * 1024 MAX_PACKAGE_UNPACKED_BYTES = 40 * 1024 * 1024 MAX_PACKAGE_FILES = 4_096 EXPECTED_PACKAGE_KEYS = { "@playwright/mcp@0.0.78", "playwright@1.62.0-alpha-1783623505000", "playwright-core@1.62.0-alpha-1783623505000", } class ReplayError(RuntimeError): """Fail-closed public-safe replay error class.""" @dataclass(frozen=True) class ReplayPolicy: work_item_id: str candidate_id: str adapter_id: str risk_class: str owner_lane: str expires_at: str artifact_policy_checksum: str audit_branch: str audit_commit: str controller_receipt_sha256: str verifier_receipt_sha256: str artifact_ref: str artifact_runtime_ref: str bundle_manifest_sha256: str cyclonedx_sbom_sha256: str package_integrities: dict[str, str] runtime_image_ref: str runtime_platform: str runtime_user: str runtime_entrypoint: tuple[str, ...] runtime_arguments: tuple[str, ...] pids_limit: int memory_limit: str cpu_limit: str tmpfs: str startup_timeout_seconds: int shutdown_timeout_seconds: int protocol_version: str server_name: str server_version: str tool_count: int tool_name_set_sha256: str tool_schema_manifest_sha256: str adapter_allowed_tools: tuple[str, ...] adapter_denied_tools: tuple[str, ...] promotion_blockers_after_replay: tuple[str, ...] checksum: str @dataclass(frozen=True) class SourceEvidence: artifact_run_id: str artifact_trace_id: str artifact_controller_receipt_sha256: str artifact_verifier_receipt_sha256: str @dataclass(frozen=True) class ReplayObservation: server_name: str server_version: str protocol_version: str tool_count: int tool_name_set_sha256: str tool_schema_manifest_sha256: str process_exit_code: int shutdown_mode: str ephemeral_container_removed: bool def _canonical_json_bytes(payload: Any) -> bytes: return json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def _sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _read_json_bytes(path: Path, error_class: str) -> tuple[bytes, dict[str, Any]]: try: raw = path.read_bytes() except OSError as exc: raise ReplayError(error_class) from exc if not raw or len(raw) > MAX_JSON_BYTES: raise ReplayError(error_class) try: payload = json.loads(raw) except json.JSONDecodeError as exc: raise ReplayError(error_class) from exc if not isinstance(payload, dict): raise ReplayError(error_class) return raw, payload def _require_dict(value: Any, error_class: str) -> dict[str, Any]: if not isinstance(value, dict): raise ReplayError(error_class) return value def _require_text(value: Any, error_class: str) -> str: if not isinstance(value, str) or not value.strip(): raise ReplayError(error_class) return value.strip() def _require_text_list(value: Any, error_class: str) -> tuple[str, ...]: if not isinstance(value, list) or not value: raise ReplayError(error_class) rows = tuple(_require_text(item, error_class) for item in value) if len(set(rows)) != len(rows): raise ReplayError(error_class) return rows def _require_positive_int(value: Any, error_class: str) -> int: if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise ReplayError(error_class) return value def _validate_sha256(value: Any, error_class: str, *, prefix: bool = False) -> str: text = _require_text(value, error_class) digest = text.removeprefix("sha256:") if prefix else text if not SHA256_HEX_PATTERN.fullmatch(digest): raise ReplayError(error_class) if prefix and not text.startswith("sha256:"): raise ReplayError(error_class) return text def _validate_oci_ref(value: Any, error_class: str) -> str: ref = _require_text(value, error_class) if not OCI_REF_PATTERN.fullmatch(ref): raise ReplayError(error_class) return ref def load_policy(path: Path) -> ReplayPolicy: raw, payload = _read_json_bytes(path, "replay_policy_unreadable") if payload.get("schema_version") != POLICY_SCHEMA_VERSION: raise ReplayError("replay_policy_schema_invalid") if payload.get("policy_version") != "1.0.0": raise ReplayError("replay_policy_version_invalid") if payload.get("scope") != "offline_protocol_and_tool_schema_compatibility_only": raise ReplayError("replay_scope_invalid") if payload.get("risk_class") != "high": raise ReplayError("replay_risk_class_invalid") expiry_text = _require_text(payload.get("expires_at"), "replay_expiry_invalid") try: expiry = datetime.fromisoformat(expiry_text) except ValueError as exc: raise ReplayError("replay_expiry_invalid") from exc if expiry.tzinfo is None or expiry <= datetime.now(timezone.utc): raise ReplayError("replay_policy_expired") for field in ( "exit_condition", "replacement_action", "rollback_ref", "post_verifier", ): _require_text(payload.get(field), f"replay_{field}_invalid") artifact = _require_dict(payload.get("artifact_source"), "artifact_source_invalid") audit_commit = _require_text(artifact.get("audit_commit"), "audit_commit_invalid") if not GIT_SHA_PATTERN.fullmatch(audit_commit): raise ReplayError("audit_commit_invalid") package_integrities = _require_dict( artifact.get("package_integrities"), "package_integrities_invalid" ) if set(package_integrities) != EXPECTED_PACKAGE_KEYS: raise ReplayError("package_integrities_invalid") normalized_integrities: dict[str, str] = {} for key, value in package_integrities.items(): digest = _require_text(value, "package_integrity_invalid") if not SHA512_HEX_PATTERN.fullmatch(digest): raise ReplayError("package_integrity_invalid") normalized_integrities[key] = digest runtime = _require_dict(payload.get("runtime"), "runtime_invalid") if ( runtime.get("network_mode") != "none" or runtime.get("read_only_root") is not True or runtime.get("no_new_privileges") is not True or runtime.get("cap_drop") != ["ALL"] ): raise ReplayError("runtime_isolation_invalid") entrypoint = _require_text_list(runtime.get("entrypoint"), "entrypoint_invalid") arguments = _require_text_list(runtime.get("arguments"), "arguments_invalid") if entrypoint != ("node", "node_modules/@playwright/mcp/cli.js"): raise ReplayError("entrypoint_not_allowlisted") required_arguments = { "--headless", "--isolated", "--allowed-origins", "https://awoooi.wooo.work", "--output-dir", "/tmp/mcp-output", "--executable-path", "/nonexistent/browser", } if set(arguments) != required_arguments or len(arguments) != 8: raise ReplayError("arguments_not_allowlisted") if runtime.get("user") != "65534:65534": raise ReplayError("runtime_user_invalid") protocol = _require_dict( payload.get("protocol_contract"), "protocol_contract_invalid" ) allowed = _require_text_list( protocol.get("adapter_allowed_tools"), "adapter_allowed_tools_invalid" ) denied = _require_text_list( protocol.get("adapter_denied_tools"), "adapter_denied_tools_invalid" ) if set(allowed) & set(denied): raise ReplayError("adapter_tool_partition_invalid") if set(allowed) != { "browser_close", "browser_navigate", "browser_snapshot", "browser_take_screenshot", }: raise ReplayError("adapter_allowed_tools_not_minimal") if "browser_run_code_unsafe" not in denied or "browser_file_upload" not in denied: raise ReplayError("adapter_dangerous_tools_not_denied") boundaries = _require_dict( payload.get("execution_boundaries"), "execution_boundaries_invalid" ) required_false = { "direct_gateway_registration_allowed", "adapter_registration_allowed", "browser_start_allowed", "tool_call_allowed", "navigation_allowed", "authenticated_session_allowed", "raw_request_or_response_body_storage_allowed", "external_rag_write_allowed", "production_write_allowed", "shadow_allowed", "canary_allowed", "promotion_allowed", } if any(boundaries.get(field) is not False for field in required_false): raise ReplayError("execution_boundary_invalid") blockers = _require_text_list( payload.get("promotion_blockers_after_replay"), "promotion_blockers_after_replay_invalid", ) if len(blockers) < 4: raise ReplayError("promotion_blockers_after_replay_invalid") return ReplayPolicy( work_item_id=_require_text(payload.get("work_item_id"), "work_item_id_invalid"), candidate_id=_require_text(payload.get("candidate_id"), "candidate_id_invalid"), adapter_id=_require_text(payload.get("adapter_id"), "adapter_id_invalid"), risk_class="high", owner_lane=_require_text(payload.get("owner_lane"), "owner_lane_invalid"), expires_at=expiry_text, artifact_policy_checksum=_validate_sha256( artifact.get("policy_checksum"), "artifact_policy_checksum_invalid", prefix=True, ), audit_branch=_require_text(artifact.get("audit_branch"), "audit_branch_invalid"), audit_commit=audit_commit, controller_receipt_sha256=_validate_sha256( artifact.get("controller_receipt_sha256"), "controller_receipt_sha256_invalid", ), verifier_receipt_sha256=_validate_sha256( artifact.get("verifier_receipt_sha256"), "verifier_receipt_sha256_invalid", ), artifact_ref=_validate_oci_ref(artifact.get("artifact_ref"), "artifact_ref_invalid"), artifact_runtime_ref=_validate_oci_ref( artifact.get("artifact_runtime_ref"), "artifact_runtime_ref_invalid" ), bundle_manifest_sha256=_validate_sha256( artifact.get("bundle_manifest_sha256"), "bundle_manifest_sha256_invalid" ), cyclonedx_sbom_sha256=_validate_sha256( artifact.get("cyclonedx_sbom_sha256"), "cyclonedx_sbom_sha256_invalid" ), package_integrities=normalized_integrities, runtime_image_ref=_validate_oci_ref( runtime.get("image_ref"), "runtime_image_ref_invalid" ), runtime_platform=_require_text( runtime.get("platform"), "runtime_platform_invalid" ), runtime_user=runtime["user"], runtime_entrypoint=entrypoint, runtime_arguments=arguments, pids_limit=_require_positive_int(runtime.get("pids_limit"), "pids_limit_invalid"), memory_limit=_require_text(runtime.get("memory_limit"), "memory_limit_invalid"), cpu_limit=_require_text(runtime.get("cpu_limit"), "cpu_limit_invalid"), tmpfs=_require_text(runtime.get("tmpfs"), "tmpfs_invalid"), startup_timeout_seconds=_require_positive_int( runtime.get("startup_timeout_seconds"), "startup_timeout_invalid" ), shutdown_timeout_seconds=_require_positive_int( runtime.get("shutdown_timeout_seconds"), "shutdown_timeout_invalid" ), protocol_version=_require_text( protocol.get("protocol_version"), "protocol_version_invalid" ), server_name=_require_text(protocol.get("server_name"), "server_name_invalid"), server_version=_require_text( protocol.get("server_version"), "server_version_invalid" ), tool_count=_require_positive_int(protocol.get("tool_count"), "tool_count_invalid"), tool_name_set_sha256=_validate_sha256( protocol.get("tool_name_set_sha256"), "tool_name_set_sha256_invalid" ), tool_schema_manifest_sha256=_validate_sha256( protocol.get("tool_schema_manifest_sha256"), "tool_schema_manifest_sha256_invalid", ), adapter_allowed_tools=tuple(sorted(allowed)), adapter_denied_tools=tuple(sorted(denied)), promotion_blockers_after_replay=blockers, checksum=f"sha256:{_sha256_hex(_canonical_json_bytes(payload))}", ) def validate_source_evidence( policy: ReplayPolicy, *, artifact_policy_path: Path, controller_receipt_path: Path, verifier_receipt_path: Path, ) -> SourceEvidence: _, artifact_policy = _read_json_bytes( artifact_policy_path, "artifact_policy_unreadable" ) if artifact_policy.get("schema_version") != ARTIFACT_POLICY_SCHEMA_VERSION: raise ReplayError("artifact_policy_schema_invalid") checksum = f"sha256:{_sha256_hex(_canonical_json_bytes(artifact_policy))}" if checksum != policy.artifact_policy_checksum: raise ReplayError("artifact_policy_checksum_mismatch") controller_raw, controller = _read_json_bytes( controller_receipt_path, "artifact_controller_receipt_unreadable" ) verifier_raw, verifier = _read_json_bytes( verifier_receipt_path, "artifact_verifier_receipt_unreadable" ) controller_sha = _sha256_hex(controller_raw) verifier_sha = _sha256_hex(verifier_raw) if controller_sha != policy.controller_receipt_sha256: raise ReplayError("artifact_controller_receipt_checksum_mismatch") if verifier_sha != policy.verifier_receipt_sha256: raise ReplayError("artifact_verifier_receipt_checksum_mismatch") if ( controller.get("schema_version") != ARTIFACT_CONTROLLER_SCHEMA_VERSION or controller.get("status") != "completed_internal_mirror_pending_independent_verifier" ): raise ReplayError("artifact_controller_receipt_terminal_invalid") if ( verifier.get("schema_version") != ARTIFACT_VERIFIER_SCHEMA_VERSION or verifier.get("status") != "completed_internal_mirror_verified" or _require_dict( verifier.get("independent_post_verifier"), "artifact_post_verifier_invalid", ).get("status") != "passed" ): raise ReplayError("artifact_verifier_receipt_terminal_invalid") for payload in (controller, verifier): if any( payload.get(field) is not False for field in ( "promotion_allowed", "replay_allowed", "shadow_allowed", "canary_allowed", ) ): raise ReplayError("artifact_receipt_boundary_invalid") if payload.get("internal_mirror_ref") != policy.artifact_ref: raise ReplayError("artifact_receipt_ref_mismatch") if payload.get("policy_checksum") != policy.artifact_policy_checksum: raise ReplayError("artifact_receipt_policy_checksum_mismatch") if controller.get("runtime_mirror_ref") != policy.artifact_runtime_ref: raise ReplayError("artifact_runtime_ref_mismatch") if verifier.get("runtime_mirror_ref") != policy.artifact_runtime_ref: raise ReplayError("artifact_runtime_ref_mismatch") if controller.get("bundle_manifest_sha256") != policy.bundle_manifest_sha256: raise ReplayError("artifact_manifest_receipt_mismatch") if controller.get("cyclonedx_sbom_sha256") != policy.cyclonedx_sbom_sha256: raise ReplayError("artifact_sbom_receipt_mismatch") if verifier.get("controller_receipt_sha256") != controller_sha: raise ReplayError("artifact_receipt_chain_invalid") if ( controller.get("run_id") != verifier.get("run_id") or controller.get("trace_id") != verifier.get("trace_id") ): raise ReplayError("artifact_receipt_identity_mismatch") expected_packages = policy.package_integrities observed_packages: dict[str, str] = {} rows = controller.get("packages") if not isinstance(rows, list): raise ReplayError("artifact_package_receipts_invalid") for row in rows: if not isinstance(row, dict): raise ReplayError("artifact_package_receipts_invalid") key = f"{row.get('name')}@{row.get('version')}" digest = row.get("tarball_sha512_hex") if not isinstance(digest, str): raise ReplayError("artifact_package_receipts_invalid") observed_packages[key] = digest if observed_packages != expected_packages: raise ReplayError("artifact_package_receipts_mismatch") return SourceEvidence( artifact_run_id=_require_text(controller.get("run_id"), "artifact_run_id_invalid"), artifact_trace_id=_require_text( controller.get("trace_id"), "artifact_trace_id_invalid" ), artifact_controller_receipt_sha256=controller_sha, artifact_verifier_receipt_sha256=verifier_sha, ) def _run_command(command: list[str], *, timeout: int, error_class: str) -> str: try: completed = subprocess.run( command, check=False, capture_output=True, text=True, timeout=timeout, ) except (OSError, subprocess.TimeoutExpired) as exc: raise ReplayError(error_class) from exc if completed.returncode != 0: raise ReplayError(error_class) return completed.stdout def _docker_cli() -> str: docker = shutil.which("docker") if not docker: raise ReplayError("docker_cli_unavailable") return docker def _pull_exact_image(docker: str, ref: str, platform: str) -> None: _run_command( [docker, "pull", "--platform", platform, ref], timeout=300, error_class="immutable_image_pull_failed", ) repo_digests_raw = _run_command( [docker, "image", "inspect", "--format", "{{json .RepoDigests}}", ref], timeout=30, error_class="immutable_image_inspect_failed", ).strip() try: repo_digests = json.loads(repo_digests_raw) except json.JSONDecodeError as exc: raise ReplayError("immutable_image_inspect_invalid") from exc if not isinstance(repo_digests, list) or ref not in repo_digests: raise ReplayError("immutable_image_digest_readback_mismatch") def _safe_relative_member(name: str) -> Path | None: pure = PurePosixPath(name) if pure.is_absolute() or ".." in pure.parts or not pure.parts: raise ReplayError("artifact_archive_path_invalid") if pure.parts[0] != "package": raise ReplayError("artifact_archive_root_invalid") if len(pure.parts) == 1: return None return Path(*pure.parts[1:]) def _extract_package(tarball: Path, destination: Path) -> tuple[int, int]: file_count = 0 unpacked_bytes = 0 try: archive = tarfile.open(tarball, mode="r:gz") except (OSError, tarfile.TarError) as exc: raise ReplayError("artifact_tarball_invalid") from exc with archive: for member in archive.getmembers(): relative = _safe_relative_member(member.name) if relative is None: continue target = destination / relative if member.isdir(): target.mkdir(parents=True, exist_ok=True) continue if not member.isfile() or member.issym() or member.islnk(): raise ReplayError("artifact_archive_member_type_invalid") file_count += 1 unpacked_bytes += member.size if ( file_count > MAX_PACKAGE_FILES or unpacked_bytes > MAX_PACKAGE_UNPACKED_BYTES ): raise ReplayError("artifact_archive_limits_exceeded") source = archive.extractfile(member) if source is None: raise ReplayError("artifact_archive_member_unreadable") content = source.read(member.size + 1) if len(content) != member.size: raise ReplayError("artifact_archive_member_size_mismatch") target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(content) return file_count, unpacked_bytes def _validate_package_json( directory: Path, *, expected_name: str, expected_version: str ) -> None: _, package = _read_json_bytes(directory / "package.json", "package_json_invalid") if package.get("name") != expected_name or package.get("version") != expected_version: raise ReplayError("package_identity_mismatch") dependencies = package.get("dependencies", {}) if not isinstance(dependencies, dict): raise ReplayError("package_dependencies_invalid") for dependency, version in dependencies.items(): if dependency in {"playwright", "playwright-core"} and ( version != "1.62.0-alpha-1783623505000" ): raise ReplayError("package_dependency_version_mismatch") def materialize_verified_node_modules( policy: ReplayPolicy, *, directory: Path ) -> tuple[Path, int, int]: docker = _docker_cli() _pull_exact_image(docker, policy.artifact_ref, policy.runtime_platform) bundle = directory / "bundle" node_modules = directory / "node_modules" bundle.mkdir(parents=True, exist_ok=False) node_modules.mkdir(parents=True, exist_ok=False) container_name = f"awoooi-mcp-artifact-{uuid.uuid4().hex[:20]}" try: _run_command( [ docker, "create", "--platform", policy.runtime_platform, "--name", container_name, policy.artifact_ref, "/nonexistent/no-start", ], timeout=60, error_class="artifact_container_create_failed", ) _run_command( [docker, "cp", f"{container_name}:/artifact/.", str(bundle)], timeout=60, error_class="artifact_bundle_copy_failed", ) finally: subprocess.run( [docker, "rm", "-f", container_name], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30, ) manifest_raw, manifest = _read_json_bytes( bundle / "bundle-manifest.json", "artifact_bundle_manifest_invalid" ) sbom_raw, sbom = _read_json_bytes( bundle / "sbom.cdx.json", "artifact_bundle_sbom_invalid" ) if _sha256_hex(manifest_raw) != policy.bundle_manifest_sha256: raise ReplayError("artifact_bundle_manifest_checksum_mismatch") if _sha256_hex(sbom_raw) != policy.cyclonedx_sbom_sha256: raise ReplayError("artifact_bundle_sbom_checksum_mismatch") if ( manifest.get("schema_version") != ARTIFACT_BUNDLE_SCHEMA_VERSION or manifest.get("candidate_id") != policy.candidate_id or manifest.get("policy_checksum") != policy.artifact_policy_checksum or manifest.get("package_count") != 3 ): raise ReplayError("artifact_bundle_manifest_identity_mismatch") boundaries = _require_dict( manifest.get("operation_boundaries"), "artifact_bundle_boundaries_invalid" ) if any( boundaries.get(field) is not False for field in ( "mcp_server_started", "browser_started", "external_tool_call_performed", "external_rag_write_performed", "production_write_performed", "runtime_request_or_response_body_stored", ) ): raise ReplayError("artifact_bundle_boundary_invalid") if ( sbom.get("bomFormat") != "CycloneDX" or sbom.get("specVersion") != "1.5" or not isinstance(sbom.get("components"), list) or len(sbom["components"]) != 3 ): raise ReplayError("artifact_bundle_sbom_contract_invalid") rows = manifest.get("packages") if not isinstance(rows, list) or len(rows) != 3: raise ReplayError("artifact_bundle_package_set_invalid") seen: set[str] = set() total_files = 0 total_bytes = 0 for row in rows: if not isinstance(row, dict): raise ReplayError("artifact_bundle_package_set_invalid") name = _require_text(row.get("name"), "artifact_package_name_invalid") version = _require_text( row.get("version"), "artifact_package_version_invalid" ) key = f"{name}@{version}" if key in seen or policy.package_integrities.get(key) != row.get( "tarball_sha512_hex" ): raise ReplayError("artifact_package_integrity_contract_invalid") seen.add(key) relative_tarball = _require_text( row.get("tarball_path"), "artifact_tarball_path_invalid" ) tarball_path = bundle / relative_tarball try: tarball_raw = tarball_path.read_bytes() except OSError as exc: raise ReplayError("artifact_tarball_unreadable") from exc if hashlib.sha512(tarball_raw).hexdigest() != policy.package_integrities[key]: raise ReplayError("artifact_tarball_checksum_mismatch") package_dir = ( node_modules / "@playwright" / "mcp" if name == "@playwright/mcp" else node_modules / name ) package_dir.mkdir(parents=True, exist_ok=False) files, size = _extract_package(tarball_path, package_dir) total_files += files total_bytes += size _validate_package_json( package_dir, expected_name=name, expected_version=version ) if seen != set(policy.package_integrities): raise ReplayError("artifact_bundle_package_set_mismatch") for path in node_modules.rglob("*"): if path.is_dir(): path.chmod(0o755) elif path.is_file(): path.chmod(0o444) node_modules.chmod(0o755) directory.chmod(0o755) return node_modules, total_files, total_bytes def _write_rpc(process: subprocess.Popen[str], payload: dict[str, Any]) -> None: if process.stdin is None: raise ReplayError("mcp_stdin_unavailable") try: process.stdin.write(_canonical_json_bytes(payload).decode("utf-8") + "\n") process.stdin.flush() except (BrokenPipeError, OSError) as exc: raise ReplayError("mcp_protocol_write_failed") from exc def _read_rpc( process: subprocess.Popen[str], *, timeout_seconds: int, expected_id: int ) -> dict[str, Any]: if process.stdout is None: raise ReplayError("mcp_stdout_unavailable") ready, _, _ = select.select([process.stdout], [], [], timeout_seconds) if not ready: raise ReplayError("mcp_protocol_response_timeout") line = process.stdout.readline(MAX_PROTOCOL_LINE_BYTES + 1) if not line or len(line.encode("utf-8")) > MAX_PROTOCOL_LINE_BYTES: raise ReplayError("mcp_protocol_response_invalid") try: payload = json.loads(line) except json.JSONDecodeError as exc: raise ReplayError("mcp_protocol_response_invalid") from exc if ( not isinstance(payload, dict) or payload.get("jsonrpc") != "2.0" or payload.get("id") != expected_id or "error" in payload or not isinstance(payload.get("result"), dict) ): raise ReplayError("mcp_protocol_response_invalid") return payload["result"] def _tool_hashes(tools: Any) -> tuple[int, str, str, tuple[str, ...]]: if not isinstance(tools, list) or not tools: raise ReplayError("mcp_tool_list_invalid") normalized: list[dict[str, Any]] = [] names: list[str] = [] for tool in tools: if not isinstance(tool, dict): raise ReplayError("mcp_tool_schema_invalid") name = _require_text(tool.get("name"), "mcp_tool_name_invalid") input_schema = tool.get("inputSchema") if not isinstance(input_schema, dict): raise ReplayError("mcp_tool_schema_invalid") row: dict[str, Any] = {"name": name, "inputSchema": input_schema} if "annotations" in tool: if not isinstance(tool["annotations"], dict): raise ReplayError("mcp_tool_annotations_invalid") row["annotations"] = tool["annotations"] names.append(name) normalized.append(row) if len(set(names)) != len(names): raise ReplayError("mcp_tool_names_not_unique") sorted_names = tuple(sorted(names)) normalized.sort(key=lambda row: row["name"]) return ( len(tools), _sha256_hex(_canonical_json_bytes(list(sorted_names))), _sha256_hex(_canonical_json_bytes(normalized)), sorted_names, ) def run_protocol_replay( policy: ReplayPolicy, *, node_modules: Path, client_name: str ) -> ReplayObservation: docker = _docker_cli() _pull_exact_image(docker, policy.runtime_image_ref, policy.runtime_platform) container_name = f"awoooi-mcp-replay-{uuid.uuid4().hex[:20]}" command = [ docker, "run", "--rm", "--platform", policy.runtime_platform, "--name", container_name, "--network", "none", "--read-only", "--security-opt", "no-new-privileges", "--cap-drop", "ALL", "--pids-limit", str(policy.pids_limit), "--memory", policy.memory_limit, "--cpus", policy.cpu_limit, "--user", policy.runtime_user, "--env", "HOME=/tmp", "--env", "CI=1", "--tmpfs", policy.tmpfs, "--stop-timeout", str(policy.shutdown_timeout_seconds), "-i", "-v", f"{node_modules.resolve()}:/work/node_modules:ro", "-w", "/work", policy.runtime_image_ref, *policy.runtime_entrypoint, *policy.runtime_arguments, ] process: subprocess.Popen[str] | None = None stderr_file = tempfile.TemporaryFile(mode="w+t", encoding="utf-8") shutdown_mode = "not_started" try: try: process = subprocess.Popen( command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr_file, text=True, bufsize=1, env={**os.environ, "DOCKER_CLI_HINTS": "false"}, ) except OSError as exc: raise ReplayError("mcp_process_start_failed") from exc _write_rpc( process, { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": policy.protocol_version, "capabilities": {}, "clientInfo": {"name": client_name, "version": "1"}, }, }, ) initialize = _read_rpc( process, timeout_seconds=policy.startup_timeout_seconds, expected_id=1, ) server_info = _require_dict( initialize.get("serverInfo"), "mcp_server_info_invalid" ) if ( initialize.get("protocolVersion") != policy.protocol_version or server_info.get("name") != policy.server_name or server_info.get("version") != policy.server_version ): raise ReplayError("mcp_initialize_contract_mismatch") _write_rpc( process, { "jsonrpc": "2.0", "method": "notifications/initialized", "params": {}, }, ) _write_rpc( process, {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}, ) tool_result = _read_rpc( process, timeout_seconds=policy.startup_timeout_seconds, expected_id=2, ) tool_count, names_hash, schemas_hash, names = _tool_hashes( tool_result.get("tools") ) if ( tool_count != policy.tool_count or names_hash != policy.tool_name_set_sha256 or schemas_hash != policy.tool_schema_manifest_sha256 ): raise ReplayError("mcp_tool_contract_mismatch") if set(names) != set(policy.adapter_allowed_tools) | set( policy.adapter_denied_tools ): raise ReplayError("mcp_adapter_tool_partition_mismatch") if process.stdin is None: raise ReplayError("mcp_stdin_unavailable") process.stdin.close() try: exit_code = process.wait(timeout=policy.shutdown_timeout_seconds) shutdown_mode = "stdin_eof_clean_exit" except subprocess.TimeoutExpired as exc: raise ReplayError("mcp_process_shutdown_timeout") from exc if exit_code != 0: raise ReplayError("mcp_process_exit_invalid") finally: if process is not None and process.poll() is None: process.terminate() try: process.wait(timeout=policy.shutdown_timeout_seconds) shutdown_mode = "terminated_after_replay_error" except subprocess.TimeoutExpired: process.kill() process.wait(timeout=policy.shutdown_timeout_seconds) shutdown_mode = "killed_after_replay_error" subprocess.run( [docker, "rm", "-f", container_name], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30, ) stderr_file.close() inspect = subprocess.run( [docker, "container", "inspect", container_name], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30, ) if inspect.returncode == 0: raise ReplayError("ephemeral_container_cleanup_failed") return 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=shutdown_mode, ephemeral_container_removed=True, ) def build_receipt( *, policy: ReplayPolicy, source: SourceEvidence, run_id: str, trace_id: str, mode: str, observation: ReplayObservation | None, package_file_count: int, package_unpacked_bytes: int, ) -> dict[str, Any]: applied = observation is not None status = ( "completed_protocol_schema_replay_pending_independent_verifier" if applied else "completed_check_mode_no_runtime_execution" ) return { "schema_version": RECEIPT_SCHEMA_VERSION, "run_id": run_id, "trace_id": trace_id, "work_item_id": policy.work_item_id, "candidate_id": policy.candidate_id, "adapter_id": policy.adapter_id, "risk_class": policy.risk_class, "mode": mode, "status": status, "observed_at": datetime.now(timezone.utc).isoformat(), "policy_checksum": policy.checksum, "policy_expires_at": policy.expires_at, "artifact_audit_branch": policy.audit_branch, "artifact_audit_commit": policy.audit_commit, "artifact_ref": policy.artifact_ref, "runtime_image_ref": policy.runtime_image_ref, "artifact_run_id": source.artifact_run_id, "artifact_trace_id": source.artifact_trace_id, "protocol_observation": ( { "protocol_version": observation.protocol_version, "server_name": observation.server_name, "server_version": observation.server_version, "tool_count": observation.tool_count, "tool_name_set_sha256": observation.tool_name_set_sha256, "tool_schema_manifest_sha256": observation.tool_schema_manifest_sha256, "adapter_allowed_tools": list(policy.adapter_allowed_tools), "adapter_denied_tool_count": len(policy.adapter_denied_tools), "process_exit_code": observation.process_exit_code, "shutdown_mode": observation.shutdown_mode, } if observation else None ), "direct_gateway_registration_allowed": False, "adapter_registration_allowed": False, "browser_started": False, "tool_call_performed": False, "navigation_performed": False, "raw_request_or_response_body_stored": False, "external_rag_write_performed": False, "production_write_performed": False, "shadow_allowed": False, "canary_allowed": False, "promotion_allowed": False, "promotion_blockers": list(policy.promotion_blockers_after_replay), "stage_receipts": { "sensor_source_receipt": { "artifact_controller_receipt_sha256": ( source.artifact_controller_receipt_sha256 ), "artifact_verifier_receipt_sha256": ( source.artifact_verifier_receipt_sha256 ), "artifact_audit_commit": policy.audit_commit, }, "normalized_asset_identity": { "candidate_id": policy.candidate_id, "adapter_id": policy.adapter_id, "artifact_ref": policy.artifact_ref, "runtime_image_ref": policy.runtime_image_ref, }, "source_of_truth_diff": { "change_state": ( "protocol_schema_replay_verified_by_controller" if applied else "protocol_schema_replay_ready_for_bounded_execution" ), "direct_registration_state": "disabled", }, "decision": { "candidate_action": ( "run_independent_protocol_schema_replay_verifier" if applied else "run_bounded_protocol_schema_replay" ), "browser_shadow_or_canary_allowed": False, }, "risk_policy": { "risk_class": "high", "network_mode": "none", "read_only_root": True, "no_new_privileges": True, "cap_drop_all": True, "non_root_user": policy.runtime_user, "tool_call_allowed": False, }, "check_mode": { "status": "passed", "artifact_receipt_chain_verified": True, "exact_artifact_and_runtime_refs_verified": True, "adapter_tool_partition_verified": True, }, "bounded_execution": { "status": ( "mcp_initialize_and_tools_list_completed" if applied else "not_executed" ), "package_file_count": package_file_count, "package_unpacked_bytes": package_unpacked_bytes, "mcp_process_started": applied, "browser_started": False, "mcp_tool_call_performed": False, "network_enabled": False, }, "independent_post_verifier": { "status": "pending_external_verifier" if applied else "not_requested", "implementation": "standalone_no_controller_import", }, "rollback_or_no_write_terminal": { "status": ( "mcp_process_terminated_and_ephemeral_container_removed" if applied else "no_runtime_execution" ), "ephemeral_container_removed": ( observation.ephemeral_container_removed if observation else True ), "candidate_registration_state": "disabled", "production_route_changed": False, }, "learning_writeback": { "status": "receipt_ready_for_gitea_audit_branch_writeback", "external_rag_write_performed": False, "km_write_performed": False, }, }, } def _failure_receipt( *, policy: ReplayPolicy, run_id: str, trace_id: str, error_class: str ) -> dict[str, Any]: return { "schema_version": RECEIPT_SCHEMA_VERSION, "run_id": run_id, "trace_id": trace_id, "work_item_id": policy.work_item_id, "candidate_id": policy.candidate_id, "risk_class": policy.risk_class, "status": "blocked_with_safe_next_action", "error_class": error_class, "observed_at": datetime.now(timezone.utc).isoformat(), "policy_checksum": policy.checksum, "browser_started": False, "tool_call_performed": False, "external_rag_write_performed": False, "production_write_performed": False, "promotion_allowed": False, "stage_receipts": { "rollback_or_no_write_terminal": { "status": "candidate_disabled_ephemeral_process_cleanup_attempted", "candidate_registration_state": "disabled", "production_route_changed": False, }, "learning_writeback": { "status": "blocked_receipt_ready_for_gitea_audit_branch_writeback", "external_rag_write_performed": False, }, }, } def _write_receipt(path: Path, receipt: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(_canonical_json_bytes(receipt) + b"\n") def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--policy", type=Path, required=True) parser.add_argument("--artifact-policy", type=Path, required=True) parser.add_argument("--artifact-controller-receipt", type=Path, required=True) parser.add_argument("--artifact-verifier-receipt", type=Path, required=True) parser.add_argument("--run-id", required=True) parser.add_argument("--trace-id", required=True) parser.add_argument("--receipt", type=Path) parser.add_argument("command", choices=("check", "replay")) parser.add_argument("--apply", action="store_true") return parser def main(argv: list[str] | None = None) -> int: args = _build_parser().parse_args(argv) try: normalized_run_id = str(uuid.UUID(args.run_id)) except (ValueError, AttributeError) as exc: raise ReplayError("run_id_invalid") from exc if normalized_run_id != args.run_id: raise ReplayError("run_id_not_canonical") if ( not TRACE_PATTERN.fullmatch(args.trace_id) or args.trace_id != f"mcp-replay-{normalized_run_id}" ): raise ReplayError("run_trace_identity_mismatch") if args.command == "check" and args.apply: raise ReplayError("check_mode_cannot_apply") if args.command == "replay" and not args.apply: raise ReplayError("replay_requires_apply") if args.command == "replay" and args.receipt is None: raise ReplayError("apply_receipt_path_required") policy = load_policy(args.policy) try: source = validate_source_evidence( policy, artifact_policy_path=args.artifact_policy, controller_receipt_path=args.artifact_controller_receipt, verifier_receipt_path=args.artifact_verifier_receipt, ) observation = None package_file_count = 0 package_unpacked_bytes = 0 if args.command == "replay" and args.apply: with tempfile.TemporaryDirectory(prefix="awoooi-mcp-replay-") as temp: node_modules, package_file_count, package_unpacked_bytes = ( materialize_verified_node_modules(policy, directory=Path(temp)) ) observation = run_protocol_replay( policy, node_modules=node_modules, client_name="awoooi-external-mcp-replay-controller", ) receipt = build_receipt( policy=policy, source=source, run_id=normalized_run_id, trace_id=args.trace_id, mode="apply" if observation else "check", observation=observation, package_file_count=package_file_count, package_unpacked_bytes=package_unpacked_bytes, ) except ReplayError as exc: if args.receipt is not None: _write_receipt( args.receipt, _failure_receipt( policy=policy, run_id=normalized_run_id, trace_id=args.trace_id, error_class=str(exc), ), ) raise if args.receipt is not None: _write_receipt(args.receipt, receipt) print(json.dumps(receipt, ensure_ascii=False, sort_keys=True)) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except ReplayError as exc: print( json.dumps( { "schema_version": RECEIPT_SCHEMA_VERSION, "status": "blocked_with_safe_next_action", "error_class": str(exc), }, sort_keys=True, ), file=sys.stderr, ) raise SystemExit(1) from None