1331 lines
49 KiB
Python
1331 lines
49 KiB
Python
#!/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
|