from __future__ import annotations import base64 import hashlib import importlib.util import io import json import sys import tarfile import tempfile import unittest import subprocess from copy import deepcopy from pathlib import Path from urllib.parse import quote ROOT = Path(__file__).resolve().parents[3] VERIFIER_PATH = ROOT / "scripts/security/verify_external_mcp_artifact_receipt.py" RUN_ID = "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee" TRACE_ID = f"mcp-artifact-{RUN_ID}" def _load_verifier(): spec = importlib.util.spec_from_file_location( "verify_external_mcp_artifact_receipt", VERIFIER_PATH ) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module verifier = _load_verifier() def _canonical(payload) -> bytes: return json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode() def _fixture(): package_data = { "@playwright/mcp": b"root-package", "playwright": b"playwright-package", "playwright-core": b"playwright-core-package", } versions = { "@playwright/mcp": "0.0.78", "playwright": "1.62.0-alpha-1783623505000", "playwright-core": "1.62.0-alpha-1783623505000", } packages = [] with tempfile.TemporaryDirectory() as temp: root = Path(temp) private_key = root / "private.pem" public_key = root / "public.der" subprocess.run( [ "openssl", "ecparam", "-name", "prime256v1", "-genkey", "-noout", "-out", str(private_key), ], check=True, capture_output=True, ) subprocess.run( [ "openssl", "pkey", "-in", str(private_key), "-pubout", "-outform", "DER", "-out", str(public_key), ], check=True, capture_output=True, ) public_key_base64 = base64.b64encode(public_key.read_bytes()).decode() for index, (name, data) in enumerate(package_data.items()): integrity = "sha512-" + base64.b64encode( hashlib.sha512(data).digest() ).decode() message = f"{name}@{versions[name]}:{integrity}".encode() message_path = root / f"message-{index}.bin" signature_path = root / f"signature-{index}.der" message_path.write_bytes(message) subprocess.run( [ "openssl", "dgst", "-sha256", "-sign", str(private_key), "-out", str(signature_path), str(message_path), ], check=True, capture_output=True, ) packages.append( { "name": name, "version": versions[name], "integrity": integrity, "signature": base64.b64encode( signature_path.read_bytes() ).decode(), "slsa_subject_name": ( f"pkg:npm/{quote(name, safe='/')}@{versions[name]}" ), "slsa_subject_sha512_hex": hashlib.sha512(data).hexdigest(), "shasum_sha1": hashlib.sha1( data, usedforsecurity=False ).hexdigest(), } ) policy = { "schema_version": verifier.POLICY_SCHEMA_VERSION, "work_item_id": "MCP-ARTIFACT-PLAYWRIGHT-TEST", "candidate_id": "external.playwright-verifier", "risk_level": "high", "registry_contract": { "signature_key": { "public_key_der_base64": public_key_base64, } }, "packages": packages, "internal_mirror": { "push_repository": ( "registry.wooo.work/awoooi/mcp-artifacts/playwright-mcp" ), "runtime_repository": ( "192.168.0.110:5000/awoooi/mcp-artifacts/playwright-mcp" ), }, } policy_checksum = "sha256:" + hashlib.sha256(_canonical(policy)).hexdigest() sbom = { "bomFormat": "CycloneDX", "specVersion": "1.5", "components": [ { "bom-ref": f"pkg:npm/{quote(row['name'], safe='/')}@{row['version']}" } for row in packages ], } sbom_bytes = _canonical(sbom) + b"\n" files = {"artifact/sbom.cdx.json": sbom_bytes} manifest_rows = [] receipt_rows = [] for row in packages: safe_name = row["name"].replace("@", "").replace("/", "-") filename = f"{safe_name}-{row['version']}.tgz" statement = { "_type": verifier.IN_TOTO_STATEMENT_TYPE, "predicateType": verifier.SLSA_PREDICATE_TYPE, "subject": [ { "name": row["slsa_subject_name"], "digest": {"sha512": row["slsa_subject_sha512_hex"]}, } ], } attestation = _canonical( { "attestations": [ { "predicateType": verifier.SLSA_PREDICATE_TYPE, "bundle": { "dsseEnvelope": { "payload": base64.b64encode( _canonical(statement) ).decode() } }, } ] } ) files[f"artifact/tarballs/{filename}"] = package_data[row["name"]] files[f"artifact/attestations/{filename}.attestations.json"] = attestation manifest_rows.append( { "name": row["name"], "version": row["version"], "tarball_path": f"tarballs/{filename}", "tarball_sha512_hex": row["slsa_subject_sha512_hex"], "tarball_sha1": row["shasum_sha1"], "attestation_path": ( f"attestations/{filename}.attestations.json" ), "attestation_sha256": hashlib.sha256(attestation).hexdigest(), "npm_registry_signature_verified": True, "slsa_subject_verified": True, "archive_safety_verified": True, "vulnerability_ids": [], } ) receipt_rows.append( { "name": row["name"], "version": row["version"], "tarball_sha512_hex": row["slsa_subject_sha512_hex"], "tarball_sha1": row["shasum_sha1"], "npm_registry_signature_verified": True, "slsa_subject_verified": True, "archive_safety_verified": True, "license_decision": "passed", "vulnerability_decision": "passed_no_known_osv_records", "vulnerability_ids": [], } ) manifest = { "candidate_id": policy["candidate_id"], "work_item_id": policy["work_item_id"], "policy_checksum": policy_checksum, "package_count": 3, "packages": manifest_rows, "sbom": {"sha256": hashlib.sha256(sbom_bytes).hexdigest()}, "operation_boundaries": { "mcp_server_started": False, "browser_started": False, "runtime_request_or_response_body_stored": False, }, } manifest_bytes = _canonical(manifest) + b"\n" files["artifact/bundle-manifest.json"] = manifest_bytes digest = "sha256:" + "c" * 64 receipt = { "schema_version": verifier.CONTROLLER_RECEIPT_SCHEMA_VERSION, "run_id": RUN_ID, "trace_id": TRACE_ID, "work_item_id": policy["work_item_id"], "candidate_id": policy["candidate_id"], "risk_level": "high", "mode": "apply", "status": "completed_internal_mirror_pending_independent_verifier", "policy_checksum": policy_checksum, "packages": receipt_rows, "internal_mirror_ref": ( policy["internal_mirror"]["push_repository"] + "@" + digest ), "runtime_mirror_ref": ( policy["internal_mirror"]["runtime_repository"] + "@" + digest ), "bundle_manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(), "cyclonedx_sbom_sha256": hashlib.sha256(sbom_bytes).hexdigest(), "promotion_allowed": False, "replay_allowed": False, "shadow_allowed": False, "canary_allowed": False, "stage_receipts": { "sensor_source_receipt": { "redirects_followed": 0, "runtime_request_or_response_body_stored": False, "upstream_supply_chain_attestation_bundled": True, }, "check_mode": { "status": "passed", "mcp_process_started": False, "browser_started": False, "production_write_performed": False, }, "bounded_execution": { "status": "internal_oci_bundle_pushed", "internal_registry_write_performed": True, "external_runtime_mutation_performed": False, "production_route_changed": False, }, "independent_post_verifier": { "status": "pending_external_verifier", "internal_digest_ref_verified": False, }, }, } return policy, receipt, files def _write_image_archive(path: Path, files: dict[str, bytes], *, symlink=False): layer_buffer = io.BytesIO() with tarfile.open(fileobj=layer_buffer, mode="w") as layer: for name, data in files.items(): member = tarfile.TarInfo(name) member.size = len(data) layer.addfile(member, io.BytesIO(data)) if symlink: member = tarfile.TarInfo("artifact/unsafe-link") member.type = tarfile.SYMTYPE member.linkname = "../../outside" layer.addfile(member) layer_bytes = layer_buffer.getvalue() manifest_bytes = json.dumps([{"Layers": ["layer.tar"]}]).encode() with tarfile.open(path, mode="w") as outer: for name, data in ( ("manifest.json", manifest_bytes), ("layer.tar", layer_bytes), ): member = tarfile.TarInfo(name) member.size = len(data) outer.addfile(member, io.BytesIO(data)) class ExternalMcpArtifactIndependentVerifierTest(unittest.TestCase): def test_controller_receipt_identity_and_boundaries_are_validated(self) -> None: policy, receipt, _ = _fixture() identity = verifier.validate_controller_receipt(policy, receipt) self.assertEqual(identity["run_id"], RUN_ID) self.assertEqual(identity["trace_id"], TRACE_ID) self.assertEqual(identity["digest"], "sha256:" + "c" * 64) def test_controller_receipt_rejects_promotion_claim(self) -> None: policy, receipt, _ = _fixture() receipt["canary_allowed"] = True with self.assertRaisesRegex( verifier.VerifierError, "controller_receipt_promotion_boundary_invalid", ): verifier.validate_controller_receipt(policy, receipt) def test_bundle_files_are_rehashed_independently(self) -> None: policy, receipt, files = _fixture() evidence = verifier.verify_bundle_files( files, policy=policy, controller_receipt=receipt, ) self.assertEqual(evidence["verified_package_count"], 3) self.assertEqual(evidence["artifact_file_count"], 8) def test_bundle_tarball_tamper_is_rejected(self) -> None: policy, receipt, files = _fixture() tampered = deepcopy(files) root_tarball = next( key for key in tampered if key.startswith("artifact/tarballs/playwright-mcp-") ) tampered[root_tarball] += b"tamper" with self.assertRaisesRegex( verifier.VerifierError, "bundle_package_evidence_mismatch", ): verifier.verify_bundle_files( tampered, policy=policy, controller_receipt=receipt, ) def test_bundle_extra_file_is_rejected(self) -> None: policy, receipt, files = _fixture() files["artifact/unexpected.bin"] = b"unexpected" with self.assertRaisesRegex( verifier.VerifierError, "bundle_file_set_invalid", ): verifier.verify_bundle_files( files, policy=policy, controller_receipt=receipt, ) def test_independent_npm_signature_rejects_tamper(self) -> None: policy, _, files = _fixture() package = deepcopy(policy["packages"][0]) package["signature"] = base64.b64encode(b"invalid-signature").decode() tarball_path = next( key for key in files if key.startswith("artifact/tarballs/playwright-mcp-") ) with self.assertRaisesRegex( verifier.VerifierError, "bundled_npm_signature_verification_failed", ): verifier.verify_bundled_npm_signature( policy, package, files[tarball_path], ) def test_independent_slsa_subject_rejects_tamper(self) -> None: policy, _, files = _fixture() package = deepcopy(policy["packages"][0]) package["slsa_subject_sha512_hex"] = "0" * 128 attestation_path = next( key for key in files if key.startswith("artifact/attestations/playwright-mcp-") ) with self.assertRaisesRegex( verifier.VerifierError, "bundled_slsa_subject_mismatch", ): verifier.verify_bundled_slsa_subject( package, files[attestation_path], ) def test_image_archive_is_inspected_as_data(self) -> None: _, _, files = _fixture() with tempfile.TemporaryDirectory() as temp: archive = Path(temp) / "image.tar" _write_image_archive(archive, files) extracted = verifier.extract_image_files(archive) self.assertEqual(extracted, files) def test_image_archive_rejects_symlink(self) -> None: _, _, files = _fixture() with tempfile.TemporaryDirectory() as temp: archive = Path(temp) / "image.tar" _write_image_archive(archive, files, symlink=True) with self.assertRaisesRegex( verifier.VerifierError, "image_layer_unsafe_entry", ): verifier.extract_image_files(archive) def test_verifier_receipt_keeps_runtime_promotion_disabled(self) -> None: policy, receipt, files = _fixture() identity = verifier.validate_controller_receipt(policy, receipt) evidence = verifier.verify_bundle_files( files, policy=policy, controller_receipt=receipt, ) result = verifier.build_verifier_receipt( identity=identity, controller_receipt_bytes=_canonical(receipt) + b"\n", bundle_evidence=evidence, ) self.assertEqual(result["status"], "completed_internal_mirror_verified") self.assertFalse(result["promotion_allowed"]) self.assertFalse(result["replay_allowed"]) self.assertFalse(result["shadow_allowed"]) self.assertFalse(result["canary_allowed"]) self.assertFalse( result["independent_post_verifier"]["container_or_mcp_started"] ) if __name__ == "__main__": unittest.main()