Files
ai-contract-platform/yuqei-infra/scripts/build-installable-release.py
T

1075 lines
48 KiB
Python

#!/usr/bin/env python3
"""Validate a Linux build bundle and emit its canonical installable manifest.
The assembler copies nothing and trusts no receipt claim without re-reading the
referenced bytes. It refuses missing/extra roles, duplicate identities or
paths, digest drift, traversal, symlinks, stale scanner metadata, and any
non-passing vulnerability or unresolved license report.
"""
from __future__ import annotations
import argparse
import errno
import hashlib
import json
import os
import re
import stat
import sys
from pathlib import Path
from typing import Any
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
from linux_release_common import ( # noqa: E402
COMMAND_TIMEOUTS_SECONDS,
COMPOSE_GATE_POLICY,
CONTROLLED_SUBPROCESS_ENV,
INSTALLABLE_MANIFEST_SCHEMA,
LOCAL_DOCKER_HOST,
PINNED_EXECUTABLES,
REQUIRED_IMAGE_SPECS,
REQUIRED_ROLES,
REQUIRED_TOOL_VERSIONS,
TARGET,
TRIVY_DB_REPOSITORY_TRANSPORT,
ReleaseValidationError,
canonical_json_bytes,
ensure_regular_file_under,
file_record,
inspect_docker_archive,
inspect_oci_archive,
load_json_file,
load_policy,
load_source_manifest,
parse_rfc3339,
validate_receipt_payload_digest,
validate_compose_gate_report,
validate_candidate_checkout,
validate_manifest_commit_tree,
validate_relative_path,
validate_spdx,
validate_tag_binding,
validate_trivy_db_metadata,
validate_trivy_license_scan,
validate_trivy_scan,
verify_file_record,
verify_source_archive,
synthetic_compose_gate_env_bytes,
with_payload_digest,
)
RELEASE_CONTROL_PATHS = {
"runner": "yuqei-infra/scripts/run-linux-release-build.py",
"common": "yuqei-infra/scripts/linux_release_common.py",
"policy": "yuqei-infra/release/linux-release-security-policy.v1.json",
}
def _mapping(value: Any, label: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise ReleaseValidationError(f"{label} must be a JSON object")
return value
def _list(value: Any, label: str) -> list[Any]:
if not isinstance(value, list):
raise ReleaseValidationError(f"{label} must be a JSON array")
return value
def _record_path(record: Any, label: str) -> str:
value = _mapping(record, label).get("path")
if not isinstance(value, str):
raise ReleaseValidationError(f"{label}.path must be a string")
return value
def _compare_summary(record: dict[str, Any], summary: dict[str, Any], *, label: str) -> None:
drift = {
key: {"receipt": record.get(key), "actual": value}
for key, value in summary.items()
if record.get(key) != value
}
if drift:
raise ReleaseValidationError(f"{label} parsed metadata drift: {json.dumps(drift, sort_keys=True)}")
def _path_relative_to_root(root: Path, path: Path, *, label: str) -> str:
try:
relative = path.resolve(strict=True).relative_to(root.resolve(strict=True))
except (OSError, ValueError) as error:
raise ReleaseValidationError(f"{label} must be within the artifact root") from error
result = relative.as_posix()
ensure_regular_file_under(root, result)
return result
def _reject_unlisted_entries(root: Path, expected_files: set[str], output: Path) -> None:
output_relative = None
try:
output_relative = output.resolve(strict=False).relative_to(root.resolve(strict=True)).as_posix()
except (OSError, ValueError):
pass
actual_files: set[str] = set()
for path in root.rglob("*"):
relative = path.relative_to(root).as_posix()
if path.is_symlink():
raise ReleaseValidationError(f"installable artifact tree contains a symlink: {relative}")
if path.is_file():
actual_files.add(relative)
elif not path.is_dir():
raise ReleaseValidationError(f"installable artifact tree contains a special entry: {relative}")
if output_relative:
actual_files.discard(output_relative)
if actual_files != expected_files:
raise ReleaseValidationError(
"artifact tree differs from the receipt: "
f"missing={sorted(expected_files - actual_files)}, unlisted={sorted(actual_files - expected_files)}"
)
def _write_exclusive_manifest(output: Path, payload: dict[str, Any]) -> None:
"""Durably create an owner-only manifest without following or replacing paths."""
parent = output.parent
if parent.is_symlink() or not parent.is_dir():
raise ReleaseValidationError(
"installable manifest parent must be a real directory"
)
if os.name != "nt":
parent_metadata = parent.stat()
if (
parent_metadata.st_uid != os.geteuid()
or stat.S_IMODE(parent_metadata.st_mode) & 0o022
):
raise ReleaseValidationError(
"installable manifest parent must be owned by the executor and not group/world writable"
)
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
flags |= getattr(os, "O_NOFOLLOW", 0)
descriptor = -1
try:
descriptor = os.open(output, flags, 0o600)
opened = os.fstat(descriptor)
if not stat.S_ISREG(opened.st_mode) or opened.st_nlink != 1:
raise ReleaseValidationError(
"installable manifest output must be a single-link regular file"
)
data = canonical_json_bytes(payload)
view = memoryview(data)
while view:
written = os.write(descriptor, view)
if written <= 0:
raise OSError("short write while creating installable manifest")
view = view[written:]
if os.name != "nt":
os.fchmod(descriptor, 0o600)
os.fsync(descriptor)
final = os.fstat(descriptor)
if not stat.S_ISREG(final.st_mode) or final.st_nlink != 1:
raise ReleaseValidationError(
"installable manifest output changed identity while being written"
)
try:
path_identity = os.lstat(output)
except OSError as error:
raise ReleaseValidationError(
"installable manifest output path disappeared while being written"
) from error
if (
not stat.S_ISREG(path_identity.st_mode)
or path_identity.st_nlink != 1
or path_identity.st_dev != final.st_dev
or path_identity.st_ino != final.st_ino
):
raise ReleaseValidationError(
"installable manifest output path changed identity while being written"
)
except FileExistsError as error:
raise ReleaseValidationError(
"installable manifest output must not already exist"
) from error
except OSError as error:
if error.errno == errno.ELOOP:
raise ReleaseValidationError(
"installable manifest output must not be a symbolic link"
) from error
raise
finally:
if descriptor >= 0:
os.close(descriptor)
if os.name != "nt":
directory_descriptor = os.open(
output.parent,
os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
)
try:
os.fsync(directory_descriptor)
finally:
os.close(directory_descriptor)
def assemble_installable_release(
*,
artifact_root: Path,
receipt_path: Path,
expected_receipt_sha256: str,
repo: Path,
output: Path,
write_output: bool = True,
) -> dict[str, Any]:
if artifact_root.is_symlink() or not artifact_root.is_dir():
raise ReleaseValidationError("artifact root must be a real directory")
receipt_relative = _path_relative_to_root(artifact_root, receipt_path, label="build receipt")
receipt_bytes = receipt_path.read_bytes()
if not re.fullmatch(r"[0-9a-f]{64}", expected_receipt_sha256):
raise ReleaseValidationError("expected build receipt raw SHA-256 is invalid")
if hashlib.sha256(receipt_bytes).hexdigest() != expected_receipt_sha256:
raise ReleaseValidationError(
"build receipt does not match the independently supplied raw SHA-256"
)
receipt = validate_receipt_payload_digest(load_json_file(receipt_path, label="build receipt"))
if canonical_json_bytes(receipt) != receipt_bytes:
raise ReleaseValidationError("build receipt is not serialized as canonical JSON")
policy_record = _mapping(receipt.get("policy"), "receipt policy")
policy_path = verify_file_record(artifact_root, policy_record, label="receipt policy")
policy = load_policy(policy_path)
if policy_record.get("schema_version") != policy["schema_version"] or policy_record.get("target") != TARGET:
raise ReleaseValidationError("receipt policy metadata drift")
candidate = _mapping(receipt.get("candidate"), "receipt candidate")
required_candidate = {
"commit",
"tag",
"tag_object_oid",
"tree",
"commit_tree",
"commit_timestamp",
"source_manifest_payload_sha256",
"source_manifest",
"source_archive",
}
if set(candidate) != required_candidate:
raise ReleaseValidationError("receipt candidate fields are missing or unknown")
source_manifest_path = verify_file_record(
artifact_root, candidate["source_manifest"], label="source manifest"
)
source_manifest = load_source_manifest(source_manifest_path, commit=candidate["commit"])
if source_manifest["manifest_payload_sha256"] != candidate["source_manifest_payload_sha256"]:
raise ReleaseValidationError("source manifest payload digest drift in build receipt")
if source_manifest["candidate"]["tree"] != candidate["tree"]:
raise ReleaseValidationError("source manifest tree drift in build receipt")
if source_manifest["candidate"]["commit_timestamp"] != candidate["commit_timestamp"]:
raise ReleaseValidationError("source manifest timestamp drift in build receipt")
source_archive_path = verify_file_record(
artifact_root, candidate["source_archive"], label="source archive"
)
archive_summary = verify_source_archive(source_archive_path, source_manifest)
for key in ("root_prefix", "file_count"):
if candidate["source_archive"].get(key) != archive_summary[key]:
raise ReleaseValidationError(f"source archive {key} drift in build receipt")
commit_tree = validate_manifest_commit_tree(
repo, commit=candidate["commit"], manifest=source_manifest
)
if candidate["commit_tree"] != commit_tree:
raise ReleaseValidationError("candidate commit-tree evidence drift in build receipt")
tag_binding = validate_tag_binding(
repo,
tag=candidate["tag"],
commit=candidate["commit"],
expected_tree=candidate["tree"],
)
if candidate["tag_object_oid"] != tag_binding["tag_object_oid"]:
raise ReleaseValidationError("annotated tag object drift in build receipt")
builder = _mapping(receipt.get("builder"), "receipt builder")
required_builder_fields = {
"os",
"architecture",
"target_platform",
"tools",
"executables",
"controlled_environment",
"docker_host",
"command_timeouts_seconds",
"buildx_builder",
"base_image_transports",
"candidate_checkout",
"release_controls",
"skopeo_used",
}
if set(builder) != required_builder_fields:
raise ReleaseValidationError("receipt builder fields are missing or unknown")
if builder.get("os") != "linux" or builder.get("architecture") != "amd64":
raise ReleaseValidationError("build receipt was not produced by a Linux amd64 builder")
if builder.get("target_platform") != TARGET["platform"]:
raise ReleaseValidationError("build receipt target platform drift")
if builder.get("tools") != REQUIRED_TOOL_VERSIONS:
raise ReleaseValidationError("build receipt tool versions are not the pinned toolchain")
if builder.get("command_timeouts_seconds") != COMMAND_TIMEOUTS_SECONDS:
raise ReleaseValidationError("build receipt external-command timeout contract drift")
if builder.get("controlled_environment") != CONTROLLED_SUBPROCESS_ENV:
raise ReleaseValidationError("build receipt controlled subprocess environment drift")
if builder.get("docker_host") != LOCAL_DOCKER_HOST:
raise ReleaseValidationError("build receipt Docker daemon endpoint drift")
candidate_checkout = _mapping(
builder.get("candidate_checkout"), "receipt candidate checkout"
)
if set(candidate_checkout) != {
"head",
"worktree_clean",
"replace_refs_present",
"status_sha256",
}:
raise ReleaseValidationError(
"receipt candidate checkout fields are missing or unknown"
)
actual_checkout = validate_candidate_checkout(repo, commit=candidate["commit"])
if candidate_checkout != actual_checkout:
raise ReleaseValidationError("receipt candidate checkout evidence drift")
source_records = {
item["path"]: item for item in source_manifest["files"]["items"]
}
release_controls = _mapping(
builder.get("release_controls"), "receipt release controls"
)
if set(release_controls) != set(RELEASE_CONTROL_PATHS):
raise ReleaseValidationError("receipt release control inventory drift")
for name, relative in RELEASE_CONTROL_PATHS.items():
record = _mapping(release_controls[name], f"receipt release control {name}")
if set(record) != {"path", "size_bytes", "sha256", "git_blob_oid"}:
raise ReleaseValidationError(
f"receipt release control fields are missing or unknown for {name}"
)
source_record = source_records.get(relative)
expected_record = (
{
"path": relative,
"size_bytes": source_record["size"],
"sha256": source_record["sha256"],
"git_blob_oid": source_record["git_blob_oid"],
}
if isinstance(source_record, dict)
else None
)
if record != expected_record:
raise ReleaseValidationError(
f"receipt release control source binding drift for {name}"
)
executable_records = _mapping(builder.get("executables"), "receipt executables")
if set(executable_records) != set(PINNED_EXECUTABLES):
raise ReleaseValidationError("build receipt pinned executable inventory drift")
for name, expected_path in PINNED_EXECUTABLES.items():
executable = _mapping(executable_records[name], f"receipt executable {name}")
if (
executable.get("path") != expected_path
or executable.get("owner_uid") != 0
or executable.get("mode") != "0755"
or not isinstance(executable.get("size_bytes"), int)
or isinstance(executable.get("size_bytes"), bool)
or executable["size_bytes"] <= 0
or not isinstance(executable.get("sha256"), str)
or not re.fullmatch(r"[0-9a-f]{64}", executable["sha256"])
):
raise ReleaseValidationError(f"build receipt executable evidence drift for {name}")
buildx_builder = _mapping(builder.get("buildx_builder"), "receipt Buildx builder")
if (
buildx_builder.get("driver") not in policy["builder"]["allowed_drivers"]
or buildx_builder.get("node_status") != "running"
or buildx_builder.get("exporters") != policy["builder"]["required_exporters"]
or len(_list(buildx_builder.get("platforms"), "Buildx builder platforms")) == 0
or policy["builder"]["required_platform"] not in buildx_builder["platforms"]
):
raise ReleaseValidationError("build receipt does not prove the required Buildx builder capabilities")
if buildx_builder.get("buildkit_version") != policy["builder"]["required_buildkit_version"]:
raise ReleaseValidationError("build receipt BuildKit version drift")
for receipt_key, policy_key in (
("driver_image", "required_driver_image"),
("driver_image_digest", "required_driver_image_digest"),
("local_driver_image_id", "required_local_driver_image_id"),
):
if buildx_builder.get(receipt_key) != policy["builder"][policy_key]:
raise ReleaseValidationError(f"build receipt {receipt_key} drift")
if buildx_builder.get("image_transport") != "domestic_registry_mirror":
raise ReleaseValidationError("BuildKit driver image transport is not the pinned domestic mirror")
config_root = _mapping(buildx_builder.get("config_root"), "receipt Buildx config root")
if (
config_root.get("path") != CONTROLLED_SUBPROCESS_ENV["BUILDX_CONFIG"]
or config_root.get("selected_builder_instance")
!= f"instances/{buildx_builder.get('name')}"
or config_root.get("root_owned") is not True
or config_root.get("group_or_other_writable") is not False
or config_root.get("symlinks_allowed") is not False
or not isinstance(config_root.get("entry_count"), int)
or isinstance(config_root.get("entry_count"), bool)
or config_root["entry_count"] < 1
or not isinstance(config_root.get("metadata_sha256"), str)
or not re.fullmatch(r"[0-9a-f]{64}", config_root["metadata_sha256"])
):
raise ReleaseValidationError("build receipt fixed Buildx config evidence drift")
if not isinstance(buildx_builder.get("inspect_sha256"), str) or not re.fullmatch(
r"[0-9a-f]{64}", buildx_builder["inspect_sha256"]
):
raise ReleaseValidationError("build receipt has no Buildx inspection digest")
if not isinstance(buildx_builder.get("container_inspect_sha256"), str) or not re.fullmatch(
r"[0-9a-f]{64}", buildx_builder["container_inspect_sha256"]
):
raise ReleaseValidationError("build receipt has no BuildKit container inspection digest")
transport_records = _list(
builder.get("base_image_transports"), "receipt base-image transports"
)
expected_transports = {
image["base_image"]["transport_reference"]: image["base_image"]["content_digest"]
for image in policy["images"]
}
if len(transport_records) != len(expected_transports):
raise ReleaseValidationError("base-image transport evidence is missing or duplicated")
observed_transports: set[str] = set()
for record in transport_records:
item = _mapping(record, "base-image transport record")
reference = item.get("transport_reference")
if reference in observed_transports or expected_transports.get(reference) != item.get("content_digest"):
raise ReleaseValidationError("base-image transport reference or digest drift")
observed_transports.add(reference)
if (
item.get("transport") != "domestic_registry_mirror"
or item.get("media_type") != "application/vnd.oci.image.manifest.v1+json"
or not isinstance(item.get("inspect_sha256"), str)
or not re.fullmatch(r"[0-9a-f]{64}", item["inspect_sha256"])
):
raise ReleaseValidationError("base-image mirror inspection evidence is invalid")
if observed_transports != set(expected_transports):
raise ReleaseValidationError("base-image transport evidence does not cover the policy")
if builder.get("skopeo_used") is not False:
raise ReleaseValidationError("Skopeo is outside the supported release pipeline")
build = _mapping(receipt.get("build"), "receipt build")
if build.get("parallelism") != 1 or build.get("image_order") != list(REQUIRED_ROLES):
raise ReleaseValidationError("four images were not recorded in the required serial order")
started = parse_rfc3339(build.get("started_at"), label="build.started_at")
completed = parse_rfc3339(build.get("completed_at"), label="build.completed_at")
if completed < started:
raise ReleaseValidationError("build completion precedes build start")
if build.get("source_date_epoch") != int(
parse_rfc3339(candidate["commit_timestamp"], label="candidate.commit_timestamp").timestamp()
):
raise ReleaseValidationError("SOURCE_DATE_EPOCH is not bound to the candidate commit timestamp")
compose_gate = _mapping(build.get("compose_gate"), "receipt production Compose gate")
if (
set(compose_gate)
!= {
"synthetic_environment",
"production_secrets_used",
"execution_engine",
"validator",
"environment",
"report",
}
or compose_gate.get("synthetic_environment") is not True
or compose_gate.get("production_secrets_used") is not False
or compose_gate.get("execution_engine")
!= "self_bound_release_runner_builtin_v1"
):
raise ReleaseValidationError(
"production Compose gate fields or self-bound execution engine drifted"
)
compose_validator = _mapping(
compose_gate.get("validator"), "receipt production Compose gate validator"
)
expected_validator = source_records.get(COMPOSE_GATE_POLICY["validator"])
if (
expected_validator is None
or compose_validator.get("source_path") != COMPOSE_GATE_POLICY["validator"]
or compose_validator.get("sha256") != expected_validator["sha256"]
or compose_validator.get("size_bytes") != expected_validator["size"]
):
raise ReleaseValidationError("production Compose gate validator source binding drift")
compose_env_record = _mapping(
compose_gate.get("environment"), "receipt synthetic Compose gate environment"
)
compose_env_path = verify_file_record(
artifact_root,
compose_env_record,
label="synthetic Compose gate environment",
)
expected_compose_env = synthetic_compose_gate_env_bytes()
if compose_env_path.read_bytes() != expected_compose_env:
raise ReleaseValidationError("synthetic Compose gate environment content drift")
compose_report_record = _mapping(
compose_gate.get("report"), "receipt production Compose gate report"
)
compose_report_path = verify_file_record(
artifact_root,
compose_report_record,
label="production Compose gate report",
)
compose_report_bytes = compose_report_path.read_bytes()
compose_report_payload = load_json_file(
compose_report_path, label="production Compose gate report"
)
if canonical_json_bytes(compose_report_payload) != compose_report_bytes:
raise ReleaseValidationError("production Compose gate report is not canonical JSON")
compose_report_summary = validate_compose_gate_report(
compose_report_payload,
policy,
expected_env_sha256=compose_env_record["sha256"],
source_file_sha256={path: item["sha256"] for path, item in source_records.items()},
)
_compare_summary(
compose_report_record,
compose_report_summary,
label="production Compose gate report",
)
security = _mapping(receipt.get("security"), "receipt security")
if security.get("passed") is not True or security.get("fail_closed") is not True:
raise ReleaseValidationError("build receipt security gate did not pass fail closed")
if security.get("thresholds") != policy["vulnerability_scan"]["thresholds"]:
raise ReleaseValidationError("receipt vulnerability threshold drift")
if security.get("exemptions") != policy["vulnerability_scan"]["exemptions"]:
raise ReleaseValidationError("receipt vulnerability waiver policy drift")
claimed_vulnerability_counts = _mapping(
security.get("vulnerability_counts"), "receipt vulnerability counts"
)
claimed_waiver_ids = _list(security.get("waiver_ids"), "receipt waiver IDs")
claimed_license_inventory = _mapping(
security.get("license_inventory"), "receipt license inventory"
)
database_record = _mapping(security.get("database"), "receipt Trivy database")
database_path = verify_file_record(
artifact_root, database_record, label="Trivy database metadata"
)
validated_at = parse_rfc3339(database_record.get("validated_at"), label="database.validated_at")
database_summary = validate_trivy_db_metadata(
load_json_file(database_path, label="Trivy database metadata"),
policy,
at=validated_at,
)
_compare_summary(database_record, database_summary, label="Trivy database")
repository_transport = _mapping(
database_record.get("repository_transport"),
"receipt Trivy database repository transport",
)
if (
repository_transport.get("reference")
!= TRIVY_DB_REPOSITORY_TRANSPORT["reference"]
or repository_transport.get("transport")
!= TRIVY_DB_REPOSITORY_TRANSPORT["transport"]
or repository_transport.get("schema_version")
!= TRIVY_DB_REPOSITORY_TRANSPORT["schema_version"]
or repository_transport.get("media_type")
not in TRIVY_DB_REPOSITORY_TRANSPORT["allowed_media_types"]
or repository_transport.get("stable_during_download") is not True
):
raise ReleaseValidationError("Trivy database repository transport evidence drift")
resolved_digest = repository_transport.get("resolved_digest")
if not isinstance(resolved_digest, str) or not re.fullmatch(
r"sha256:[0-9a-f]{64}", resolved_digest
):
raise ReleaseValidationError("Trivy database repository resolved digest is invalid")
for field in (
"before_download_inspect_sha256",
"after_download_inspect_sha256",
):
value = repository_transport.get(field)
if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
raise ReleaseValidationError(
f"Trivy database repository {field} evidence is invalid"
)
if not (started <= validated_at <= completed):
raise ReleaseValidationError("Trivy database was not validated during the recorded build")
images = _list(receipt.get("images"), "receipt images")
roles = [image.get("role") if isinstance(image, dict) else None for image in images]
if len(roles) != len(set(roles)):
raise ReleaseValidationError("duplicate image role in build receipt")
if roles != list(REQUIRED_ROLES):
missing = sorted(set(REQUIRED_ROLES) - set(roles))
extra = sorted(str(role) for role in set(roles) - set(REQUIRED_ROLES))
raise ReleaseValidationError(f"image roles are incomplete or unordered: missing={missing}, extra={extra}")
tracked_paths: dict[str, str] = {receipt_relative.casefold(): receipt_relative}
expected_files: set[str] = {receipt_relative}
manifest_images: list[dict[str, Any]] = []
oci_manifest_digests: set[str] = set()
aggregate_vulnerability_counts = {
bucket: {severity: 0 for severity in policy["vulnerability_scan"]["thresholds"]}
for bucket in ("total", "exempted", "blocking")
}
applied_waiver_ids: set[str] = set()
aggregate_license_inventory = {
"scanner": "trivy",
"scanner_version": REQUIRED_TOOL_VERSIONS["trivy"],
"report_count": 0,
"result_count": 0,
"license_count": 0,
"unresolved_count": 0,
"passed": True,
}
def track(record: dict[str, Any], label: str) -> None:
path = _record_path(record, label)
validate_relative_path(path, label=f"{label}.path")
folded = path.casefold()
if folded in tracked_paths:
raise ReleaseValidationError(
f"duplicate or case-colliding artifact path: {path} and {tracked_paths[folded]}"
)
tracked_paths[folded] = path
expected_files.add(path)
track(policy_record, "policy")
track(_mapping(candidate["source_manifest"], "source manifest"), "source manifest")
track(_mapping(candidate["source_archive"], "source archive"), "source archive")
track(database_record, "Trivy database metadata")
track(compose_env_record, "synthetic Compose gate environment")
track(compose_report_record, "production Compose gate report")
if compose_env_record["path"] != "validation/synthetic-compose-gate.env":
raise ReleaseValidationError("synthetic Compose gate environment path drift")
if compose_report_record["path"] != "validation/production-compose-gate.json":
raise ReleaseValidationError("production Compose gate report path drift")
for index, (raw_image, expected) in enumerate(zip(images, REQUIRED_IMAGE_SPECS, strict=True), 1):
image = _mapping(raw_image, f"receipt image {index}")
if image.get("sequence") != index:
raise ReleaseValidationError(f"image sequence drift for {expected['role']}")
if image.get("role") != expected["role"] or image.get("image_name") != expected["name"]:
raise ReleaseValidationError(f"image identity drift for sequence {index}")
if image.get("environment_key") != expected["environment_key"]:
raise ReleaseValidationError(f"deployment environment key drift for {expected['role']}")
if image.get("dockerfile") != expected["dockerfile"]:
raise ReleaseValidationError(f"Dockerfile drift for {expected['role']}")
if image.get("base_image") != expected["base_image"]:
raise ReleaseValidationError(f"base-image transport/digest drift for {expected['role']}")
if image.get("platform") != TARGET["platform"]:
raise ReleaseValidationError(f"platform drift for {expected['role']}")
expected_ref = f"{expected['name']}:{candidate['commit']}"
if image.get("image_ref") != expected_ref:
raise ReleaseValidationError(f"image reference is not commit-bound for {expected['role']}")
oci_record = _mapping(image.get("oci"), f"{expected['role']} OCI record")
docker_record = _mapping(
image.get("docker_archive"), f"{expected['role']} Docker archive record"
)
sbom_record = _mapping(image.get("sbom"), f"{expected['role']} SBOM record")
scan_record = _mapping(image.get("scan"), f"{expected['role']} scan record")
license_scan_record = _mapping(
image.get("license_scan"), f"{expected['role']} license scan record"
)
for record, kind, required_prefix in (
(oci_record, "OCI", "images/"),
(docker_record, "Docker archive", "images-docker/"),
(sbom_record, "SBOM", "sbom/"),
(scan_record, "scan", "scans/"),
(license_scan_record, "license scan", "licenses/"),
):
track(record, f"{expected['role']} {kind}")
if not _record_path(record, f"{expected['role']} {kind}").startswith(required_prefix):
raise ReleaseValidationError(f"{expected['role']} {kind} path is outside {required_prefix}")
oci_path = verify_file_record(artifact_root, oci_record, label=f"{expected['role']} OCI")
oci_summary = inspect_oci_archive(oci_path)
_compare_summary(oci_record, oci_summary, label=f"{expected['role']} OCI")
if oci_summary["revision"] != candidate["commit"] or oci_summary["version"] != candidate["tag"]:
raise ReleaseValidationError(f"OCI candidate labels drift for {expected['role']}")
if (
oci_summary["release_role"] != expected["role"]
or oci_summary["title"] != expected["name"]
):
raise ReleaseValidationError(f"OCI role labels drift for {expected['role']}")
manifest_digest = oci_summary["manifest_digest"]
if manifest_digest in oci_manifest_digests:
raise ReleaseValidationError(f"duplicate OCI manifest digest across image roles: {manifest_digest}")
oci_manifest_digests.add(manifest_digest)
docker_path = verify_file_record(
artifact_root, docker_record, label=f"{expected['role']} Docker archive"
)
docker_summary = inspect_docker_archive(
docker_path, expected_repo_tag=expected_ref
)
_compare_summary(
docker_record, docker_summary, label=f"{expected['role']} Docker archive"
)
if docker_summary["config_digest"] != oci_summary["config_digest"]:
raise ReleaseValidationError(
f"OCI/Docker archive image identity drift for {expected['role']}"
)
if (
docker_summary["revision"] != candidate["commit"]
or docker_summary["version"] != candidate["tag"]
or docker_summary["release_role"] != expected["role"]
or docker_summary["title"] != expected["name"]
):
raise ReleaseValidationError(f"Docker archive role labels drift for {expected['role']}")
sbom_path = verify_file_record(artifact_root, sbom_record, label=f"{expected['role']} SBOM")
sbom_summary = validate_spdx(
load_json_file(sbom_path, label=f"{expected['role']} SBOM"),
policy,
expected_role=expected["role"],
expected_manifest_digest=oci_summary["manifest_digest"],
)
_compare_summary(sbom_record, sbom_summary, label=f"{expected['role']} SBOM")
scan_path = verify_file_record(artifact_root, scan_record, label=f"{expected['role']} scan")
scan_evaluated_at = parse_rfc3339(
scan_record.get("waiver_evaluated_at"),
label=f"{expected['role']} scan waiver_evaluated_at",
)
if not (started <= scan_evaluated_at <= completed):
raise ReleaseValidationError(
f"{expected['role']} vulnerability waiver was not evaluated during the build"
)
scan_summary = validate_trivy_scan(
load_json_file(scan_path, label=f"{expected['role']} Trivy scan"),
policy,
expected_config_digest=oci_summary["config_digest"],
expected_artifact_basename=docker_path.name,
role=expected["role"],
at=scan_evaluated_at,
)
_compare_summary(scan_record, scan_summary, label=f"{expected['role']} scan")
if scan_record.get("scanner") != "trivy" or scan_record.get("scanner_version") != REQUIRED_TOOL_VERSIONS["trivy"]:
raise ReleaseValidationError(f"scanner identity drift for {expected['role']}")
if scan_record.get("passed") is not True:
raise ReleaseValidationError(f"scan is not passing for {expected['role']}")
for bucket in aggregate_vulnerability_counts:
for severity in aggregate_vulnerability_counts[bucket]:
aggregate_vulnerability_counts[bucket][severity] += scan_summary["counts"][
bucket
][severity]
applied_waiver_ids.update(scan_summary["waiver_ids"])
license_scan_path = verify_file_record(
artifact_root,
license_scan_record,
label=f"{expected['role']} license scan",
)
license_scan_summary = validate_trivy_license_scan(
load_json_file(
license_scan_path, label=f"{expected['role']} Trivy license scan"
),
expected_config_digest=oci_summary["config_digest"],
expected_artifact_basename=docker_path.name,
)
_compare_summary(
license_scan_record,
license_scan_summary,
label=f"{expected['role']} license scan",
)
if (
license_scan_record.get("scanner") != "trivy"
or license_scan_record.get("scanner_version")
!= REQUIRED_TOOL_VERSIONS["trivy"]
):
raise ReleaseValidationError(
f"license scanner identity drift for {expected['role']}"
)
if license_scan_record.get("passed") is not True:
raise ReleaseValidationError(
f"license scan is not passing for {expected['role']}"
)
aggregate_license_inventory["report_count"] += 1
aggregate_license_inventory["result_count"] += license_scan_summary[
"result_count"
]
aggregate_license_inventory["license_count"] += license_scan_summary[
"license_count"
]
aggregate_license_inventory["unresolved_count"] += license_scan_summary[
"unresolved_count"
]
manifest_images.append(
{
"role": expected["role"],
"image_name": expected["name"],
"environment_key": expected["environment_key"],
"image_ref": expected_ref,
"platform": TARGET["platform"],
"manifest_digest": manifest_digest,
"config_digest": oci_summary["config_digest"],
"registry_reference": f"{expected['name']}@{manifest_digest}",
"offline_image_id": oci_summary["config_digest"],
"oci": {key: oci_record[key] for key in ("path", "sha256", "size_bytes")},
"docker_archive": {
key: docker_record[key] for key in ("path", "sha256", "size_bytes")
},
"sbom": {
key: sbom_record[key]
for key in (
"path",
"sha256",
"size_bytes",
"spdx_version",
"package_count",
"non_root_package_count",
"root_package_spdx_id",
"root_package_name",
"root_role",
"root_manifest_digest",
)
},
"scan": {key: scan_record[key] for key in ("path", "sha256", "size_bytes")},
"license_scan": {
key: license_scan_record[key]
for key in (
"path",
"sha256",
"size_bytes",
"scanner",
"scanner_version",
"schema_version",
"artifact_name",
"artifact_type",
"image_id",
"result_count",
"license_count",
"unresolved_count",
"passed",
)
},
}
)
if claimed_vulnerability_counts != aggregate_vulnerability_counts:
raise ReleaseValidationError("receipt aggregate vulnerability counts drift")
if claimed_waiver_ids != sorted(applied_waiver_ids):
raise ReleaseValidationError("receipt applied waiver ID inventory drift")
if claimed_license_inventory != aggregate_license_inventory:
raise ReleaseValidationError("receipt aggregate license inventory drift")
if output.parent.resolve(strict=True) != artifact_root.resolve(strict=True):
raise ReleaseValidationError("installable manifest must be written directly in the artifact root")
_reject_unlisted_entries(artifact_root, expected_files, output)
indexed_records = [
{
"path": path,
"sha256": sha_record["sha256"],
"size_bytes": sha_record["size_bytes"],
}
for sha_record in (
policy_record,
candidate["source_manifest"],
candidate["source_archive"],
database_record,
compose_env_record,
compose_report_record,
*(
artifact
for image in images
for artifact in (
image["oci"],
image["docker_archive"],
image["sbom"],
image["scan"],
image["license_scan"],
)
),
)
for path in [sha_record["path"]]
]
indexed_records.append(file_record(receipt_path, receipt_relative))
indexed_records.sort(key=lambda item: item["path"].encode("utf-8"))
installable_without_digest = {
"schema_version": INSTALLABLE_MANIFEST_SCHEMA,
"candidate": {
"commit": candidate["commit"],
"tag": candidate["tag"],
"tag_object_oid": candidate["tag_object_oid"],
"tree": candidate["tree"],
"commit_tree": candidate["commit_tree"],
"source_manifest_payload_sha256": candidate["source_manifest_payload_sha256"],
"checkout": candidate_checkout,
},
"target": TARGET,
"policy": {
"path": policy_record["path"],
"schema_version": policy["schema_version"],
"sha256": policy_record["sha256"],
"size_bytes": policy_record["size_bytes"],
"target": TARGET,
},
"build_receipt": {
**file_record(receipt_path, receipt_relative),
"payload_sha256": receipt["receipt_payload_sha256"],
},
"security_gate": {
"passed": True,
"fail_closed": True,
"thresholds": policy["vulnerability_scan"]["thresholds"],
"exemptions": policy["vulnerability_scan"]["exemptions"],
"vulnerability_counts": aggregate_vulnerability_counts,
"waiver_ids": sorted(applied_waiver_ids),
"trivy_database": {
**database_summary,
"repository_transport": repository_transport,
},
"license_inventory": aggregate_license_inventory,
},
"execution_contract": {
"command_timeouts_seconds": COMMAND_TIMEOUTS_SECONDS,
"controlled_environment": CONTROLLED_SUBPROCESS_ENV,
"docker_host": LOCAL_DOCKER_HOST,
"pinned_executables": PINNED_EXECUTABLES,
"release_controls": release_controls,
},
"compose_gate": {
"passed": True,
"synthetic_environment": True,
"production_secrets_used": False,
"execution_engine": "self_bound_release_runner_builtin_v1",
"validator": compose_validator,
"environment": {
key: compose_env_record[key] for key in ("path", "sha256", "size_bytes")
},
"report": {
key: compose_report_record[key]
for key in ("path", "sha256", "size_bytes")
},
"case_count": compose_report_summary["case_count"],
"compose_inputs": compose_report_summary["compose_inputs"],
},
"images": manifest_images,
"docker_import_plan": {
"schema_version": "yuqei.docker-image-import-plan.v1",
"implicit_preloaded_images_allowed": False,
"images": [
{
"sequence": index,
"role": image["role"],
"environment_key": image["environment_key"],
"archive": image["docker_archive"],
"command_argv": [
PINNED_EXECUTABLES["docker"],
"--host",
LOCAL_DOCKER_HOST,
"image",
"load",
"--input",
image["docker_archive"]["path"],
],
"expected_repo_tag": image["image_ref"],
"expected_loaded_image_id": image["config_digest"],
"expected_oci_manifest_digest": image["manifest_digest"],
"offline_compose_override": image["offline_image_id"],
"registry_reference_for_online_release": image["registry_reference"],
"repo_digests_expected_after_load": [],
}
for index, image in enumerate(manifest_images, 1)
],
"consumer_requirement": (
"An installer must verify docker load produced the exact config-based image ID, "
"then use that sha256 image ID as the offline Compose override. The registry "
"reference is online-only because docker load does not restore RepoDigests. "
"This assembler does not import images."
),
},
"files": indexed_records,
"claims_not_made": [
"This manifest does not claim that an installer consumed its Docker import plan.",
"This manifest does not claim installation, deployment, migration, health, UAT, rollback, or public-release approval."
],
}
installable = with_payload_digest(installable_without_digest, "manifest_payload_sha256")
if write_output:
_write_exclusive_manifest(output, installable)
return installable
def validate_installable_release_manifest(
*,
artifact_root: Path,
manifest_path: Path,
expected_receipt_sha256: str,
repo: Path,
) -> dict[str, Any]:
"""Rebuild and byte-compare an existing manifest with the strict assembler."""
manifest_relative = _path_relative_to_root(
artifact_root,
manifest_path,
label="installable release manifest",
)
if Path(manifest_relative).parent != Path("."):
raise ReleaseValidationError(
"installable release manifest must be directly in the artifact root"
)
manifest_bytes = manifest_path.read_bytes()
manifest = load_json_file(manifest_path, label="installable release manifest")
if canonical_json_bytes(manifest) != manifest_bytes:
raise ReleaseValidationError(
"installable release manifest is not serialized as canonical JSON"
)
build_receipt = _mapping(manifest.get("build_receipt"), "installable build_receipt")
if set(build_receipt) != {"path", "sha256", "size_bytes", "payload_sha256"}:
raise ReleaseValidationError("installable build_receipt fields are missing or unknown")
receipt_relative = _record_path(build_receipt, "installable build_receipt")
receipt_path = ensure_regular_file_under(artifact_root, receipt_relative)
expected = assemble_installable_release(
artifact_root=artifact_root,
receipt_path=receipt_path,
expected_receipt_sha256=expected_receipt_sha256,
repo=repo,
output=manifest_path,
write_output=False,
)
if manifest != expected:
raise ReleaseValidationError(
"installable release manifest differs from the strict reconstructed manifest"
)
return manifest
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--artifact-root", type=Path, required=True)
parser.add_argument("--receipt", type=Path, help="defaults to ARTIFACT_ROOT/build-receipt.json")
parser.add_argument(
"--expected-build-receipt-sha256",
required=True,
help="Independently supplied raw SHA-256 of the canonical build receipt.",
)
parser.add_argument("--repo", type=Path, default=Path.cwd(), help="Git repository containing the exact tag")
parser.add_argument("--output", type=Path, help="defaults to ARTIFACT_ROOT/installable-release-manifest.json")
args = parser.parse_args(argv)
root = args.artifact_root.resolve()
receipt = args.receipt.resolve() if args.receipt else root / "build-receipt.json"
output = args.output.resolve() if args.output else root / "installable-release-manifest.json"
try:
manifest = assemble_installable_release(
artifact_root=root,
receipt_path=receipt,
expected_receipt_sha256=args.expected_build_receipt_sha256,
repo=args.repo.resolve(),
output=output,
)
print(
json.dumps(
{
"status": "passed",
"output": str(output),
"commit": manifest["candidate"]["commit"],
"images": len(manifest["images"]),
},
sort_keys=True,
)
)
return 0
except (OSError, ReleaseValidationError) as error:
print(f"installable_release_error: {error}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())