diff --git a/.gitea/workflows/mcp-external-artifact-mirror.yaml b/.gitea/workflows/mcp-external-artifact-mirror.yaml new file mode 100644 index 000000000..4017c8f30 --- /dev/null +++ b/.gitea/workflows/mcp-external-artifact-mirror.yaml @@ -0,0 +1,248 @@ +# AWOOOI external MCP immutable artifact mirror. +# +# This Gitea-only lane accepts executable bytes from exact npm registry URLs in +# committed policy, verifies them, mirrors a scratch OCI data bundle to Harbor, +# independently reads the digest back without starting a container, then writes +# the two receipts to Gitea main with a normal non-force push. It never installs +# or starts the MCP, opens a browser, writes external content to RAG, or changes a +# production route. + +name: MCP External Artifact Mirror + +on: + workflow_dispatch: + schedule: + - cron: "13 2 * * 3" # Wednesday 10:13 Asia/Taipei: pinned artifact revalidation + push: + branches: + - main + paths: + - config/mcp/playwright-mcp-artifact-policy.json + - scripts/security/external_mcp_artifact_controller.py + - scripts/security/verify_external_mcp_artifact_receipt.py + - scripts/security/tests/test_external_mcp_artifact_controller.py + - scripts/security/tests/test_verify_external_mcp_artifact_receipt.py + - .gitea/workflows/mcp-external-artifact-mirror.yaml + +concurrency: + group: awoooi-mcp-external-artifact-mirror + cancel-in-progress: false + +env: + GITEA_SOURCE_URL: http://192.168.0.110:3001/wooo/awoooi.git + HARBOR_REGISTRY: registry.wooo.work + POLICY_PATH: config/mcp/playwright-mcp-artifact-policy.json + CONTROLLER_RECEIPT_PATH: docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-controller.snapshot.json + VERIFIER_RECEIPT_PATH: docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-verifier.snapshot.json + +jobs: + mirror-and-verify: + runs-on: awoooi-non110-host + timeout-minutes: 25 + steps: + - name: Bootstrap bounded runner tools + run: | + set -euo pipefail + if command -v apk >/dev/null 2>&1; then + apk add --no-cache bash coreutils curl docker-cli docker-cli-buildx git openssl python3 + fi + command -v docker >/dev/null + command -v git >/dev/null + command -v openssl >/dev/null + command -v python3 >/dev/null + + - name: Checkout exact source from Gitea + run: | + set -euo pipefail + git init . + git remote remove origin 2>/dev/null || true + git remote add origin "${GITEA_SOURCE_URL}" + git fetch --no-tags --prune --depth=1 origin "${GITHUB_SHA}" + git checkout --force -B main FETCH_HEAD + + - name: Validate controller verifier and freeze boundaries + run: | + set -euo pipefail + python3 -m py_compile \ + scripts/security/external_mcp_artifact_controller.py \ + scripts/security/verify_external_mcp_artifact_receipt.py \ + scripts/security/tests/test_external_mcp_artifact_controller.py \ + scripts/security/tests/test_verify_external_mcp_artifact_receipt.py + python3 scripts/security/tests/test_external_mcp_artifact_controller.py + python3 scripts/security/tests/test_verify_external_mcp_artifact_receipt.py + python3 - <<'PY' + import json + from pathlib import Path + + paths = [ + Path("config/mcp/playwright-mcp-artifact-policy.json"), + Path(".gitea/workflows/mcp-external-artifact-mirror.yaml"), + ] + text = "\n".join(path.read_text(encoding="utf-8").lower() for path in paths) + forbidden = ( + "uses" + ":", + "actions" + "/checkout", + "github" + "." + "com", + "api" + "." + "github" + "." + "com", + "raw" + "." + "githubusercontent" + "." + "com", + "codeload" + "." + "github" + "." + "com", + "ghcr" + "." + "io", + "npx" + " -y", + "@" + "latest", + ":" + "latest", + ) + hits = [value for value in forbidden if value in text] + if hits: + raise SystemExit(f"freeze_or_floating_reference_violation:{hits}") + policy = json.loads(paths[0].read_text(encoding="utf-8")) + if policy.get("risk_level") != "high": + raise SystemExit("artifact_policy_risk_level_invalid") + if (policy.get("catalog_source") or {}).get("kind") != "discovery_only": + raise SystemExit("marketplace_must_remain_discovery_only") + if any( + (policy.get("promotion_contract") or {}).get(field) is not False + for field in ("deployment_allowed", "shadow_allowed", "canary_allowed") + ): + raise SystemExit("runtime_promotion_must_remain_disabled") + print("freeze_boundary_verified=true") + print("runtime_promotion_enabled=false") + PY + git diff --check + + - name: Wait for bounded host build capacity + env: + HOST_WEB_BUILD_PRESSURE_ATTEMPTS: "18" + HOST_WEB_BUILD_PRESSURE_SLEEP_SECONDS: "10" + run: bash scripts/ci/wait-host-web-build-pressure.sh + + - name: Controlled Harbor mirror and independent readback + env: + HARBOR_PASSWORD: ${{ secrets.HARBOR_PASSWORD }} + HARBOR_USERNAME: ${{ secrets.HARBOR_USERNAME }} + run: | + set -euo pipefail + set +x + cleanup_registry_auth() { + docker logout "${HARBOR_REGISTRY}" >/dev/null 2>&1 || true + } + trap cleanup_registry_auth EXIT + + registry_status="$( + curl --silent --show-error --output /dev/null \ + --write-out '%{http_code}' --max-time 10 \ + "https://${HARBOR_REGISTRY}/v2/" || true + )" + if [ "${registry_status}" != "200" ] && [ "${registry_status}" != "401" ]; then + echo "BLOCKER harbor_registry_readiness_status_${registry_status:-000}" + exit 1 + fi + printf '%s\n' "${HARBOR_PASSWORD}" | \ + docker login "${HARBOR_REGISTRY}" \ + --username "${HARBOR_USERNAME}" \ + --password-stdin >/dev/null + + RUN_ID="$(python3 -c 'import uuid; print(uuid.uuid4())')" + TRACE_ID="mcp-artifact-${RUN_ID}" + python3 scripts/security/external_mcp_artifact_controller.py \ + --policy "${POLICY_PATH}" \ + --run-id "${RUN_ID}" \ + --trace-id "${TRACE_ID}" \ + --receipt "${CONTROLLER_RECEIPT_PATH}" \ + mirror --apply + python3 scripts/security/verify_external_mcp_artifact_receipt.py \ + --policy "${POLICY_PATH}" \ + --controller-receipt "${CONTROLLER_RECEIPT_PATH}" \ + --verifier-receipt "${VERIFIER_RECEIPT_PATH}" + + python3 - <<'PY' + import json + import os + import re + from pathlib import Path + + controller = json.loads(Path(os.environ["CONTROLLER_RECEIPT_PATH"]).read_text()) + verifier = json.loads(Path(os.environ["VERIFIER_RECEIPT_PATH"]).read_text()) + if controller.get("status") != "completed_internal_mirror_pending_independent_verifier": + raise SystemExit("controller_terminal_invalid") + if verifier.get("status") != "completed_internal_mirror_verified": + raise SystemExit("independent_verifier_terminal_invalid") + if controller.get("run_id") != verifier.get("run_id"): + raise SystemExit("controller_verifier_run_id_mismatch") + if controller.get("trace_id") != verifier.get("trace_id"): + raise SystemExit("controller_verifier_trace_id_mismatch") + if any( + verifier.get(field) is not False + for field in ("promotion_allowed", "replay_allowed", "shadow_allowed", "canary_allowed") + ): + raise SystemExit("verifier_promotion_boundary_invalid") + digest = verifier.get("internal_digest") or "" + if not re.fullmatch(r"sha256:[0-9a-f]{64}", digest): + raise SystemExit("verifier_internal_digest_invalid") + print(f"controlled_apply_run_id={verifier['run_id']}") + print(f"controlled_apply_trace_id={verifier['trace_id']}") + print(f"internal_mirror_digest={digest}") + print("container_or_mcp_started=false") + print("production_route_changed=false") + PY + + - name: Commit verified receipts to Gitea main + env: + CD_PUSH_TOKEN: ${{ secrets.CD_PUSH_TOKEN }} + run: | + set -euo pipefail + set +x + test -s "${CONTROLLER_RECEIPT_PATH}" + test -s "${VERIFIER_RECEIPT_PATH}" + git config user.email "mcp-artifact-controller@awoooi.internal" + git config user.name "AWOOOI MCP Artifact Controller" + git add "${CONTROLLER_RECEIPT_PATH}" "${VERIFIER_RECEIPT_PATH}" + git diff --cached --quiet && { + echo "receipt_writeback=no_change" + exit 0 + } + git commit -m "chore(mcp): record verified Playwright artifact mirror [skip ci] [metadata-only]" + + ASKPASS_PATH="$(mktemp)" + export ASKPASS_PATH + python3 - <<'PY' + import os + from pathlib import Path + + path = Path(os.environ["ASKPASS_PATH"]) + path.write_text( + "#!/bin/sh\n" + "case \"$1\" in\n" + " *Username*) printf '%s\\n' wooo ;;\n" + " *Password*) printf '%s\\n' \"$CD_PUSH_TOKEN\" ;;\n" + " *) exit 1 ;;\n" + "esac\n", + encoding="utf-8", + ) + path.chmod(0o700) + PY + cleanup_push_auth() { + rm -f "${ASKPASS_PATH}" + } + trap cleanup_push_auth EXIT + export GIT_ASKPASS="${ASKPASS_PATH}" + export GIT_TERMINAL_PROMPT=0 + git remote remove gitea 2>/dev/null || true + git remote add gitea "${GITEA_SOURCE_URL}" + + for attempt in 1 2 3; do + git fetch --no-tags --depth=100 gitea main + if ! git merge --no-edit gitea/main; then + git merge --abort || true + echo "BLOCKER receipt_writeback_merge_conflict" + exit 1 + fi + if git push gitea HEAD:main; then + echo "receipt_writeback=verified_normal_push" + echo "receipt_commit_sha=$(git rev-parse HEAD)" + exit 0 + fi + echo "receipt_writeback_retry=${attempt}" + sleep 5 + done + echo "BLOCKER receipt_writeback_non_fast_forward_after_bounded_retry" + exit 1 diff --git a/config/mcp/playwright-mcp-artifact-policy.json b/config/mcp/playwright-mcp-artifact-policy.json new file mode 100644 index 000000000..daead51a9 --- /dev/null +++ b/config/mcp/playwright-mcp-artifact-policy.json @@ -0,0 +1,149 @@ +{ + "schema_version": "awoooi_external_mcp_artifact_policy_v1", + "work_item_id": "MCP-ARTIFACT-PLAYWRIGHT-001", + "candidate_id": "external.playwright-verifier", + "risk_level": "high", + "catalog_source": { + "kind": "discovery_only", + "name": "mcp-so", + "url": "https://mcp.so/servers/playwright-mcp", + "executable_supply_chain_authority": false + }, + "registry_contract": { + "registry_origin": "https://registry.npmjs.org", + "keys_url": "https://registry.npmjs.org/-/npm/v1/keys", + "allowed_hosts": [ + "registry.npmjs.org" + ], + "redirects_allowed": false, + "maximum_metadata_bytes": 1048576, + "maximum_tarball_bytes": 8388608, + "maximum_unpacked_bytes": 33554432, + "signature_key": { + "keyid": "SHA256:DhQ8wR5APBvFHLF/+Tc+AYvPOdTpcIDqOhxsBHRwC7U", + "keytype": "ecdsa-sha2-nistp256", + "scheme": "ecdsa-sha2-nistp256", + "expires": null, + "public_key_der_base64": "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEY6Ya7W++7aUPzvMTrezH6Ycx3c+HOKYCcNGybJZSCJq/fd7Qa8uuAKtdIkUQtQiEKERhAmE5lMMJhP8OkDOa2g==" + } + }, + "packages": [ + { + "name": "@playwright/mcp", + "version": "0.0.78", + "metadata_url": "https://registry.npmjs.org/@playwright%2Fmcp/0.0.78", + "tarball_url": "https://registry.npmjs.org/@playwright/mcp/-/mcp-0.0.78.tgz", + "integrity": "sha512-XLTUeA6mEN9sQ+hJ4dfG8EIkDbxS0K3Trc2RBkUJuf02TgE2FQRNTMtq/aJfhyRMINsRl/Ybc4sxcWLtFn4/TQ==", + "shasum_sha1": "305c96c4ac0179bd37622fe4ed4162493513b33e", + "signature": "MEQCIFFsdei1ioeXzOhsBHdVv1XBungfkFzNx6eHS+igHlDaAiBfcETN9LM6vcv8GU8ch9KtODfcx3NXW0HW775r66Rzag==", + "attestations_url": "https://registry.npmjs.org/-/npm/v1/attestations/@playwright%2fmcp@0.0.78", + "slsa_subject_name": "pkg:npm/%40playwright/mcp@0.0.78", + "slsa_subject_sha512_hex": "5cb4d4780ea610df6c43e849e1d7c6f042240dbc52d0add3adcd91064509b9fd364e013615044d4ccb6afda25f87244c20db1197f61b738b317162ed167e3f4d", + "license": "Apache-2.0", + "dependencies": { + "playwright": "1.62.0-alpha-1783623505000", + "playwright-core": "1.62.0-alpha-1783623505000" + } + }, + { + "name": "playwright", + "version": "1.62.0-alpha-1783623505000", + "metadata_url": "https://registry.npmjs.org/playwright/1.62.0-alpha-1783623505000", + "tarball_url": "https://registry.npmjs.org/playwright/-/playwright-1.62.0-alpha-1783623505000.tgz", + "integrity": "sha512-6KV9h4PP3hqu4NaGdxxcijWfYh9LJcFI/R2sP4TTC4I5cFo3oRawN0ETlW5MkE3cQEgKhhoj0KUNz4sfpCT0Tg==", + "shasum_sha1": "d04815de36073de29cc0b02c69bfd657231ec0c1", + "signature": "MEUCIQDw1UimZ6bCEYoTrl+bW4mopC7zomhYdObstPBijNCbMgIgUYnCgVS8CaC3sIfjJwe4RL3RY3SO3Zxy92PcP63vr7w=", + "attestations_url": "https://registry.npmjs.org/-/npm/v1/attestations/playwright@1.62.0-alpha-1783623505000", + "slsa_subject_name": "pkg:npm/playwright@1.62.0-alpha-1783623505000", + "slsa_subject_sha512_hex": "e8a57d8783cfde1aaee0d686771c5c8a359f621f4b25c148fd1dac3f84d30b8239705a37a116b0374113956e4c904ddc40480a861a23d0a50dcf8b1fa424f44e", + "license": "Apache-2.0", + "dependencies": { + "playwright-core": "1.62.0-alpha-1783623505000" + }, + "optional_dependencies": { + "fsevents": "2.3.2" + } + }, + { + "name": "playwright-core", + "version": "1.62.0-alpha-1783623505000", + "metadata_url": "https://registry.npmjs.org/playwright-core/1.62.0-alpha-1783623505000", + "tarball_url": "https://registry.npmjs.org/playwright-core/-/playwright-core-1.62.0-alpha-1783623505000.tgz", + "integrity": "sha512-CPJZdsA/KGT2QQlekiV6Wt+QlQrZHVSZ6oiNtOI/bYYOIVLM8jfKGWTM4zQiyd4UN+40Cq4cA6lxmZHZbtPvJQ==", + "shasum_sha1": "3984745e615ad065b24a89edeeec1a653addc807", + "signature": "MEUCIFhV50jjIBy0rF0fm8SO5i0BHs7OULVIVg64Z+KOng7EAiEAy8QtBfSmtVqDHBXb+1x2VdWuChhvd2GWWaJHGm8O9eU=", + "attestations_url": "https://registry.npmjs.org/-/npm/v1/attestations/playwright-core@1.62.0-alpha-1783623505000", + "slsa_subject_name": "pkg:npm/playwright-core@1.62.0-alpha-1783623505000", + "slsa_subject_sha512_hex": "08f25976c03f2864f641095e92257a5adf90950ad91d5499ea888db4e23f6d860e2152ccf237ca1964cce33422c9de1437ee340aae1c03a9719991d96ed3ef25", + "license": "Apache-2.0", + "dependencies": {} + } + ], + "internal_mirror": { + "artifact_kind": "oci_scratch_bundle", + "push_repository": "registry.wooo.work/awoooi/mcp-artifacts/playwright-mcp", + "runtime_repository": "192.168.0.110:5000/awoooi/mcp-artifacts/playwright-mcp", + "tag": "0.0.78-bundle-v1", + "digest_pin_required": true, + "sbom_required": true, + "upstream_signature_required": true, + "slsa_subject_match_required": true, + "vulnerability_scan_required": true, + "vulnerability_db_max_age_hours": 168, + "allowed_licenses": [ + "Apache-2.0" + ] + }, + "replay_contract": { + "registered_adapter": "playwright_public_accessibility_verifier_v1", + "adapter_implemented": false, + "allowed_origins": [ + "https://awoooi.wooo.work" + ], + "first_canary_url": "https://awoooi.wooo.work/zh-TW/automation/mcp", + "authenticated_session_allowed": false, + "browser_profile_persistence_allowed": false, + "file_upload_allowed": false, + "download_allowed": false, + "pdf_write_allowed": false, + "browser_install_allowed": false, + "arbitrary_navigation_allowed": false, + "production_form_submit_allowed": false, + "production_write_allowed": false, + "screenshot_and_accessibility_receipts_only": true + }, + "promotion_contract": { + "deployment_allowed": false, + "shadow_allowed": false, + "canary_allowed": false, + "required_before_shadow": [ + "internal_mirror_digest_verified", + "cyclonedx_sbom_verified", + "npm_registry_signatures_verified", + "slsa_subjects_verified", + "vulnerability_decision_passed", + "license_decision_passed", + "registered_replay_adapter_verified", + "rollback_receipt_verified" + ], + "rollback": { + "action": "disable_candidate_and_retain_immutable_bundle", + "fallback_verifier": "awoooi_repo_owned_playwright_test_runner", + "external_artifact_delete_allowed": false, + "production_route_change_allowed": false + } + }, + "prohibited": [ + "github_api_or_github_actions", + "actions_checkout", + "github_source_archive_or_ghcr", + "npx_y", + "at_latest", + "floating_container_tag", + "unverified_redirect", + "secret_read_or_log", + "request_or_response_body_storage", + "external_rag_write", + "production_mutation" + ] +} diff --git a/docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-check.snapshot.json b/docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-check.snapshot.json new file mode 100644 index 000000000..0b97de270 --- /dev/null +++ b/docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-check.snapshot.json @@ -0,0 +1 @@ +{"bundle_manifest_sha256":"54b1d7ff24fc6f8ca89d4d36adcdb43e1cf79ed6018a4cd4f0d630c857bffb2c","canary_allowed":false,"candidate_id":"external.playwright-verifier","cyclonedx_sbom_sha256":"092500242530e40f361f52cc662edda9e6c83f76edf81dfe9574235546c22be7","internal_mirror_ref":null,"mode":"check","observed_at":"2026-07-15T01:42:22.303666+00:00","package_count":3,"packages":[{"archive_file_count":7,"archive_safety_verified":true,"attestation_sha256":"4830a73d5f551fefc969e297b042cbd956f88c954fe0f02c11101a995b03ede1","license_decision":"passed","metadata_sha256":"8e6bbce53663e1c9488e02a748e2b20e8898fb58cd4ec745e09339bc546599b3","name":"@playwright/mcp","npm_registry_signature_verified":true,"slsa_subject_verified":true,"tarball_sha1":"305c96c4ac0179bd37622fe4ed4162493513b33e","tarball_sha512_hex":"5cb4d4780ea610df6c43e849e1d7c6f042240dbc52d0add3adcd91064509b9fd364e013615044d4ccb6afda25f87244c20db1197f61b738b317162ed167e3f4d","unpacked_bytes":84517,"version":"0.0.78","vulnerability_decision":"passed_no_known_osv_records","vulnerability_ids":[]},{"archive_file_count":62,"archive_safety_verified":true,"attestation_sha256":"f9188fc1fc80c15cb8cc0c962f13c56cf38ebba5e5cf5092565c258b464d7df3","license_decision":"passed","metadata_sha256":"032e79d40dd69d8f573b6b2a6b7bb48dbf331e8dc642db625d33598dd5c478d4","name":"playwright","npm_registry_signature_verified":true,"slsa_subject_verified":true,"tarball_sha1":"d04815de36073de29cc0b02c69bfd657231ec0c1","tarball_sha512_hex":"e8a57d8783cfde1aaee0d686771c5c8a359f621f4b25c148fd1dac3f84d30b8239705a37a116b0374113956e4c904ddc40480a861a23d0a50dcf8b1fa424f44e","unpacked_bytes":5062430,"version":"1.62.0-alpha-1783623505000","vulnerability_decision":"passed_no_known_osv_records","vulnerability_ids":[]},{"archive_file_count":104,"archive_safety_verified":true,"attestation_sha256":"99df3734d80709dc8b7826c661e4bc05a57bf3fbe61c4c838921d5c04df9e02b","license_decision":"passed","metadata_sha256":"1477249209c785ce155309e4e4d210aaa1e74ec6ef172115da310763b8024505","name":"playwright-core","npm_registry_signature_verified":true,"slsa_subject_verified":true,"tarball_sha1":"3984745e615ad065b24a89edeeec1a653addc807","tarball_sha512_hex":"08f25976c03f2864f641095e92257a5adf90950ad91d5499ea888db4e23f6d860e2152ccf237ca1964cce33422c9de1437ee340aae1c03a9719991d96ed3ef25","unpacked_bytes":12815408,"version":"1.62.0-alpha-1783623505000","vulnerability_decision":"passed_no_known_osv_records","vulnerability_ids":[]}],"policy_checksum":"sha256:44c9f79c2af59f1d2703cbd2181b43f4c4dd4ee9f70164f606bdda70d823e78b","promotion_allowed":false,"replay_allowed":false,"risk_level":"high","run_id":"26aa5be0-b71c-40fd-8244-ac74c53fdd86","runtime_mirror_ref":null,"schema_version":"awoooi_external_mcp_artifact_receipt_v1","shadow_allowed":false,"stage_receipts":{"bounded_execution":{"external_runtime_mutation_performed":false,"internal_registry_write_performed":false,"production_route_changed":false,"status":"not_executed"},"check_mode":{"browser_started":false,"mcp_process_started":false,"production_write_performed":false,"status":"passed"},"decision":{"candidate_action":"run_independent_check_verifier_then_mirror_exact_bundle","runtime_promotion_allowed":false},"independent_post_verifier":{"bundle_manifest_sha256":"54b1d7ff24fc6f8ca89d4d36adcdb43e1cf79ed6018a4cd4f0d630c857bffb2c","controller_registry_digest_check":false,"cyclonedx_sbom_sha256":"092500242530e40f361f52cc662edda9e6c83f76edf81dfe9574235546c22be7","internal_digest_ref_verified":false,"package_digest_count":3,"status":"pending_external_verifier"},"learning_writeback":{"km_write_performed":false,"rag_write_performed":false,"status":"receipt_ready_for_committed_writeback"},"normalized_asset_identity":{"candidate_id":"external.playwright-verifier","exact_root_version":"0.0.78","package_count":3},"risk_policy":{"exact_versions":true,"immutable_upstream_digests":true,"license_decision":"passed","npm_registry_signatures_verified":true,"risk_level":"high","slsa_subjects_verified":true,"vulnerability_decision":"passed_no_known_osv_records"},"rollback_or_no_write_terminal":{"external_artifact_delete_allowed":false,"production_route_change_allowed":false,"rollback_action":"disable_candidate_and_retain_immutable_bundle","status":"no_write_pending_external_verifier"},"sensor_source_receipt":{"osv":"api.osv.dev","redirects_followed":0,"registry":"registry.npmjs.org","runtime_request_or_response_body_stored":false,"upstream_supply_chain_attestation_bundled":true},"source_of_truth_diff":{"change_state":"baseline_created","policy_checksum":"sha256:44c9f79c2af59f1d2703cbd2181b43f4c4dd4ee9f70164f606bdda70d823e78b"}},"status":"completed_check_mode_pending_independent_verifier","trace_id":"mcp-artifact-26aa5be0-b71c-40fd-8244-ac74c53fdd86","work_item_id":"MCP-ARTIFACT-PLAYWRIGHT-001"} diff --git a/docs/operations/mcp-control-plane-catalog.snapshot.json b/docs/operations/mcp-control-plane-catalog.snapshot.json index 67186d1af..312253008 100644 --- a/docs/operations/mcp-control-plane-catalog.snapshot.json +++ b/docs/operations/mcp-control-plane-catalog.snapshot.json @@ -294,22 +294,26 @@ "title": "Playwright UI verifier MCP", "source_type": "external", "catalog_source": "mcp-so", - "catalog_url": "https://mcp.so/server/playwright-mcp/microsoft", + "catalog_url": "https://mcp.so/servers/playwright-mcp", "decision": "internalize_then_canary", "risk_tier": "high", "scope": "allowlisted_origins_only", "data_boundary": "public_and_test_surfaces", "deployment_allowed": false, - "version_state": "unresolved", - "digest_state": "missing", - "sbom_state": "missing", - "signature_state": "missing", + "version_state": "exact_0.0.78_check_verified", + "digest_state": "upstream_sha512_pinned_internal_mirror_pending", + "sbom_state": "cyclonedx_1.5_check_verified_internal_mirror_pending", + "signature_state": "npm_registry_ecdsa_and_slsa_subject_verified", "rag_policy": "screenshots_and_accessibility_receipts_only", "blockers": [ - "floating_install_examples_rejected", - "origin_allowlist_and_session_redaction_verifier_missing" + "internal_mirror_and_independent_digest_readback_pending", + "registered_public_origin_replay_adapter_missing", + "shadow_canary_verifier_and_rollback_execution_pending" ], - "evidence_refs": [] + "evidence_refs": [ + "config/mcp/playwright-mcp-artifact-policy.json", + "docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-check.snapshot.json" + ] }, { "capability_id": "external.google-search-console-readonly", diff --git a/scripts/security/external_mcp_artifact_controller.py b/scripts/security/external_mcp_artifact_controller.py new file mode 100644 index 000000000..fce2da11b --- /dev/null +++ b/scripts/security/external_mcp_artifact_controller.py @@ -0,0 +1,1330 @@ +#!/usr/bin/env python3 +"""Verify and mirror an exact external MCP artifact bundle. + +The controller treats marketplace pages as discovery-only input. Executable +artifacts are accepted only from the exact allowlisted registry URLs committed +in policy, after digest, npm registry signature, SLSA subject, dependency, +license, archive-safety, and OSV checks pass. The mirror target is a scratch OCI +bundle in the internal Harbor registry; this controller never starts the MCP +server, a browser, a replay, or a production canary. +""" + +from __future__ import annotations + +import argparse +import base64 +import binascii +import hashlib +import io +import json +import re +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, Protocol +from urllib.error import HTTPError, URLError +from urllib.parse import quote, urlparse +from urllib.request import HTTPRedirectHandler, Request, build_opener + + +POLICY_SCHEMA_VERSION = "awoooi_external_mcp_artifact_policy_v1" +RECEIPT_SCHEMA_VERSION = "awoooi_external_mcp_artifact_receipt_v1" +SBOM_SPEC_VERSION = "1.5" +SLSA_PREDICATE_TYPE = "https://slsa.dev/provenance/v1" +IN_TOTO_STATEMENT_TYPE = "https://in-toto.io/Statement/v1" +OSV_QUERY_URL = "https://api.osv.dev/v1/query" +OSV_HOST = "api.osv.dev" +TRACE_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{3,160}$") +SHA1_PATTERN = re.compile(r"^[0-9a-f]{40}$") +SHA256_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") +SHA512_HEX_PATTERN = re.compile(r"^[0-9a-f]{128}$") +VERSION_PATTERN = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+(?:-[0-9A-Za-z.-]+)?$") +VULNERABILITY_ID_PATTERN = re.compile(r"^[A-Za-z0-9._:+-]{1,160}$") +REQUIRED_PACKAGE_NAMES = { + "@playwright/mcp", + "playwright", + "playwright-core", +} + + +class ControllerError(RuntimeError): + """Fail-closed error containing only a public-safe error class.""" + + +class HTTPClient(Protocol): + def get_bytes(self, url: str, *, maximum_bytes: int) -> bytes: ... + + def post_json( + self, + url: str, + payload: dict[str, Any], + *, + maximum_bytes: int, + ) -> bytes: ... + + +class _NoRedirect(HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 + return None + + +class BoundedHTTPClient: + """HTTPS client with redirects disabled and bounded response reads.""" + + def __init__(self, allowed_hosts: frozenset[str], timeout_seconds: int = 20): + self._allowed_hosts = allowed_hosts + self._timeout_seconds = timeout_seconds + self._opener = build_opener(_NoRedirect()) + + def get_bytes(self, url: str, *, maximum_bytes: int) -> bytes: + self._validate_url(url) + request = Request( + url, + method="GET", + headers={ + "Accept": "application/json, application/octet-stream", + "User-Agent": "awoooi-external-mcp-artifact-controller/1", + }, + ) + return self._read(request, maximum_bytes) + + def post_json( + self, + url: str, + payload: dict[str, Any], + *, + maximum_bytes: int, + ) -> bytes: + self._validate_url(url) + request = Request( + url, + method="POST", + data=_canonical_json_bytes(payload), + headers={ + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "awoooi-external-mcp-artifact-controller/1", + }, + ) + return self._read(request, maximum_bytes) + + def _validate_url(self, url: str) -> None: + parsed = urlparse(url) + try: + port = parsed.port + except ValueError as exc: + raise ControllerError("source_url_invalid") from exc + if ( + parsed.scheme != "https" + or (parsed.hostname or "").lower() not in self._allowed_hosts + or parsed.username is not None + or parsed.password is not None + or port not in {None, 443} + or parsed.query + or parsed.fragment + ): + raise ControllerError("source_url_not_allowlisted") + + def _read(self, request: Request, maximum_bytes: int) -> bytes: + try: + with self._opener.open(request, timeout=self._timeout_seconds) as response: + if response.status != 200: + raise ControllerError("source_http_status_invalid") + data = response.read(maximum_bytes + 1) + except HTTPError as exc: + if 300 <= exc.code < 400: + raise ControllerError("source_redirect_rejected") from exc + raise ControllerError(f"source_http_{exc.code}") from exc + except (URLError, TimeoutError, OSError) as exc: + raise ControllerError("source_unavailable") from exc + if len(data) > maximum_bytes: + raise ControllerError("source_payload_too_large") + return data + + +@dataclass(frozen=True) +class PackagePolicy: + name: str + version: str + metadata_url: str + tarball_url: str + integrity: str + sha512_hex: str + shasum_sha1: str + signature: str + attestations_url: str + slsa_subject_name: str + slsa_subject_sha512_hex: str + license: str + dependencies: dict[str, str] + optional_dependencies: dict[str, str] + + @property + def filename(self) -> str: + safe_name = self.name.replace("@", "").replace("/", "-") + return f"{safe_name}-{self.version}.tgz" + + @property + def bom_ref(self) -> str: + return f"pkg:npm/{quote(self.name, safe='/')}@{self.version}" + + +@dataclass(frozen=True) +class Policy: + work_item_id: str + candidate_id: str + risk_level: str + allowed_hosts: frozenset[str] + maximum_metadata_bytes: int + maximum_tarball_bytes: int + maximum_unpacked_bytes: int + signature_keyid: str + signature_keytype: str + signature_scheme: str + signature_public_key_der_base64: str + packages: tuple[PackagePolicy, ...] + push_repository: str + runtime_repository: str + mirror_tag: str + allowed_licenses: frozenset[str] + checksum: str + + @property + def push_ref(self) -> str: + return f"{self.push_repository}:{self.mirror_tag}" + + +@dataclass(frozen=True) +class VerifiedPackage: + policy: PackagePolicy + tarball: bytes + attestation: bytes + metadata_sha256: str + tarball_sha512_hex: str + tarball_sha1: str + attestation_sha256: str + unpacked_bytes: int + archive_file_count: int + license_file_count: int + osv_response_sha256: str + vulnerability_ids: tuple[str, ...] + + +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 _require_text(value: Any, label: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ControllerError(f"invalid_{label}") + return value.strip() + + +def _require_positive_int(value: Any, label: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ControllerError(f"invalid_{label}") + return value + + +def _parse_json(data: bytes, error_class: str) -> dict[str, Any]: + try: + payload = json.loads(data.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ControllerError(error_class) from exc + if not isinstance(payload, dict): + raise ControllerError(error_class) + return payload + + +def _parse_sha512_integrity(value: Any, label: str) -> tuple[str, str]: + integrity = _require_text(value, label) + if not integrity.startswith("sha512-"): + raise ControllerError(f"invalid_{label}") + try: + digest = base64.b64decode(integrity.removeprefix("sha512-"), validate=True) + except (ValueError, binascii.Error) as exc: + raise ControllerError(f"invalid_{label}") from exc + if len(digest) != 64: + raise ControllerError(f"invalid_{label}") + return integrity, digest.hex() + + +def _validate_exact_version(value: Any, label: str) -> str: + version = _require_text(value, label) + if ( + not VERSION_PATTERN.fullmatch(version) + or "latest" in version.lower() + or version.startswith(("^", "~", ">", "<", "*")) + ): + raise ControllerError(f"invalid_{label}") + return version + + +def _validate_https_url(url: str, allowed_hosts: frozenset[str], label: str) -> None: + parsed = urlparse(url) + try: + port = parsed.port + except ValueError as exc: + raise ControllerError(f"invalid_{label}") from exc + if ( + parsed.scheme != "https" + or (parsed.hostname or "").lower() not in allowed_hosts + or parsed.username is not None + or parsed.password is not None + or port not in {None, 443} + or parsed.query + or parsed.fragment + ): + raise ControllerError(f"invalid_{label}") + + +def load_policy(path: Path) -> Policy: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ControllerError("policy_unreadable") from exc + if not isinstance(payload, dict) or payload.get("schema_version") != POLICY_SCHEMA_VERSION: + raise ControllerError("policy_schema_invalid") + + work_item_id = _require_text(payload.get("work_item_id"), "work_item_id") + candidate_id = _require_text(payload.get("candidate_id"), "candidate_id") + if candidate_id != "external.playwright-verifier": + raise ControllerError("candidate_id_not_allowlisted") + if payload.get("risk_level") != "high": + raise ControllerError("risk_level_invalid") + + catalog = payload.get("catalog_source") + if not isinstance(catalog, dict): + raise ControllerError("catalog_source_invalid") + if ( + catalog.get("kind") != "discovery_only" + or catalog.get("executable_supply_chain_authority") is not False + ): + raise ControllerError("catalog_must_be_discovery_only") + + registry = payload.get("registry_contract") + if not isinstance(registry, dict): + raise ControllerError("registry_contract_invalid") + allowed_hosts = registry.get("allowed_hosts") + if allowed_hosts != ["registry.npmjs.org"]: + raise ControllerError("registry_hosts_invalid") + host_set = frozenset(allowed_hosts) + if registry.get("redirects_allowed") is not False: + raise ControllerError("registry_redirect_policy_invalid") + registry_origin = _require_text(registry.get("registry_origin"), "registry_origin") + keys_url = _require_text(registry.get("keys_url"), "keys_url") + _validate_https_url(registry_origin, host_set, "registry_origin") + _validate_https_url(keys_url, host_set, "keys_url") + if urlparse(registry_origin).path not in {"", "/"}: + raise ControllerError("registry_origin_invalid") + if urlparse(keys_url).path != "/-/npm/v1/keys": + raise ControllerError("keys_url_invalid") + + signature_key = registry.get("signature_key") + if not isinstance(signature_key, dict): + raise ControllerError("signature_key_invalid") + signature_keyid = _require_text(signature_key.get("keyid"), "signature_keyid") + signature_keytype = _require_text( + signature_key.get("keytype"), "signature_keytype" + ) + signature_scheme = _require_text( + signature_key.get("scheme"), "signature_scheme" + ) + if signature_keytype != "ecdsa-sha2-nistp256" or signature_scheme != signature_keytype: + raise ControllerError("signature_scheme_not_allowlisted") + public_key = _require_text( + signature_key.get("public_key_der_base64"), "signature_public_key" + ) + try: + decoded_key = base64.b64decode(public_key, validate=True) + except (ValueError, binascii.Error) as exc: + raise ControllerError("signature_public_key_invalid") from exc + if len(decoded_key) < 80: + raise ControllerError("signature_public_key_invalid") + + rows = payload.get("packages") + if not isinstance(rows, list) or len(rows) != len(REQUIRED_PACKAGE_NAMES): + raise ControllerError("package_set_invalid") + packages: list[PackagePolicy] = [] + seen: set[str] = set() + for row in rows: + if not isinstance(row, dict): + raise ControllerError("package_policy_invalid") + name = _require_text(row.get("name"), "package_name") + if name in seen: + raise ControllerError("duplicate_package_name") + seen.add(name) + version = _validate_exact_version(row.get("version"), "package_version") + integrity, sha512_hex = _parse_sha512_integrity( + row.get("integrity"), "package_integrity" + ) + shasum = _require_text(row.get("shasum_sha1"), "package_shasum") + if not SHA1_PATTERN.fullmatch(shasum): + raise ControllerError("package_shasum_invalid") + slsa_sha512 = _require_text( + row.get("slsa_subject_sha512_hex"), "slsa_subject_sha512" + ) + if not SHA512_HEX_PATTERN.fullmatch(slsa_sha512) or slsa_sha512 != sha512_hex: + raise ControllerError("slsa_subject_digest_mismatch") + metadata_url = _require_text(row.get("metadata_url"), "metadata_url") + tarball_url = _require_text(row.get("tarball_url"), "tarball_url") + attestations_url = _require_text( + row.get("attestations_url"), "attestations_url" + ) + for url, label in ( + (metadata_url, "metadata_url"), + (tarball_url, "tarball_url"), + (attestations_url, "attestations_url"), + ): + _validate_https_url(url, host_set, label) + dependencies = _validate_dependency_map(row.get("dependencies", {})) + optional_dependencies = _validate_dependency_map( + row.get("optional_dependencies", {}) + ) + packages.append( + PackagePolicy( + name=name, + version=version, + metadata_url=metadata_url, + tarball_url=tarball_url, + integrity=integrity, + sha512_hex=sha512_hex, + shasum_sha1=shasum, + signature=_require_text(row.get("signature"), "package_signature"), + attestations_url=attestations_url, + slsa_subject_name=_require_text( + row.get("slsa_subject_name"), "slsa_subject_name" + ), + slsa_subject_sha512_hex=slsa_sha512, + license=_require_text(row.get("license"), "package_license"), + dependencies=dependencies, + optional_dependencies=optional_dependencies, + ) + ) + if seen != REQUIRED_PACKAGE_NAMES: + raise ControllerError("package_set_invalid") + package_versions = {item.name: item.version for item in packages} + for package in packages: + for dependency, version in package.dependencies.items(): + if package_versions.get(dependency) != version: + raise ControllerError("dependency_closure_invalid") + + mirror = payload.get("internal_mirror") + if not isinstance(mirror, dict): + raise ControllerError("internal_mirror_invalid") + if ( + mirror.get("artifact_kind") != "oci_scratch_bundle" + or mirror.get("digest_pin_required") is not True + or mirror.get("sbom_required") is not True + or mirror.get("upstream_signature_required") is not True + or mirror.get("slsa_subject_match_required") is not True + or mirror.get("vulnerability_scan_required") is not True + ): + raise ControllerError("internal_mirror_gate_invalid") + push_repository = _require_text( + mirror.get("push_repository"), "push_repository" + ) + runtime_repository = _require_text( + mirror.get("runtime_repository"), "runtime_repository" + ) + if push_repository != "registry.wooo.work/awoooi/mcp-artifacts/playwright-mcp": + raise ControllerError("push_repository_not_internal") + if runtime_repository != "192.168.0.110:5000/awoooi/mcp-artifacts/playwright-mcp": + raise ControllerError("runtime_repository_not_internal") + mirror_tag = _require_text(mirror.get("tag"), "mirror_tag") + if mirror_tag != "0.0.78-bundle-v1" or "latest" in mirror_tag.lower(): + raise ControllerError("mirror_tag_invalid") + allowed_licenses = mirror.get("allowed_licenses") + if allowed_licenses != ["Apache-2.0"]: + raise ControllerError("allowed_licenses_invalid") + for package in packages: + if package.license not in allowed_licenses: + raise ControllerError("package_license_not_allowed") + + replay = payload.get("replay_contract") + promotion = payload.get("promotion_contract") + if not isinstance(replay, dict) or not isinstance(promotion, dict): + raise ControllerError("promotion_contract_invalid") + required_false = ( + "authenticated_session_allowed", + "browser_profile_persistence_allowed", + "file_upload_allowed", + "download_allowed", + "pdf_write_allowed", + "browser_install_allowed", + "arbitrary_navigation_allowed", + "production_form_submit_allowed", + "production_write_allowed", + ) + if any(replay.get(field) is not False for field in required_false): + raise ControllerError("replay_boundary_invalid") + if any( + promotion.get(field) is not False + for field in ("deployment_allowed", "shadow_allowed", "canary_allowed") + ): + raise ControllerError("promotion_must_remain_disabled") + prohibited = set(payload.get("prohibited") or []) + if not { + "github_api_or_github_actions", + "actions_checkout", + "github_source_archive_or_ghcr", + "npx_y", + "at_latest", + "floating_container_tag", + "secret_read_or_log", + "request_or_response_body_storage", + "external_rag_write", + "production_mutation", + }.issubset(prohibited): + raise ControllerError("prohibited_actions_incomplete") + + return Policy( + work_item_id=work_item_id, + candidate_id=candidate_id, + risk_level="high", + allowed_hosts=host_set, + maximum_metadata_bytes=_require_positive_int( + registry.get("maximum_metadata_bytes"), "maximum_metadata_bytes" + ), + maximum_tarball_bytes=_require_positive_int( + registry.get("maximum_tarball_bytes"), "maximum_tarball_bytes" + ), + maximum_unpacked_bytes=_require_positive_int( + registry.get("maximum_unpacked_bytes"), "maximum_unpacked_bytes" + ), + signature_keyid=signature_keyid, + signature_keytype=signature_keytype, + signature_scheme=signature_scheme, + signature_public_key_der_base64=public_key, + packages=tuple(packages), + push_repository=push_repository, + runtime_repository=runtime_repository, + mirror_tag=mirror_tag, + allowed_licenses=frozenset(allowed_licenses), + checksum="sha256:" + _sha256_hex(_canonical_json_bytes(payload)), + ) + + +def _validate_dependency_map(value: Any) -> dict[str, str]: + if not isinstance(value, dict): + raise ControllerError("dependency_map_invalid") + result: dict[str, str] = {} + for name, version in value.items(): + dependency = _require_text(name, "dependency_name") + result[dependency] = _validate_exact_version(version, "dependency_version") + return dict(sorted(result.items())) + + +def _verify_registry_key(policy: Policy, payload: dict[str, Any]) -> None: + rows = payload.get("keys") + if not isinstance(rows, list): + raise ControllerError("registry_keys_invalid") + matches = [ + row + for row in rows + if isinstance(row, dict) and row.get("keyid") == policy.signature_keyid + ] + if len(matches) != 1: + raise ControllerError("registry_signature_key_missing") + key = matches[0] + if ( + key.get("keytype") != policy.signature_keytype + or key.get("scheme") != policy.signature_scheme + or key.get("key") != policy.signature_public_key_der_base64 + or key.get("expires") is not None + ): + raise ControllerError("registry_signature_key_drift") + + +def verify_npm_signature( + *, + public_key_der_base64: str, + signature_base64: str, + message: bytes, +) -> None: + try: + key_bytes = base64.b64decode(public_key_der_base64, validate=True) + signature = base64.b64decode(signature_base64, validate=True) + except (ValueError, binascii.Error) as exc: + raise ControllerError("npm_signature_encoding_invalid") from exc + try: + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import ec + + public_key = serialization.load_der_public_key(key_bytes) + public_key.verify(signature, message, ec.ECDSA(hashes.SHA256())) + return + except ImportError: + pass + except Exception as exc: # cryptography exposes several backend error types + raise ControllerError("npm_signature_verification_failed") from exc + + openssl = shutil.which("openssl") + if not openssl: + raise ControllerError("npm_signature_verifier_unavailable") + with tempfile.TemporaryDirectory(prefix="awoooi-npm-signature-") as directory: + root = Path(directory) + der_path = root / "public.der" + pem_path = root / "public.pem" + signature_path = root / "signature.der" + message_path = root / "message.bin" + der_path.write_bytes(key_bytes) + signature_path.write_bytes(signature) + message_path.write_bytes(message) + convert = subprocess.run( + [ + openssl, + "pkey", + "-pubin", + "-inform", + "DER", + "-in", + str(der_path), + "-out", + str(pem_path), + ], + check=False, + capture_output=True, + text=True, + timeout=10, + ) + if convert.returncode != 0: + raise ControllerError("npm_signature_public_key_invalid") + verified = subprocess.run( + [ + openssl, + "dgst", + "-sha256", + "-verify", + str(pem_path), + "-signature", + str(signature_path), + str(message_path), + ], + check=False, + capture_output=True, + text=True, + timeout=10, + ) + if verified.returncode != 0: + raise ControllerError("npm_signature_verification_failed") + + +def _validate_metadata( + package: PackagePolicy, + metadata: dict[str, Any], + policy: Policy, +) -> None: + if metadata.get("name") != package.name or metadata.get("version") != package.version: + raise ControllerError("registry_package_identity_mismatch") + if metadata.get("license") != package.license: + raise ControllerError("registry_package_license_mismatch") + if _validate_dependency_map(metadata.get("dependencies", {})) != package.dependencies: + raise ControllerError("registry_dependency_drift") + if ( + _validate_dependency_map(metadata.get("optionalDependencies", {})) + != package.optional_dependencies + ): + raise ControllerError("registry_optional_dependency_drift") + dist = metadata.get("dist") + if not isinstance(dist, dict): + raise ControllerError("registry_dist_missing") + if ( + dist.get("integrity") != package.integrity + or dist.get("shasum") != package.shasum_sha1 + or dist.get("tarball") != package.tarball_url + ): + raise ControllerError("registry_dist_drift") + signatures = dist.get("signatures") + expected_signature = { + "keyid": policy.signature_keyid, + "sig": package.signature, + } + if not isinstance(signatures, list) or expected_signature not in signatures: + raise ControllerError("registry_package_signature_drift") + attestations = dist.get("attestations") + if not isinstance(attestations, dict): + raise ControllerError("registry_attestations_missing") + if ( + attestations.get("url") != package.attestations_url + or not isinstance(attestations.get("provenance"), dict) + or attestations["provenance"].get("predicateType") != SLSA_PREDICATE_TYPE + ): + raise ControllerError("registry_attestation_drift") + message = f"{package.name}@{package.version}:{package.integrity}".encode() + verify_npm_signature( + public_key_der_base64=policy.signature_public_key_der_base64, + signature_base64=package.signature, + message=message, + ) + + +def validate_slsa_attestation( + package: PackagePolicy, + payload: dict[str, Any], +) -> None: + rows = payload.get("attestations") + if not isinstance(rows, list): + raise ControllerError("slsa_attestation_invalid") + slsa_rows = [ + row + for row in rows + if isinstance(row, dict) and row.get("predicateType") == SLSA_PREDICATE_TYPE + ] + if len(slsa_rows) != 1: + raise ControllerError("slsa_attestation_missing") + bundle = slsa_rows[0].get("bundle") + envelope = bundle.get("dsseEnvelope") if isinstance(bundle, dict) else None + encoded = envelope.get("payload") if isinstance(envelope, dict) else None + if not isinstance(encoded, str): + raise ControllerError("slsa_dsse_payload_missing") + try: + statement = json.loads(base64.b64decode(encoded, validate=True)) + except (ValueError, binascii.Error, json.JSONDecodeError) as exc: + raise ControllerError("slsa_dsse_payload_invalid") from exc + if not isinstance(statement, dict): + raise ControllerError("slsa_statement_invalid") + if ( + statement.get("_type") != IN_TOTO_STATEMENT_TYPE + or statement.get("predicateType") != SLSA_PREDICATE_TYPE + ): + raise ControllerError("slsa_statement_type_invalid") + subjects = statement.get("subject") + expected_subject = { + "name": package.slsa_subject_name, + "digest": {"sha512": package.slsa_subject_sha512_hex}, + } + if subjects != [expected_subject]: + raise ControllerError("slsa_subject_mismatch") + + +def inspect_tarball( + package: PackagePolicy, + data: bytes, + *, + maximum_unpacked_bytes: int, +) -> tuple[int, int, int]: + try: + archive = tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") + except (tarfile.TarError, OSError) as exc: + raise ControllerError("package_tarball_invalid") from exc + unpacked_bytes = 0 + file_count = 0 + license_file_count = 0 + seen_files: set[str] = set() + package_json: dict[str, Any] | None = None + with archive: + members = archive.getmembers() + if not members or len(members) > 1024: + raise ControllerError("package_archive_entry_count_invalid") + for member in members: + path = PurePosixPath(member.name) + if ( + path.is_absolute() + or ".." in path.parts + or not path.parts + or path.parts[0] != "package" + or member.issym() + or member.islnk() + or member.isdev() + ): + raise ControllerError("package_archive_unsafe_entry") + if member.isfile(): + normalized_path = str(path) + if member.size < 0 or normalized_path in seen_files: + raise ControllerError("package_archive_duplicate_or_invalid_file") + seen_files.add(normalized_path) + file_count += 1 + unpacked_bytes += member.size + if unpacked_bytes > maximum_unpacked_bytes: + raise ControllerError("package_archive_unpacked_size_exceeded") + basename = path.name.lower() + if basename.startswith(("license", "licence", "copying")): + license_file_count += 1 + if path == PurePosixPath("package/package.json"): + extracted = archive.extractfile(member) + if extracted is None: + raise ControllerError("package_manifest_unreadable") + package_json = _parse_json( + extracted.read(262_145), "package_manifest_invalid" + ) + if package_json is None: + raise ControllerError("package_manifest_missing") + if ( + package_json.get("name") != package.name + or package_json.get("version") != package.version + or package_json.get("license") != package.license + ): + raise ControllerError("package_manifest_identity_mismatch") + if _validate_dependency_map(package_json.get("dependencies", {})) != package.dependencies: + raise ControllerError("package_manifest_dependency_drift") + if ( + _validate_dependency_map(package_json.get("optionalDependencies", {})) + != package.optional_dependencies + ): + raise ControllerError("package_manifest_optional_dependency_drift") + if license_file_count == 0: + raise ControllerError("package_license_file_missing") + return unpacked_bytes, file_count, license_file_count + + +def _query_osv( + package: PackagePolicy, + http: HTTPClient, + maximum_bytes: int, +) -> tuple[str, tuple[str, ...]]: + data = http.post_json( + OSV_QUERY_URL, + { + "package": {"ecosystem": "npm", "name": package.name}, + "version": package.version, + }, + maximum_bytes=maximum_bytes, + ) + payload = _parse_json(data, "osv_response_invalid") + rows = payload.get("vulns", []) + if not isinstance(rows, list): + raise ControllerError("osv_response_invalid") + vulnerability_ids: list[str] = [] + for row in rows: + identifier = row.get("id") if isinstance(row, dict) else None + if not isinstance(identifier, str) or not VULNERABILITY_ID_PATTERN.fullmatch( + identifier + ): + raise ControllerError("osv_vulnerability_id_invalid") + vulnerability_ids.append(identifier) + return _sha256_hex(data), tuple(sorted(set(vulnerability_ids))) + + +def verify_upstream_bundle(policy: Policy, http: HTTPClient) -> tuple[VerifiedPackage, ...]: + keys_bytes = http.get_bytes( + "https://registry.npmjs.org/-/npm/v1/keys", + maximum_bytes=policy.maximum_metadata_bytes, + ) + _verify_registry_key(policy, _parse_json(keys_bytes, "registry_keys_invalid")) + verified: list[VerifiedPackage] = [] + for package in policy.packages: + metadata_bytes = http.get_bytes( + package.metadata_url, + maximum_bytes=policy.maximum_metadata_bytes, + ) + metadata = _parse_json(metadata_bytes, "registry_metadata_invalid") + _validate_metadata(package, metadata, policy) + attestation_bytes = http.get_bytes( + package.attestations_url, + maximum_bytes=policy.maximum_metadata_bytes, + ) + validate_slsa_attestation( + package, + _parse_json(attestation_bytes, "slsa_attestation_invalid"), + ) + tarball = http.get_bytes( + package.tarball_url, + maximum_bytes=policy.maximum_tarball_bytes, + ) + tarball_sha512 = hashlib.sha512(tarball).hexdigest() + tarball_sha1 = hashlib.sha1(tarball, usedforsecurity=False).hexdigest() + if ( + tarball_sha512 != package.sha512_hex + or tarball_sha1 != package.shasum_sha1 + ): + raise ControllerError("package_tarball_digest_mismatch") + unpacked_bytes, archive_file_count, license_file_count = inspect_tarball( + package, + tarball, + maximum_unpacked_bytes=policy.maximum_unpacked_bytes, + ) + osv_sha256, vulnerability_ids = _query_osv( + package, + http, + policy.maximum_metadata_bytes, + ) + if vulnerability_ids: + raise ControllerError("vulnerability_decision_blocked") + verified.append( + VerifiedPackage( + policy=package, + tarball=tarball, + attestation=attestation_bytes, + metadata_sha256=_sha256_hex(metadata_bytes), + tarball_sha512_hex=tarball_sha512, + tarball_sha1=tarball_sha1, + attestation_sha256=_sha256_hex(attestation_bytes), + unpacked_bytes=unpacked_bytes, + archive_file_count=archive_file_count, + license_file_count=license_file_count, + osv_response_sha256=osv_sha256, + vulnerability_ids=vulnerability_ids, + ) + ) + return tuple(verified) + + +def build_cyclonedx_sbom( + policy: Policy, + verified: tuple[VerifiedPackage, ...], +) -> dict[str, Any]: + by_name = {item.policy.name: item for item in verified} + components: list[dict[str, Any]] = [] + dependencies: list[dict[str, Any]] = [] + for package in policy.packages: + evidence = by_name[package.name] + components.append( + { + "type": "library", + "bom-ref": package.bom_ref, + "group": package.name.split("/", 1)[0].removeprefix("@") + if package.name.startswith("@") + else "", + "name": package.name.split("/", 1)[-1], + "version": package.version, + "purl": package.bom_ref, + "hashes": [ + {"alg": "SHA-1", "content": evidence.tarball_sha1}, + {"alg": "SHA-512", "content": evidence.tarball_sha512_hex}, + ], + "licenses": [{"license": {"id": package.license}}], + "externalReferences": [ + {"type": "distribution", "url": package.tarball_url}, + {"type": "build-meta", "url": package.attestations_url}, + ], + } + ) + dependencies.append( + { + "ref": package.bom_ref, + "dependsOn": [ + by_name[name].policy.bom_ref + for name in sorted(package.dependencies) + ], + } + ) + serial = uuid.uuid5(uuid.NAMESPACE_URL, f"{policy.candidate_id}:{policy.checksum}") + return { + "bomFormat": "CycloneDX", + "specVersion": SBOM_SPEC_VERSION, + "serialNumber": f"urn:uuid:{serial}", + "version": 1, + "metadata": { + "component": { + "type": "application", + "bom-ref": f"awoooi:mcp-artifact:{policy.candidate_id}", + "name": policy.candidate_id, + "version": policy.packages[0].version, + }, + "tools": { + "components": [ + { + "type": "application", + "name": "awoooi-external-mcp-artifact-controller", + "version": "1", + } + ] + }, + }, + "components": components, + "dependencies": dependencies, + } + + +def _write_bundle( + directory: Path, + policy: Policy, + verified: tuple[VerifiedPackage, ...], + sbom: dict[str, Any], +) -> tuple[str, str]: + bundle = directory / "bundle" + tarballs = bundle / "tarballs" + attestations = bundle / "attestations" + tarballs.mkdir(parents=True) + attestations.mkdir(parents=True) + package_rows: list[dict[str, Any]] = [] + for item in verified: + package = item.policy + tarball_path = tarballs / package.filename + attestation_path = attestations / f"{package.filename}.attestations.json" + tarball_path.write_bytes(item.tarball) + attestation_path.write_bytes(item.attestation) + package_rows.append( + { + "name": package.name, + "version": package.version, + "tarball_path": f"tarballs/{package.filename}", + "tarball_sha512_hex": item.tarball_sha512_hex, + "tarball_sha1": item.tarball_sha1, + "metadata_sha256": item.metadata_sha256, + "attestation_path": ( + f"attestations/{package.filename}.attestations.json" + ), + "attestation_sha256": item.attestation_sha256, + "npm_registry_signature_verified": True, + "slsa_subject_verified": True, + "archive_safety_verified": True, + "license": package.license, + "license_file_count": item.license_file_count, + "osv_response_sha256": item.osv_response_sha256, + "vulnerability_ids": list(item.vulnerability_ids), + } + ) + sbom_bytes = _canonical_json_bytes(sbom) + b"\n" + sbom_path = bundle / "sbom.cdx.json" + sbom_path.write_bytes(sbom_bytes) + manifest = { + "schema_version": "awoooi_external_mcp_artifact_bundle_v1", + "candidate_id": policy.candidate_id, + "work_item_id": policy.work_item_id, + "policy_checksum": policy.checksum, + "package_count": len(package_rows), + "packages": package_rows, + "sbom": { + "path": "sbom.cdx.json", + "format": "CycloneDX", + "spec_version": SBOM_SPEC_VERSION, + "sha256": _sha256_hex(sbom_bytes), + }, + "operation_boundaries": { + "mcp_server_started": False, + "browser_started": False, + "external_tool_call_performed": False, + "external_rag_write_performed": False, + "production_write_performed": False, + "runtime_request_or_response_body_stored": False, + "upstream_supply_chain_attestation_bundled": True, + }, + } + manifest_bytes = _canonical_json_bytes(manifest) + b"\n" + (bundle / "bundle-manifest.json").write_bytes(manifest_bytes) + dockerfile = ( + "FROM scratch\n" + "COPY bundle/ /artifact/\n" + f'LABEL org.opencontainers.image.title="{policy.candidate_id}"\n' + f'LABEL org.opencontainers.image.version="{policy.packages[0].version}"\n' + f'LABEL io.awoooi.policy-checksum="{policy.checksum}"\n' + ) + (directory / "Dockerfile").write_text(dockerfile, encoding="utf-8") + return _sha256_hex(manifest_bytes), _sha256_hex(sbom_bytes) + + +def _run_command(command: list[str], *, cwd: Path, timeout: int) -> str: + try: + completed = subprocess.run( + command, + cwd=cwd, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ControllerError("bounded_execution_command_failed") from exc + if completed.returncode != 0: + raise ControllerError("bounded_execution_command_failed") + return completed.stdout + completed.stderr + + +def mirror_bundle( + policy: Policy, + verified: tuple[VerifiedPackage, ...], + sbom: dict[str, Any], +) -> tuple[str, str, str, str]: + docker = shutil.which("docker") + if not docker: + raise ControllerError("docker_cli_unavailable") + with tempfile.TemporaryDirectory(prefix="awoooi-playwright-mcp-bundle-") as temp: + context = Path(temp) + manifest_sha256, sbom_sha256 = _write_bundle( + context, + policy, + verified, + sbom, + ) + _run_command( + [ + docker, + "build", + "--network=none", + "--pull=false", + "--provenance=false", + "--sbom=false", + "--file", + str(context / "Dockerfile"), + "--tag", + policy.push_ref, + ".", + ], + cwd=context, + timeout=300, + ) + push_output = _run_command( + [docker, "push", policy.push_ref], + cwd=context, + timeout=300, + ) + matches = re.findall(r"digest:\s+(sha256:[0-9a-f]{64})", push_output) + if not matches: + inspected = _run_command( + [ + docker, + "image", + "inspect", + "--format", + "{{json .RepoDigests}}", + policy.push_ref, + ], + cwd=context, + timeout=30, + ) + try: + repo_digests = json.loads(inspected.strip()) + except json.JSONDecodeError as exc: + raise ControllerError("internal_mirror_digest_missing") from exc + matches = [ + value.rsplit("@", 1)[1] + for value in repo_digests + if isinstance(value, str) + and value.startswith(policy.push_repository + "@sha256:") + ] + digest = matches[-1] if matches else "" + if not SHA256_PATTERN.fullmatch(digest): + raise ControllerError("internal_mirror_digest_invalid") + digest_ref = f"{policy.push_repository}@{digest}" + _run_command( + [docker, "buildx", "imagetools", "inspect", digest_ref], + cwd=context, + timeout=60, + ) + runtime_ref = f"{policy.runtime_repository}@{digest}" + return digest_ref, runtime_ref, manifest_sha256, sbom_sha256 + + +def build_receipt( + *, + policy: Policy, + verified: tuple[VerifiedPackage, ...], + sbom_sha256: str, + bundle_manifest_sha256: str, + run_id: str, + trace_id: str, + mode: str, + internal_mirror_ref: str | None, + runtime_mirror_ref: str | None, +) -> dict[str, Any]: + observed_at = datetime.now(timezone.utc).isoformat() + apply_performed = internal_mirror_ref is not None + terminal = ( + "completed_internal_mirror_pending_independent_verifier" + if apply_performed + else "completed_check_mode_pending_independent_verifier" + ) + package_rows = [ + { + "name": item.policy.name, + "version": item.policy.version, + "metadata_sha256": item.metadata_sha256, + "tarball_sha512_hex": item.tarball_sha512_hex, + "tarball_sha1": item.tarball_sha1, + "attestation_sha256": item.attestation_sha256, + "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": list(item.vulnerability_ids), + "unpacked_bytes": item.unpacked_bytes, + "archive_file_count": item.archive_file_count, + } + for item in verified + ] + 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_level": policy.risk_level, + "mode": mode, + "status": terminal, + "observed_at": observed_at, + "policy_checksum": policy.checksum, + "package_count": len(package_rows), + "packages": package_rows, + "internal_mirror_ref": internal_mirror_ref, + "runtime_mirror_ref": runtime_mirror_ref, + "bundle_manifest_sha256": bundle_manifest_sha256, + "cyclonedx_sbom_sha256": sbom_sha256, + "promotion_allowed": False, + "replay_allowed": False, + "shadow_allowed": False, + "canary_allowed": False, + "stage_receipts": { + "sensor_source_receipt": { + "registry": "registry.npmjs.org", + "osv": "api.osv.dev", + "redirects_followed": 0, + "runtime_request_or_response_body_stored": False, + "upstream_supply_chain_attestation_bundled": True, + }, + "normalized_asset_identity": { + "candidate_id": policy.candidate_id, + "exact_root_version": policy.packages[0].version, + "package_count": len(package_rows), + }, + "source_of_truth_diff": { + "change_state": "baseline_created", + "policy_checksum": policy.checksum, + }, + "decision": { + "candidate_action": ( + "run_independent_mirror_verifier_then_implement_replay_adapter" + if apply_performed + else "run_independent_check_verifier_then_mirror_exact_bundle" + ), + "runtime_promotion_allowed": False, + }, + "risk_policy": { + "risk_level": "high", + "exact_versions": True, + "immutable_upstream_digests": True, + "npm_registry_signatures_verified": True, + "slsa_subjects_verified": True, + "license_decision": "passed", + "vulnerability_decision": "passed_no_known_osv_records", + }, + "check_mode": { + "status": "passed", + "mcp_process_started": False, + "browser_started": False, + "production_write_performed": False, + }, + "bounded_execution": { + "status": "internal_oci_bundle_pushed" if apply_performed else "not_executed", + "internal_registry_write_performed": apply_performed, + "external_runtime_mutation_performed": False, + "production_route_changed": False, + }, + "independent_post_verifier": { + "status": "pending_external_verifier", + "controller_registry_digest_check": apply_performed, + "internal_digest_ref_verified": False, + "bundle_manifest_sha256": bundle_manifest_sha256, + "cyclonedx_sbom_sha256": sbom_sha256, + "package_digest_count": len(package_rows), + }, + "rollback_or_no_write_terminal": { + "status": ( + "pending_external_verifier" + if apply_performed + else "no_write_pending_external_verifier" + ), + "rollback_action": "disable_candidate_and_retain_immutable_bundle", + "external_artifact_delete_allowed": False, + "production_route_change_allowed": False, + }, + "learning_writeback": { + "status": "receipt_ready_for_committed_writeback", + "rag_write_performed": False, + "km_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("--run-id", required=True) + parser.add_argument("--trace-id", required=True) + parser.add_argument("--receipt", type=Path) + parser.add_argument("command", choices=("check", "mirror")) + 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 ControllerError("run_id_invalid") from exc + if normalized_run_id != args.run_id: + raise ControllerError("run_id_not_canonical") + if not TRACE_PATTERN.fullmatch(args.trace_id): + raise ControllerError("trace_id_invalid") + if args.trace_id != f"mcp-artifact-{normalized_run_id}": + raise ControllerError("run_trace_identity_mismatch") + if args.command == "check" and args.apply: + raise ControllerError("check_mode_cannot_apply") + if args.command == "mirror" and args.apply and args.receipt is None: + raise ControllerError("apply_receipt_path_required") + policy = load_policy(args.policy) + http = BoundedHTTPClient(policy.allowed_hosts | frozenset({OSV_HOST})) + verified = verify_upstream_bundle(policy, http) + sbom = build_cyclonedx_sbom(policy, verified) + with tempfile.TemporaryDirectory(prefix="awoooi-playwright-mcp-check-") as temp: + manifest_sha256, sbom_sha256 = _write_bundle( + Path(temp), + policy, + verified, + sbom, + ) + internal_ref = None + runtime_ref = None + if args.command == "mirror" and args.apply: + internal_ref, runtime_ref, manifest_sha256, sbom_sha256 = mirror_bundle( + policy, + verified, + sbom, + ) + receipt = build_receipt( + policy=policy, + verified=verified, + sbom_sha256=sbom_sha256, + bundle_manifest_sha256=manifest_sha256, + run_id=normalized_run_id, + trace_id=args.trace_id, + mode=("apply" if args.command == "mirror" and args.apply else "check"), + internal_mirror_ref=internal_ref, + runtime_mirror_ref=runtime_ref, + ) + 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 ControllerError 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 diff --git a/scripts/security/tests/test_external_mcp_artifact_controller.py b/scripts/security/tests/test_external_mcp_artifact_controller.py new file mode 100644 index 000000000..cfce6178c --- /dev/null +++ b/scripts/security/tests/test_external_mcp_artifact_controller.py @@ -0,0 +1,393 @@ +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() diff --git a/scripts/security/tests/test_verify_external_mcp_artifact_receipt.py b/scripts/security/tests/test_verify_external_mcp_artifact_receipt.py new file mode 100644 index 000000000..d99a1ce95 --- /dev/null +++ b/scripts/security/tests/test_verify_external_mcp_artifact_receipt.py @@ -0,0 +1,471 @@ +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() diff --git a/scripts/security/verify_external_mcp_artifact_receipt.py b/scripts/security/verify_external_mcp_artifact_receipt.py new file mode 100644 index 000000000..3c24f61f0 --- /dev/null +++ b/scripts/security/verify_external_mcp_artifact_receipt.py @@ -0,0 +1,688 @@ +#!/usr/bin/env python3 +"""Independently verify an internal external-MCP artifact mirror receipt. + +This verifier does not import the artifact controller. It validates the policy +and controller receipt through a separate code path, reads the Harbor image back +by immutable digest, and inspects its layers as data. It never starts a container, +MCP server, browser, replay, or production canary. +""" + +from __future__ import annotations + +import argparse +import base64 +import binascii +import hashlib +import io +import json +import re +import shutil +import subprocess +import sys +import tarfile +import tempfile +import uuid +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import Any +from urllib.parse import quote + + +POLICY_SCHEMA_VERSION = "awoooi_external_mcp_artifact_policy_v1" +CONTROLLER_RECEIPT_SCHEMA_VERSION = "awoooi_external_mcp_artifact_receipt_v1" +VERIFIER_RECEIPT_SCHEMA_VERSION = "awoooi_external_mcp_artifact_verifier_receipt_v1" +SHA256_HEX_PATTERN = re.compile(r"^[0-9a-f]{64}$") +SHA1_HEX_PATTERN = re.compile(r"^[0-9a-f]{40}$") +SHA512_HEX_PATTERN = re.compile(r"^[0-9a-f]{128}$") +OCI_DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") +MAX_IMAGE_ARCHIVE_BYTES = 96 * 1024 * 1024 +MAX_LAYER_UNPACKED_BYTES = 64 * 1024 * 1024 +MAX_LAYER_FILES = 2_048 +SLSA_PREDICATE_TYPE = "https://slsa.dev/provenance/v1" +IN_TOTO_STATEMENT_TYPE = "https://in-toto.io/Statement/v1" + + +class VerifierError(RuntimeError): + """Fail-closed public-safe verifier error class.""" + + +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(path: Path, error_class: str) -> dict[str, Any]: + try: + data = path.read_bytes() + except OSError as exc: + raise VerifierError(error_class) from exc + if len(data) > 4 * 1024 * 1024: + raise VerifierError(error_class) + try: + payload = json.loads(data) + except json.JSONDecodeError as exc: + raise VerifierError(error_class) from exc + if not isinstance(payload, dict): + raise VerifierError(error_class) + return payload + + +def _require_dict(value: Any, error_class: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise VerifierError(error_class) + return value + + +def _expected_packages(policy: dict[str, Any]) -> dict[str, dict[str, Any]]: + rows = policy.get("packages") + if not isinstance(rows, list) or len(rows) != 3: + raise VerifierError("policy_package_set_invalid") + result: dict[str, dict[str, Any]] = {} + for row in rows: + if not isinstance(row, dict): + raise VerifierError("policy_package_set_invalid") + name = row.get("name") + version = row.get("version") + sha512_hex = row.get("slsa_subject_sha512_hex") + sha1_hex = row.get("shasum_sha1") + if ( + not isinstance(name, str) + or not isinstance(version, str) + or not isinstance(sha512_hex, str) + or not SHA512_HEX_PATTERN.fullmatch(sha512_hex) + or not isinstance(sha1_hex, str) + or not SHA1_HEX_PATTERN.fullmatch(sha1_hex) + or name in result + ): + raise VerifierError("policy_package_set_invalid") + result[name] = row + if set(result) != {"@playwright/mcp", "playwright", "playwright-core"}: + raise VerifierError("policy_package_set_invalid") + return result + + +def validate_controller_receipt( + policy: dict[str, Any], + receipt: dict[str, Any], +) -> dict[str, Any]: + """Validate the controller receipt without using controller code.""" + if policy.get("schema_version") != POLICY_SCHEMA_VERSION: + raise VerifierError("policy_schema_invalid") + if receipt.get("schema_version") != CONTROLLER_RECEIPT_SCHEMA_VERSION: + raise VerifierError("controller_receipt_schema_invalid") + policy_checksum = "sha256:" + _sha256_hex(_canonical_json_bytes(policy)) + if receipt.get("policy_checksum") != policy_checksum: + raise VerifierError("controller_receipt_policy_checksum_mismatch") + if ( + receipt.get("work_item_id") != policy.get("work_item_id") + or receipt.get("candidate_id") != policy.get("candidate_id") + or receipt.get("risk_level") != "high" + or receipt.get("mode") != "apply" + or receipt.get("status") + != "completed_internal_mirror_pending_independent_verifier" + ): + raise VerifierError("controller_receipt_identity_or_state_invalid") + + run_id = receipt.get("run_id") + trace_id = receipt.get("trace_id") + try: + normalized_run_id = str(uuid.UUID(run_id)) + except (ValueError, AttributeError) as exc: + raise VerifierError("controller_receipt_run_id_invalid") from exc + if run_id != normalized_run_id or trace_id != f"mcp-artifact-{run_id}": + raise VerifierError("controller_receipt_run_trace_mismatch") + if any( + receipt.get(field) is not False + for field in ( + "promotion_allowed", + "replay_allowed", + "shadow_allowed", + "canary_allowed", + ) + ): + raise VerifierError("controller_receipt_promotion_boundary_invalid") + + expected = _expected_packages(policy) + rows = receipt.get("packages") + if not isinstance(rows, list) or len(rows) != len(expected): + raise VerifierError("controller_receipt_package_set_invalid") + seen: set[str] = set() + for row in rows: + if not isinstance(row, dict) or row.get("name") not in expected: + raise VerifierError("controller_receipt_package_set_invalid") + package = expected[row["name"]] + if row["name"] in seen: + raise VerifierError("controller_receipt_package_set_invalid") + seen.add(row["name"]) + if ( + row.get("version") != package.get("version") + or row.get("tarball_sha512_hex") + != package.get("slsa_subject_sha512_hex") + or row.get("tarball_sha1") != package.get("shasum_sha1") + or row.get("npm_registry_signature_verified") is not True + or row.get("slsa_subject_verified") is not True + or row.get("archive_safety_verified") is not True + or row.get("license_decision") != "passed" + or row.get("vulnerability_decision") + != "passed_no_known_osv_records" + or row.get("vulnerability_ids") != [] + ): + raise VerifierError("controller_receipt_package_evidence_invalid") + if seen != set(expected): + raise VerifierError("controller_receipt_package_set_invalid") + + stages = _require_dict(receipt.get("stage_receipts"), "stage_receipts_invalid") + sensor = _require_dict( + stages.get("sensor_source_receipt"), "sensor_source_receipt_invalid" + ) + check_mode = _require_dict(stages.get("check_mode"), "check_mode_invalid") + execution = _require_dict( + stages.get("bounded_execution"), "bounded_execution_invalid" + ) + pending_verifier = _require_dict( + stages.get("independent_post_verifier"), + "independent_post_verifier_state_invalid", + ) + if ( + sensor.get("redirects_followed") != 0 + or sensor.get("runtime_request_or_response_body_stored") is not False + or sensor.get("upstream_supply_chain_attestation_bundled") is not True + or check_mode.get("status") != "passed" + or check_mode.get("mcp_process_started") is not False + or check_mode.get("browser_started") is not False + or check_mode.get("production_write_performed") is not False + or execution.get("status") != "internal_oci_bundle_pushed" + or execution.get("internal_registry_write_performed") is not True + or execution.get("external_runtime_mutation_performed") is not False + or execution.get("production_route_changed") is not False + or pending_verifier.get("status") != "pending_external_verifier" + or pending_verifier.get("internal_digest_ref_verified") is not False + ): + raise VerifierError("controller_receipt_stage_boundary_invalid") + + mirror = _require_dict(policy.get("internal_mirror"), "internal_mirror_invalid") + internal_ref = receipt.get("internal_mirror_ref") + runtime_ref = receipt.get("runtime_mirror_ref") + push_prefix = f"{mirror.get('push_repository')}@" + runtime_prefix = f"{mirror.get('runtime_repository')}@" + if ( + not isinstance(internal_ref, str) + or not internal_ref.startswith(push_prefix) + or not isinstance(runtime_ref, str) + or not runtime_ref.startswith(runtime_prefix) + ): + raise VerifierError("controller_receipt_internal_ref_invalid") + internal_digest = internal_ref.removeprefix(push_prefix) + runtime_digest = runtime_ref.removeprefix(runtime_prefix) + if ( + internal_digest != runtime_digest + or not OCI_DIGEST_PATTERN.fullmatch(internal_digest) + ): + raise VerifierError("controller_receipt_internal_digest_invalid") + for field in ("bundle_manifest_sha256", "cyclonedx_sbom_sha256"): + if not isinstance(receipt.get(field), str) or not SHA256_HEX_PATTERN.fullmatch( + receipt[field] + ): + raise VerifierError("controller_receipt_bundle_hash_invalid") + return { + "run_id": run_id, + "trace_id": trace_id, + "work_item_id": receipt["work_item_id"], + "candidate_id": receipt["candidate_id"], + "policy_checksum": policy_checksum, + "internal_ref": internal_ref, + "runtime_ref": runtime_ref, + "digest": internal_digest, + "expected_packages": expected, + } + + +def _run_command(command: list[str], *, cwd: Path, timeout: int) -> None: + try: + completed = subprocess.run( + command, + cwd=cwd, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise VerifierError("registry_readback_command_failed") from exc + if completed.returncode != 0: + raise VerifierError("registry_readback_command_failed") + + +def verify_bundled_npm_signature( + policy: dict[str, Any], + package: dict[str, Any], + tarball: bytes, +) -> None: + """Verify the npm ECDSA signature with OpenSSL through a separate path.""" + integrity = package.get("integrity") + signature_base64 = package.get("signature") + registry = _require_dict(policy.get("registry_contract"), "registry_contract_invalid") + signature_key = _require_dict( + registry.get("signature_key"), "registry_signature_key_invalid" + ) + public_key_base64 = signature_key.get("public_key_der_base64") + if ( + not isinstance(integrity, str) + or not integrity.startswith("sha512-") + or not isinstance(signature_base64, str) + or not isinstance(public_key_base64, str) + ): + raise VerifierError("bundled_npm_signature_policy_invalid") + try: + integrity_digest = base64.b64decode( + integrity.removeprefix("sha512-"), validate=True + ) + signature = base64.b64decode(signature_base64, validate=True) + public_key = base64.b64decode(public_key_base64, validate=True) + except (ValueError, binascii.Error) as exc: + raise VerifierError("bundled_npm_signature_encoding_invalid") from exc + if integrity_digest != hashlib.sha512(tarball).digest(): + raise VerifierError("bundled_npm_integrity_mismatch") + openssl = shutil.which("openssl") + if not openssl: + raise VerifierError("openssl_signature_verifier_unavailable") + message = f"{package.get('name')}@{package.get('version')}:{integrity}".encode() + with tempfile.TemporaryDirectory(prefix="awoooi-mcp-npm-verifier-") as temp: + root = Path(temp) + der_path = root / "public.der" + pem_path = root / "public.pem" + signature_path = root / "signature.der" + message_path = root / "message.bin" + der_path.write_bytes(public_key) + signature_path.write_bytes(signature) + message_path.write_bytes(message) + convert = subprocess.run( + [ + openssl, + "pkey", + "-pubin", + "-inform", + "DER", + "-in", + str(der_path), + "-out", + str(pem_path), + ], + check=False, + capture_output=True, + text=True, + timeout=10, + ) + if convert.returncode != 0: + raise VerifierError("bundled_npm_public_key_invalid") + verified = subprocess.run( + [ + openssl, + "dgst", + "-sha256", + "-verify", + str(pem_path), + "-signature", + str(signature_path), + str(message_path), + ], + check=False, + capture_output=True, + text=True, + timeout=10, + ) + if verified.returncode != 0: + raise VerifierError("bundled_npm_signature_verification_failed") + + +def verify_bundled_slsa_subject( + package: dict[str, Any], + attestation: bytes, +) -> None: + """Decode the bundled DSSE statement and independently bind its subject.""" + try: + payload = json.loads(attestation) + except json.JSONDecodeError as exc: + raise VerifierError("bundled_slsa_attestation_invalid") from exc + rows = payload.get("attestations") if isinstance(payload, dict) else None + if not isinstance(rows, list): + raise VerifierError("bundled_slsa_attestation_invalid") + matches = [ + row + for row in rows + if isinstance(row, dict) and row.get("predicateType") == SLSA_PREDICATE_TYPE + ] + if len(matches) != 1: + raise VerifierError("bundled_slsa_attestation_invalid") + bundle = matches[0].get("bundle") + envelope = bundle.get("dsseEnvelope") if isinstance(bundle, dict) else None + encoded = envelope.get("payload") if isinstance(envelope, dict) else None + if not isinstance(encoded, str): + raise VerifierError("bundled_slsa_dsse_payload_missing") + try: + statement = json.loads(base64.b64decode(encoded, validate=True)) + except (ValueError, binascii.Error, json.JSONDecodeError) as exc: + raise VerifierError("bundled_slsa_dsse_payload_invalid") from exc + expected_subject = { + "name": package.get("slsa_subject_name"), + "digest": {"sha512": package.get("slsa_subject_sha512_hex")}, + } + if ( + not isinstance(statement, dict) + or statement.get("_type") != IN_TOTO_STATEMENT_TYPE + or statement.get("predicateType") != SLSA_PREDICATE_TYPE + or statement.get("subject") != [expected_subject] + ): + raise VerifierError("bundled_slsa_subject_mismatch") + + +def read_image_files(internal_ref: str) -> dict[str, bytes]: + """Pull and inspect an OCI image as tar data without creating a container.""" + docker = shutil.which("docker") + if not docker: + raise VerifierError("docker_cli_unavailable") + with tempfile.TemporaryDirectory(prefix="awoooi-mcp-artifact-verifier-") as temp: + root = Path(temp) + archive_path = root / "image.tar" + _run_command( + [docker, "buildx", "imagetools", "inspect", internal_ref], + cwd=root, + timeout=60, + ) + _run_command([docker, "pull", internal_ref], cwd=root, timeout=180) + _run_command( + [docker, "image", "save", "--output", str(archive_path), internal_ref], + cwd=root, + timeout=120, + ) + return extract_image_files(archive_path) + + +def extract_image_files(archive_path: Path) -> dict[str, bytes]: + """Extract bounded artifact files from a Docker image archive.""" + try: + if archive_path.stat().st_size > MAX_IMAGE_ARCHIVE_BYTES: + raise VerifierError("image_archive_size_exceeded") + with tarfile.open(archive_path, mode="r:") as outer: + manifest_member = outer.getmember("manifest.json") + manifest_file = outer.extractfile(manifest_member) + if manifest_file is None: + raise VerifierError("image_manifest_missing") + manifests = json.loads(manifest_file.read(1_048_577)) + if not isinstance(manifests, list) or len(manifests) != 1: + raise VerifierError("image_manifest_invalid") + layers = manifests[0].get("Layers") + if not isinstance(layers, list) or not layers: + raise VerifierError("image_layers_missing") + files: dict[str, bytes] = {} + total_bytes = 0 + file_count = 0 + for layer_name in layers: + if not isinstance(layer_name, str): + raise VerifierError("image_layer_name_invalid") + layer_path = PurePosixPath(layer_name) + if layer_path.is_absolute() or ".." in layer_path.parts: + raise VerifierError("image_layer_name_invalid") + layer_member = outer.getmember(layer_name) + layer_file = outer.extractfile(layer_member) + if layer_file is None: + raise VerifierError("image_layer_missing") + layer_bytes = layer_file.read(MAX_IMAGE_ARCHIVE_BYTES + 1) + if len(layer_bytes) > MAX_IMAGE_ARCHIVE_BYTES: + raise VerifierError("image_layer_size_exceeded") + with tarfile.open(fileobj=io.BytesIO(layer_bytes)) as layer: + for member in layer.getmembers(): + path = PurePosixPath(member.name) + if ( + path.is_absolute() + or ".." in path.parts + or member.issym() + or member.islnk() + or member.isdev() + ): + raise VerifierError("image_layer_unsafe_entry") + if member.isdir(): + continue + if not member.isfile(): + raise VerifierError("image_layer_unsupported_entry") + if any(part.startswith(".wh.") for part in path.parts): + raise VerifierError("image_layer_whiteout_not_allowed") + normalized = str(path).lstrip("./") + if not normalized.startswith("artifact/"): + raise VerifierError("image_layer_file_outside_artifact") + extracted = layer.extractfile(member) + if extracted is None: + raise VerifierError("image_layer_file_unreadable") + data = extracted.read(member.size + 1) + if len(data) != member.size: + raise VerifierError("image_layer_file_size_mismatch") + total_bytes += len(data) + file_count += 1 + if ( + total_bytes > MAX_LAYER_UNPACKED_BYTES + or file_count > MAX_LAYER_FILES + ): + raise VerifierError("image_layer_bounds_exceeded") + files[normalized] = data + except (OSError, KeyError, tarfile.TarError, json.JSONDecodeError) as exc: + raise VerifierError("image_archive_invalid") from exc + return files + + +def verify_bundle_files( + files: dict[str, bytes], + *, + policy: dict[str, Any], + controller_receipt: dict[str, Any], +) -> dict[str, Any]: + """Recompute immutable bundle evidence from image layer files.""" + manifest_bytes = files.get("artifact/bundle-manifest.json") + sbom_bytes = files.get("artifact/sbom.cdx.json") + if manifest_bytes is None or sbom_bytes is None: + raise VerifierError("bundle_core_files_missing") + if _sha256_hex(manifest_bytes) != controller_receipt.get( + "bundle_manifest_sha256" + ) or _sha256_hex(sbom_bytes) != controller_receipt.get("cyclonedx_sbom_sha256"): + raise VerifierError("bundle_core_hash_mismatch") + try: + manifest = json.loads(manifest_bytes) + sbom = json.loads(sbom_bytes) + except json.JSONDecodeError as exc: + raise VerifierError("bundle_core_json_invalid") from exc + if not isinstance(manifest, dict) or not isinstance(sbom, dict): + raise VerifierError("bundle_core_json_invalid") + policy_checksum = "sha256:" + _sha256_hex(_canonical_json_bytes(policy)) + if ( + manifest.get("candidate_id") != policy.get("candidate_id") + or manifest.get("work_item_id") != policy.get("work_item_id") + or manifest.get("policy_checksum") != policy_checksum + or manifest.get("package_count") != 3 + or manifest.get("operation_boundaries", {}).get("mcp_server_started") + is not False + or manifest.get("operation_boundaries", {}).get("browser_started") + is not False + or manifest.get("operation_boundaries", {}).get( + "runtime_request_or_response_body_stored" + ) + is not False + ): + raise VerifierError("bundle_manifest_boundary_invalid") + expected = _expected_packages(policy) + package_rows = manifest.get("packages") + if not isinstance(package_rows, list) or len(package_rows) != len(expected): + raise VerifierError("bundle_package_set_invalid") + verified_package_names: list[str] = [] + expected_file_paths = { + "artifact/bundle-manifest.json", + "artifact/sbom.cdx.json", + } + for row in package_rows: + if not isinstance(row, dict) or row.get("name") not in expected: + raise VerifierError("bundle_package_set_invalid") + package = expected[row["name"]] + tarball_path = row.get("tarball_path") + attestation_path = row.get("attestation_path") + if not isinstance(tarball_path, str) or not isinstance(attestation_path, str): + raise VerifierError("bundle_package_path_invalid") + tarball = files.get(f"artifact/{tarball_path}") + attestation = files.get(f"artifact/{attestation_path}") + if tarball is None or attestation is None: + raise VerifierError("bundle_package_file_missing") + expected_file_paths.update( + {f"artifact/{tarball_path}", f"artifact/{attestation_path}"} + ) + if ( + hashlib.sha512(tarball).hexdigest() + != package.get("slsa_subject_sha512_hex") + or hashlib.sha1(tarball, usedforsecurity=False).hexdigest() + != package.get("shasum_sha1") + or _sha256_hex(attestation) != row.get("attestation_sha256") + or row.get("npm_registry_signature_verified") is not True + or row.get("slsa_subject_verified") is not True + or row.get("archive_safety_verified") is not True + or row.get("vulnerability_ids") != [] + ): + raise VerifierError("bundle_package_evidence_mismatch") + verify_bundled_npm_signature(policy, package, tarball) + verify_bundled_slsa_subject(package, attestation) + verified_package_names.append(row["name"]) + if len(set(verified_package_names)) != len(expected): + raise VerifierError("bundle_package_set_invalid") + if set(files) != expected_file_paths: + raise VerifierError("bundle_file_set_invalid") + + expected_refs = { + f"pkg:npm/{quote(name, safe='/')}@{row['version']}" + for name, row in expected.items() + } + components = sbom.get("components") + if ( + sbom.get("bomFormat") != "CycloneDX" + or sbom.get("specVersion") != "1.5" + or not isinstance(components, list) + or {row.get("bom-ref") for row in components if isinstance(row, dict)} + != expected_refs + ): + raise VerifierError("bundle_sbom_invalid") + manifest_sbom = _require_dict(manifest.get("sbom"), "bundle_sbom_invalid") + if manifest_sbom.get("sha256") != _sha256_hex(sbom_bytes): + raise VerifierError("bundle_sbom_hash_mismatch") + return { + "artifact_file_count": len(files), + "verified_package_count": len(verified_package_names), + "bundle_manifest_sha256": _sha256_hex(manifest_bytes), + "cyclonedx_sbom_sha256": _sha256_hex(sbom_bytes), + } + + +def build_verifier_receipt( + *, + identity: dict[str, Any], + controller_receipt_bytes: bytes, + bundle_evidence: dict[str, Any], +) -> dict[str, Any]: + return { + "schema_version": VERIFIER_RECEIPT_SCHEMA_VERSION, + "run_id": identity["run_id"], + "trace_id": identity["trace_id"], + "work_item_id": identity["work_item_id"], + "candidate_id": identity["candidate_id"], + "status": "completed_internal_mirror_verified", + "observed_at": datetime.now(timezone.utc).isoformat(), + "policy_checksum": identity["policy_checksum"], + "controller_receipt_sha256": _sha256_hex(controller_receipt_bytes), + "internal_mirror_ref": identity["internal_ref"], + "runtime_mirror_ref": identity["runtime_ref"], + "internal_digest": identity["digest"], + "promotion_allowed": False, + "replay_allowed": False, + "shadow_allowed": False, + "canary_allowed": False, + "independent_post_verifier": { + "status": "passed", + "implementation": "standalone_no_controller_import", + "registry_digest_readback_verified": True, + "npm_registry_signatures_reverified": True, + "slsa_subjects_reverified": True, + "container_or_mcp_started": False, + **bundle_evidence, + }, + "rollback_terminal": { + "status": "rollback_ready_not_promoted", + "action": "disable_candidate_and_retain_immutable_bundle", + "production_route_changed": False, + "external_artifact_delete_allowed": False, + }, + "learning_writeback": { + "status": "verifier_receipt_ready_for_committed_writeback", + "external_rag_write_performed": False, + "raw_runtime_request_or_response_body_stored": 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", required=True, type=Path) + parser.add_argument("--controller-receipt", required=True, type=Path) + parser.add_argument("--verifier-receipt", required=True, type=Path) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + policy = _read_json(args.policy, "policy_unreadable") + controller_receipt_bytes = args.controller_receipt.read_bytes() + if len(controller_receipt_bytes) > 4 * 1024 * 1024: + raise VerifierError("controller_receipt_too_large") + controller_receipt = _read_json( + args.controller_receipt, + "controller_receipt_unreadable", + ) + identity = validate_controller_receipt(policy, controller_receipt) + files = read_image_files(identity["internal_ref"]) + bundle_evidence = verify_bundle_files( + files, + policy=policy, + controller_receipt=controller_receipt, + ) + receipt = build_verifier_receipt( + identity=identity, + controller_receipt_bytes=controller_receipt_bytes, + bundle_evidence=bundle_evidence, + ) + _write_receipt(args.verifier_receipt, receipt) + print(json.dumps(receipt, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except (OSError, VerifierError) as exc: + error_class = str(exc) if isinstance(exc, VerifierError) else "verifier_io_error" + print( + json.dumps( + { + "schema_version": VERIFIER_RECEIPT_SCHEMA_VERSION, + "status": "blocked_with_safe_next_action", + "error_class": error_class, + }, + sort_keys=True, + ), + file=sys.stderr, + ) + raise SystemExit(1) from None