from __future__ import annotations import base64 import importlib.util import io import json import sys import tarfile import tempfile import unittest from contextlib import redirect_stdout from pathlib import Path from unittest.mock import patch ROOT = Path(__file__).resolve().parents[3] CONTROLLER_PATH = ROOT / "scripts/security/external_mcp_artifact_controller.py" POLICY_PATH = ROOT / "config/mcp/playwright-mcp-artifact-policy.json" WORKFLOW_PATH = ROOT / ".gitea/workflows/mcp-external-artifact-mirror.yaml" RUN_ID = "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee" TRACE_ID = f"mcp-artifact-{RUN_ID}" def _load_controller(): spec = importlib.util.spec_from_file_location( "external_mcp_artifact_controller", CONTROLLER_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 controller = _load_controller() def _verified_packages(policy): return tuple( controller.VerifiedPackage( policy=package, tarball=b"tarball", attestation=b"{}", metadata_sha256="a" * 64, tarball_sha512_hex=package.sha512_hex, tarball_sha1=package.shasum_sha1, attestation_sha256="b" * 64, unpacked_bytes=1024, archive_file_count=3, license_file_count=1, osv_response_sha256="c" * 64, vulnerability_ids=(), ) for package in policy.packages ) def _tarball( package, *, unsafe_symlink: bool = False, duplicate_manifest: bool = False, ) -> bytes: package_json = json.dumps( { "name": package.name, "version": package.version, "license": package.license, "dependencies": package.dependencies, "optionalDependencies": package.optional_dependencies, } ).encode() output = io.BytesIO() with tarfile.open(fileobj=output, mode="w:gz") as archive: for name, data in ( ("package/package.json", package_json), ("package/LICENSE", b"Apache License 2.0"), ): member = tarfile.TarInfo(name) member.size = len(data) archive.addfile(member, io.BytesIO(data)) if duplicate_manifest: member = tarfile.TarInfo("package/package.json") member.size = len(package_json) archive.addfile(member, io.BytesIO(package_json)) if unsafe_symlink: member = tarfile.TarInfo("package/escape") member.type = tarfile.SYMTYPE member.linkname = "../../outside" archive.addfile(member) return output.getvalue() class ExternalMcpArtifactControllerTest(unittest.TestCase): def test_policy_is_exact_discovery_only_and_fail_closed(self) -> None: policy = controller.load_policy(POLICY_PATH) raw = POLICY_PATH.read_text(encoding="utf-8").lower() self.assertEqual(policy.candidate_id, "external.playwright-verifier") self.assertEqual(policy.risk_level, "high") self.assertEqual(len(policy.packages), 3) self.assertEqual(policy.mirror_tag, "0.0.78-bundle-v1") self.assertNotIn("@latest", raw) self.assertNotIn(":latest", raw) self.assertNotIn("npx -y", raw) self.assertNotIn("github.com", raw) self.assertNotIn("ghcr.io", raw) self.assertTrue(all(package.version for package in policy.packages)) self.assertTrue(all(package.sha512_hex for package in policy.packages)) def test_gitea_workflow_has_no_remote_action_or_floating_supply_chain(self) -> None: raw = WORKFLOW_PATH.read_text(encoding="utf-8") lowered = raw.lower() self.assertNotIn("uses:", lowered) self.assertNotIn("actions/checkout", lowered) self.assertNotIn("github.com", lowered) self.assertNotIn("api.github.com", lowered) self.assertNotIn("raw.githubusercontent.com", lowered) self.assertNotIn("codeload.github.com", lowered) self.assertNotIn("ghcr.io", lowered) self.assertNotIn("npx -y", lowered) self.assertNotIn("@latest", lowered) self.assertNotIn(":latest", lowered) self.assertNotIn("git push --force", lowered) self.assertIn("runs-on: awoooi-non110-host", raw) self.assertIn("docker login", raw) self.assertIn("--password-stdin", raw) self.assertIn("docker logout", raw) self.assertIn("verify_external_mcp_artifact_receipt.py", raw) self.assertIn("git merge --no-edit gitea/main", raw) self.assertIn("git push gitea HEAD:main", raw) self.assertIn('cron: "13 2 * * 3"', raw) def test_dependency_closure_rejects_drift(self) -> None: payload = json.loads(POLICY_PATH.read_text(encoding="utf-8")) payload["packages"][0]["dependencies"]["playwright"] = "1.2.3" with tempfile.TemporaryDirectory() as temp: path = Path(temp) / "policy.json" path.write_text(json.dumps(payload), encoding="utf-8") with self.assertRaisesRegex( controller.ControllerError, "dependency_closure_invalid" ): controller.load_policy(path) def test_run_and_trace_identity_must_match(self) -> None: with self.assertRaisesRegex( controller.ControllerError, "run_trace_identity_mismatch" ): controller.main( [ "--policy", str(POLICY_PATH), "--run-id", RUN_ID, "--trace-id", "mcp-artifact-00000000-0000-4000-8000-000000000000", "check", ] ) def test_noncanonical_run_id_is_rejected(self) -> None: with self.assertRaisesRegex(controller.ControllerError, "run_id_not_canonical"): controller.main( [ "--policy", str(POLICY_PATH), "--run-id", RUN_ID.upper(), "--trace-id", TRACE_ID, "check", ] ) def test_check_mode_emits_bound_no_write_receipt(self) -> None: policy = controller.load_policy(POLICY_PATH) verified = _verified_packages(policy) with ( patch.object(controller, "verify_upstream_bundle", return_value=verified), patch.object(controller, "build_cyclonedx_sbom", return_value={}), patch.object( controller, "_write_bundle", return_value=("d" * 64, "e" * 64), ), io.StringIO() as output, redirect_stdout(output), ): result = controller.main( [ "--policy", str(POLICY_PATH), "--run-id", RUN_ID, "--trace-id", TRACE_ID, "check", ] ) receipt = json.loads(output.getvalue()) self.assertEqual(result, 0) self.assertEqual(receipt["run_id"], RUN_ID) self.assertEqual(receipt["trace_id"], TRACE_ID) self.assertEqual( receipt["status"], "completed_check_mode_pending_independent_verifier" ) self.assertFalse(receipt["promotion_allowed"]) self.assertFalse(receipt["canary_allowed"]) self.assertFalse( receipt["stage_receipts"]["sensor_source_receipt"] ["runtime_request_or_response_body_stored"] ) self.assertTrue( receipt["stage_receipts"]["sensor_source_receipt"] ["upstream_supply_chain_attestation_bundled"] ) self.assertEqual( receipt["stage_receipts"]["rollback_or_no_write_terminal"]["status"], "no_write_pending_external_verifier", ) self.assertEqual( receipt["stage_receipts"]["independent_post_verifier"]["status"], "pending_external_verifier", ) def test_mirror_apply_requires_receipt_path(self) -> None: with self.assertRaisesRegex( controller.ControllerError, "apply_receipt_path_required" ): controller.main( [ "--policy", str(POLICY_PATH), "--run-id", RUN_ID, "--trace-id", TRACE_ID, "mirror", "--apply", ] ) def test_tarball_inspection_accepts_exact_manifest(self) -> None: package = controller.load_policy(POLICY_PATH).packages[0] unpacked, files, licenses = controller.inspect_tarball( package, _tarball(package), maximum_unpacked_bytes=1024 * 1024, ) self.assertGreater(unpacked, 0) self.assertEqual(files, 2) self.assertEqual(licenses, 1) def test_tarball_inspection_rejects_symlink(self) -> None: package = controller.load_policy(POLICY_PATH).packages[0] with self.assertRaisesRegex( controller.ControllerError, "package_archive_unsafe_entry" ): controller.inspect_tarball( package, _tarball(package, unsafe_symlink=True), maximum_unpacked_bytes=1024 * 1024, ) def test_tarball_inspection_rejects_duplicate_manifest(self) -> None: package = controller.load_policy(POLICY_PATH).packages[0] with self.assertRaisesRegex( controller.ControllerError, "package_archive_duplicate_or_invalid_file", ): controller.inspect_tarball( package, _tarball(package, duplicate_manifest=True), maximum_unpacked_bytes=1024 * 1024, ) def test_slsa_subject_must_match_exact_tarball_digest(self) -> None: package = controller.load_policy(POLICY_PATH).packages[0] statement = { "_type": controller.IN_TOTO_STATEMENT_TYPE, "predicateType": controller.SLSA_PREDICATE_TYPE, "subject": [ { "name": package.slsa_subject_name, "digest": {"sha512": package.slsa_subject_sha512_hex}, } ], } payload = { "attestations": [ { "predicateType": controller.SLSA_PREDICATE_TYPE, "bundle": { "dsseEnvelope": { "payload": base64.b64encode( json.dumps(statement).encode() ).decode() } }, } ] } controller.validate_slsa_attestation(package, payload) bad_statement = dict(statement) bad_statement["subject"] = [ { "name": package.slsa_subject_name, "digest": {"sha512": "0" * 128}, } ] payload["attestations"][0]["bundle"]["dsseEnvelope"]["payload"] = ( base64.b64encode(json.dumps(bad_statement).encode()).decode() ) with self.assertRaisesRegex(controller.ControllerError, "slsa_subject_mismatch"): controller.validate_slsa_attestation(package, payload) def test_sbom_has_exact_npm_purls_and_dependency_edges(self) -> None: policy = controller.load_policy(POLICY_PATH) sbom = controller.build_cyclonedx_sbom( policy, _verified_packages(policy), ) refs = {row["bom-ref"] for row in sbom["components"]} self.assertIn("pkg:npm/%40playwright/mcp@0.0.78", refs) self.assertNotIn("pkg:npm/%40playwright%2Fmcp@0.0.78", refs) root = next( row for row in sbom["dependencies"] if row["ref"] == "pkg:npm/%40playwright/mcp@0.0.78" ) self.assertEqual(len(root["dependsOn"]), 2) def test_osv_records_fail_closed(self) -> None: policy = controller.load_policy(POLICY_PATH) package = policy.packages[0] class OsvClient: def post_json(self, *_args, **_kwargs): return b'{"vulns":[{"id":"OSV-TEST-1"}]}' response_hash, vulnerabilities = controller._query_osv( package, OsvClient(), policy.maximum_metadata_bytes, ) self.assertEqual(len(response_hash), 64) self.assertEqual(vulnerabilities, ("OSV-TEST-1",)) def test_mirror_receipt_never_promotes_runtime(self) -> None: policy = controller.load_policy(POLICY_PATH) verified = _verified_packages(policy) receipt = controller.build_receipt( policy=policy, verified=verified, sbom_sha256="a" * 64, bundle_manifest_sha256="b" * 64, run_id=RUN_ID, trace_id=TRACE_ID, mode="apply", internal_mirror_ref=( "registry.wooo.work/awoooi/mcp-artifacts/playwright-mcp@sha256:" + "c" * 64 ), runtime_mirror_ref=( "192.168.0.110:5000/awoooi/mcp-artifacts/playwright-mcp@sha256:" + "c" * 64 ), ) self.assertEqual( receipt["status"], "completed_internal_mirror_pending_independent_verifier", ) self.assertFalse(receipt["promotion_allowed"]) self.assertFalse(receipt["replay_allowed"]) self.assertFalse(receipt["shadow_allowed"]) self.assertFalse(receipt["canary_allowed"]) self.assertFalse( receipt["stage_receipts"]["bounded_execution"] ["production_route_changed"] ) if __name__ == "__main__": unittest.main()