689 lines
28 KiB
Python
689 lines
28 KiB
Python
#!/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
|