From d56c7642671268346dff8b13deffad8f2b2201db Mon Sep 17 00:00:00 2001 From: yuqei Date: Sun, 27 Sep 2026 02:18:04 +0800 Subject: [PATCH] release: validate infrastructure bundle and first-release recovery --- .../plan-first-release-snapshot-restore.py | 1022 +++++++++++++++++ .../build-infrastructure-image-bundle.py | 754 ++++++++++++ ..._first_release_snapshot_restore_planner.py | 570 +++++++++ .../tests/test_infrastructure_image_bundle.py | 451 ++++++++ 4 files changed, 2797 insertions(+) create mode 100644 yuqei-infra/deploy/plan-first-release-snapshot-restore.py create mode 100644 yuqei-infra/scripts/build-infrastructure-image-bundle.py create mode 100644 yuqei-infra/tests/test_first_release_snapshot_restore_planner.py create mode 100644 yuqei-infra/tests/test_infrastructure_image_bundle.py diff --git a/yuqei-infra/deploy/plan-first-release-snapshot-restore.py b/yuqei-infra/deploy/plan-first-release-snapshot-restore.py new file mode 100644 index 00000000..5b6cf695 --- /dev/null +++ b/yuqei-infra/deploy/plan-first-release-snapshot-restore.py @@ -0,0 +1,1022 @@ +#!/usr/bin/env python3 +"""Verify first-release snapshot evidence and emit a non-writing restore plan. + +This command intentionally has no apply mode. A whole-host restore must run +from an independently authorized external control plane; this planner only +verifies immutable evidence, detached signatures, and the exact recovery scope. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import re +import stat +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path, PurePosixPath +from typing import Any, Mapping + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + +EVIDENCE_SCHEMA = "yuqei.first-release-snapshot-evidence.v1" +KEYRING_SCHEMA = "yuqei.first-release-recovery-keyring.v1" +APPROVAL_SCHEMA = "yuqei.first-release-recovery-approval.v1" +NO_PRIOR_RELEASE_SCHEMA = "yuqei.no-prior-successful-release.v1" +SNAPSHOT_RECEIPT_SCHEMA = "yuqei.host-snapshot-receipt.v1" +BACKUP_MANIFEST_SCHEMA = "yuqei-production-backup.v2" +RESTORE_TEST_SCHEMA = "yuqei.first-release-restore-test.v1" +PLAN_SCHEMA = "yuqei.first-release-snapshot-restore-plan.v1" +IMAGE_IDENTITY_SCHEMA = "yuqei.release-image-identity.v2" +RECOVERY_STRATEGY = "first_release_snapshot_restore" +RECOVERY_EVIDENCE_STATUS = "runtime_signed_evidence_required" + +REQUIRED_ROLES = frozenset({"release_manager", "database_owner"}) +REQUIRED_COMPONENTS = ( + "host_runtime_state", + "postgres_consistent_backup", + "object_storage_state", + "release_environment_and_tls_bindings", +) +REQUIRED_DATABASES = ("contract", "ai", "operations") +REQUIRED_STORAGE_TARGETS = ("contract_files_volume", "minio_object_store") +MAX_EVIDENCE_AGE = timedelta(hours=24) +MAX_CLOCK_SKEW = timedelta(minutes=5) +MAX_JSON_BYTES = 8 * 1024 * 1024 +SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/+-]{0,255}$") +TRUSTED_ROOT_KEY_ID = "yuqei-first-release-recovery-root-v1" +# Provision these pins together in a separately reviewed frozen release. Empty +# values intentionally keep the planner fail closed. A caller must never be +# able to nominate its own root or keyring revocation state on the command line. +TRUSTED_ROOT_PUBLIC_KEY_SHA256 = "" +TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256 = "" +MIN_KEYRING_EPOCH: int | None = None + + +class SnapshotRestorePlanError(RuntimeError): + """Raised when first-release recovery evidence is unsafe or incomplete.""" + + +def canonical_json_bytes(value: object) -> bytes: + try: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise SnapshotRestorePlanError("value cannot be encoded as canonical JSON") from exc + + +def canonical_sha256(value: object) -> str: + return hashlib.sha256(canonical_json_bytes(value)).hexdigest() + + +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _reject_constant(value: str) -> None: + raise SnapshotRestorePlanError(f"non-finite JSON value is forbidden: {value}") + + +def _reject_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise SnapshotRestorePlanError(f"duplicate JSON key is forbidden: {key}") + result[key] = value + return result + + +def _absolute(path: Path) -> Path: + return Path(os.path.abspath(os.fspath(path))) + + +def _reject_symlink_components(path: Path, *, label: str) -> None: + absolute = _absolute(path) + current = Path(absolute.anchor) + for part in absolute.parts[1:]: + current /= part + try: + if current.exists() and current.is_symlink(): + raise SnapshotRestorePlanError(f"{label} contains a symbolic link") + except OSError as exc: + raise SnapshotRestorePlanError(f"cannot inspect {label} path") from exc + + +def _regular_file(path: Path, *, label: str, max_bytes: int = MAX_JSON_BYTES) -> Path: + lexical = _absolute(path) + _reject_symlink_components(lexical, label=label) + try: + resolved = lexical.resolve(strict=True) + metadata = resolved.stat() + except OSError as exc: + raise SnapshotRestorePlanError(f"{label} is missing") from exc + if lexical.is_symlink() or not stat.S_ISREG(metadata.st_mode): + raise SnapshotRestorePlanError(f"{label} must be a regular file") + if metadata.st_nlink != 1: + raise SnapshotRestorePlanError(f"{label} must not be hard linked") + if metadata.st_size <= 0 or metadata.st_size > max_bytes: + raise SnapshotRestorePlanError(f"{label} has an invalid size") + return resolved + + +def load_json(path: Path, *, label: str) -> dict[str, Any]: + target = _regular_file(path, label=label) + try: + payload = json.loads( + target.read_text(encoding="utf-8-sig"), + object_pairs_hook=_reject_duplicate_pairs, + parse_constant=_reject_constant, + ) + except SnapshotRestorePlanError: + raise + except (OSError, UnicodeError, ValueError) as exc: + raise SnapshotRestorePlanError(f"{label} is not strict UTF-8 JSON") from exc + if not isinstance(payload, dict): + raise SnapshotRestorePlanError(f"{label} must be a JSON object") + return payload + + +def _exact(value: object, fields: set[str], *, label: str) -> dict[str, Any]: + if not isinstance(value, dict) or set(value) != fields: + raise SnapshotRestorePlanError(f"{label} fields are missing or unknown") + return dict(value) + + +def _sha(value: object, *, label: str) -> str: + if not isinstance(value, str) or not SHA256_RE.fullmatch(value): + raise SnapshotRestorePlanError(f"{label} must be a lowercase SHA-256") + return value + + +def _identifier(value: object, *, label: str) -> str: + if not isinstance(value, str) or not IDENTIFIER_RE.fullmatch(value): + raise SnapshotRestorePlanError(f"{label} is missing or unsafe") + return value + + +def _timestamp(value: object, *, label: str) -> datetime: + if not isinstance(value, str): + raise SnapshotRestorePlanError(f"{label} must be RFC3339 text") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise SnapshotRestorePlanError(f"{label} is not RFC3339") from exc + if parsed.tzinfo is None: + raise SnapshotRestorePlanError(f"{label} must include a timezone") + return parsed.astimezone(UTC) + + +def _embedded_hash(payload: Mapping[str, Any], field: str, *, label: str) -> None: + expected = _sha(payload.get(field), label=f"{label} {field}") + unsigned = {key: value for key, value in payload.items() if key != field} + if canonical_sha256(unsigned) != expected: + raise SnapshotRestorePlanError(f"{label} payload hash mismatch") + + +def _safe_relative(value: object, *, label: str) -> str: + if not isinstance(value, str) or not value or "\\" in value or "\x00" in value: + raise SnapshotRestorePlanError(f"{label} must be a canonical relative path") + pure = PurePosixPath(value) + if pure.is_absolute() or pure.as_posix() != value or any( + part in {"", ".", ".."} for part in pure.parts + ): + raise SnapshotRestorePlanError(f"{label} is unsafe") + return value + + +def _package_artifact( + package_file: Path, + package: Mapping[str, Any], + name: str, +) -> tuple[Path, str]: + artifacts = package.get("artifacts") + if not isinstance(artifacts, dict): + raise SnapshotRestorePlanError("release package has no artifact inventory") + record = _exact(artifacts.get(name), {"path", "sha256"}, label=f"artifact {name}") + relative = _safe_relative(record.get("path"), label=f"artifact {name} path") + expected = _sha(record.get("sha256"), label=f"artifact {name} SHA-256") + root = package_file.parent.resolve() + target = _regular_file(root.joinpath(*PurePosixPath(relative).parts), label=f"artifact {name}") + try: + target.relative_to(root) + except ValueError as exc: + raise SnapshotRestorePlanError(f"artifact {name} escapes the package") from exc + if file_sha256(target) != expected: + raise SnapshotRestorePlanError(f"artifact {name} SHA-256 mismatch") + return target, expected + + +def _external_record(value: object, *, label: str) -> tuple[dict[str, Any], str, Path]: + record = _exact(value, {"path", "sha256"}, label=f"{label} record") + raw_path = record.get("path") + if not isinstance(raw_path, str) or not Path(raw_path).is_absolute(): + raise SnapshotRestorePlanError(f"{label} path must be absolute") + expected = _sha(record.get("sha256"), label=f"{label} external SHA-256") + path = _regular_file(Path(raw_path), label=label) + if file_sha256(path) != expected: + raise SnapshotRestorePlanError(f"{label} SHA-256 mismatch") + return load_json(path, label=label), expected, path + + +def _validate_package( + package_file: Path, + expected_package_sha256: str, +) -> dict[str, Any]: + package_path = _regular_file(package_file, label="release package") + expected = _sha(expected_package_sha256, label="expected package SHA-256") + actual = file_sha256(package_path) + if actual != expected: + raise SnapshotRestorePlanError("release package does not match the approved SHA-256") + package = load_json(package_path, label="release package") + _embedded_hash(package, "package_payload_sha256", label="release package") + release_id = _identifier(package.get("release_id"), label="release_id") + image_path, image_sha = _package_artifact(package_path, package, "image_identity") + identity = load_json(image_path, label="image identity") + if identity.get("schema_version") != IMAGE_IDENTITY_SCHEMA: + raise SnapshotRestorePlanError("first-release recovery requires image identity v2") + _embedded_hash(identity, "identity_payload_sha256", label="image identity") + if identity.get("release_id") != release_id: + raise SnapshotRestorePlanError("image identity release differs from the package") + recovery = identity.get("recovery") + recovery_fields = { + "strategy", + "evidence_status", + "first_release_only", + "predecessor_package_required", + "database_downgrade_allowed", + "required_restore_scopes", + "required_postgres_databases", + "required_storage_targets", + } + if not isinstance(recovery, dict) or set(recovery) != recovery_fields: + raise SnapshotRestorePlanError("package first-release recovery fields are incomplete") + if recovery.get("strategy") != RECOVERY_STRATEGY: + raise SnapshotRestorePlanError("package is not configured for first-release snapshot recovery") + if ( + recovery.get("first_release_only") is not True + or recovery.get("predecessor_package_required") is not False + or recovery.get("database_downgrade_allowed") is not False + or recovery.get("required_restore_scopes") != list(REQUIRED_COMPONENTS) + or recovery.get("required_postgres_databases") != list(REQUIRED_DATABASES) + or recovery.get("required_storage_targets") != list(REQUIRED_STORAGE_TARGETS) + or recovery.get("evidence_status") != RECOVERY_EVIDENCE_STATUS + ): + raise SnapshotRestorePlanError("package first-release recovery policy is unsafe") + if any(key in recovery for key in ("target_release_id", "application_images", "predecessor_package")): + raise SnapshotRestorePlanError("first-release package must not contain a fake predecessor") + source_commit = identity.get("source_commit") + if not isinstance(source_commit, str) or not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", source_commit): + raise SnapshotRestorePlanError("image identity source commit is invalid") + return { + "release_id": release_id, + "package_sha256": actual, + "source_commit": source_commit, + "image_identity_sha256": image_sha, + "compose_project_name": package.get("compose_project_name"), + } + + +def _validate_keyring( + keyring: Mapping[str, Any], + *, + root_public_key: Ed25519PublicKey, + now: datetime, +) -> dict[str, dict[str, Any]]: + _exact( + keyring, + { + "schema_version", + "root_key_id", + "epoch", + "issued_at", + "expires_at", + "keys", + "keyring_payload_sha256", + "root_signature_base64", + }, + label="recovery keyring", + ) + if keyring.get("schema_version") != KEYRING_SCHEMA: + raise SnapshotRestorePlanError("unsupported recovery keyring schema") + unsigned = { + key: keyring[key] + for key in ( + "schema_version", + "root_key_id", + "epoch", + "issued_at", + "expires_at", + "keys", + ) + } + embedded = _sha( + keyring.get("keyring_payload_sha256"), label="keyring payload hash" + ) + if canonical_sha256(unsigned) != embedded: + raise SnapshotRestorePlanError("recovery keyring payload hash mismatch") + if not SHA256_RE.fullmatch(TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256): + raise SnapshotRestorePlanError( + "current first-release recovery keyring pin is not provisioned" + ) + if ( + not isinstance(MIN_KEYRING_EPOCH, int) + or isinstance(MIN_KEYRING_EPOCH, bool) + or MIN_KEYRING_EPOCH < 1 + ): + raise SnapshotRestorePlanError( + "minimum first-release recovery keyring epoch is not provisioned" + ) + if keyring.get("root_key_id") != TRUSTED_ROOT_KEY_ID: + raise SnapshotRestorePlanError("recovery keyring uses the wrong root key") + try: + root_public_key.verify( + base64.b64decode(keyring.get("root_signature_base64"), validate=True), + canonical_json_bytes(unsigned), + ) + except (TypeError, ValueError, InvalidSignature) as exc: + raise SnapshotRestorePlanError("recovery keyring root signature is invalid") from exc + epoch = keyring.get("epoch") + if not isinstance(epoch, int) or isinstance(epoch, bool) or epoch < 1: + raise SnapshotRestorePlanError("recovery keyring epoch is invalid") + if epoch < MIN_KEYRING_EPOCH: + raise SnapshotRestorePlanError("recovery keyring epoch is older than the trusted minimum") + if embedded != TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256: + raise SnapshotRestorePlanError( + "recovery keyring is not the currently trusted revocation state" + ) + issued = _timestamp(keyring.get("issued_at"), label="keyring issued_at") + expires = _timestamp(keyring.get("expires_at"), label="keyring expires_at") + if issued > now + MAX_CLOCK_SKEW or now >= expires or expires <= issued: + raise SnapshotRestorePlanError("recovery keyring is expired or not yet valid") + raw_keys = keyring.get("keys") + if not isinstance(raw_keys, list) or len(raw_keys) < 2: + raise SnapshotRestorePlanError("recovery keyring has insufficient keys") + result: dict[str, dict[str, Any]] = {} + for raw in raw_keys: + item = _exact( + raw, + { + "key_id", + "approver_id", + "role", + "public_key_base64", + "not_before", + "not_after", + "revoked", + }, + label="recovery key", + ) + key_id = _identifier(item.get("key_id"), label="key_id") + if key_id in result: + raise SnapshotRestorePlanError("recovery keyring repeats a key_id") + role = item.get("role") + if role not in REQUIRED_ROLES: + raise SnapshotRestorePlanError("recovery key has an unsupported role") + if item.get("revoked") is not False: + raise SnapshotRestorePlanError("revoked recovery keys are forbidden") + try: + public_raw = base64.b64decode(item.get("public_key_base64"), validate=True) + public_key = Ed25519PublicKey.from_public_bytes(public_raw) + except (TypeError, ValueError) as exc: + raise SnapshotRestorePlanError("recovery public key is invalid") from exc + not_before = _timestamp(item.get("not_before"), label="key not_before") + not_after = _timestamp(item.get("not_after"), label="key not_after") + if not_after <= not_before: + raise SnapshotRestorePlanError("recovery key validity window is invalid") + result[key_id] = { + **item, + "public_key": public_key, + "not_before_parsed": not_before, + "not_after_parsed": not_after, + } + return result + + +def _validate_no_prior_release( + value: object, + *, + environment_id: str, + release_id: str, +) -> tuple[dict[str, Any], str, datetime]: + payload, digest, _path = _external_record(value, label="no-prior-release receipt") + _exact( + payload, + { + "schema_version", + "environment_id", + "candidate_release_id", + "successful_release_count", + "status", + "observed_at", + "receipt_payload_sha256", + }, + label="no-prior-release receipt", + ) + if payload.get("schema_version") != NO_PRIOR_RELEASE_SCHEMA: + raise SnapshotRestorePlanError("unsupported no-prior-release receipt schema") + _embedded_hash(payload, "receipt_payload_sha256", label="no-prior-release receipt") + if ( + payload.get("environment_id") != environment_id + or payload.get("candidate_release_id") != release_id + or payload.get("successful_release_count") != 0 + or payload.get("status") != "confirmed_no_prior_successful_release" + ): + raise SnapshotRestorePlanError("no-prior-release receipt does not prove a first release") + return payload, digest, _timestamp(payload.get("observed_at"), label="no-prior observed_at") + + +def _validate_snapshot_component( + value: object, + *, + host_id: str, + consistency_set_id: str, +) -> tuple[dict[str, Any], str, datetime]: + component = _exact( + value, + {"provider_receipt", "off_host", "immutable", "status"}, + label="host_runtime_state", + ) + if component.get("off_host") is not True or component.get("immutable") is not True or component.get("status") != "available": + raise SnapshotRestorePlanError("host snapshot must be off-host, immutable, and available") + receipt, digest, _path = _external_record( + component.get("provider_receipt"), label="host snapshot provider receipt" + ) + _exact( + receipt, + { + "schema_version", + "provider", + "control_plane_id", + "snapshot_id", + "source_host_id", + "consistency_set_id", + "off_host", + "immutable", + "status", + "created_at", + "receipt_payload_sha256", + }, + label="host snapshot provider receipt", + ) + if receipt.get("schema_version") != SNAPSHOT_RECEIPT_SCHEMA: + raise SnapshotRestorePlanError("unsupported host snapshot receipt schema") + _embedded_hash(receipt, "receipt_payload_sha256", label="host snapshot receipt") + if ( + receipt.get("source_host_id") != host_id + or receipt.get("consistency_set_id") != consistency_set_id + or receipt.get("off_host") is not True + or receipt.get("immutable") is not True + or receipt.get("status") != "available" + ): + raise SnapshotRestorePlanError("host snapshot receipt does not match the recovery target") + _identifier(receipt.get("provider"), label="snapshot provider") + _identifier(receipt.get("control_plane_id"), label="snapshot control_plane_id") + _identifier(receipt.get("snapshot_id"), label="snapshot_id") + return receipt, digest, _timestamp(receipt.get("created_at"), label="snapshot created_at") + + +def _validate_backup_components( + postgres_value: object, + object_value: object, + *, + release_id: str, + environment_id: str, + consistency_set_id: str, +) -> tuple[dict[str, Any], str, datetime]: + postgres = _exact( + postgres_value, + {"backup_manifest", "consistency_mode", "databases"}, + label="postgres_consistent_backup", + ) + if postgres.get("consistency_mode") != "quiesced" or postgres.get("databases") != list(REQUIRED_DATABASES): + raise SnapshotRestorePlanError("postgres backup must be quiesced and cover contract/ai/operations") + objects = _exact( + object_value, + {"backup_manifest", "stores"}, + label="object_storage_state", + ) + if objects.get("stores") != list(REQUIRED_STORAGE_TARGETS): + raise SnapshotRestorePlanError( + "object storage state must cover contract_files_volume and minio_object_store" + ) + if objects.get("backup_manifest") != postgres.get("backup_manifest"): + raise SnapshotRestorePlanError("database and object storage must share one backup manifest") + manifest, digest, _path = _external_record( + postgres.get("backup_manifest"), label="consistent backup manifest" + ) + _exact( + manifest, + { + "schema_version", + "backup_id", + "release_id", + "environment_id", + "consistency_set_id", + "consistency_mode", + "databases", + "contract_files", + "minio", + "completed_at", + "status", + "manifest_payload_sha256", + }, + label="consistent backup manifest", + ) + if manifest.get("schema_version") != BACKUP_MANIFEST_SCHEMA: + raise SnapshotRestorePlanError("unsupported consistent backup manifest schema") + _embedded_hash(manifest, "manifest_payload_sha256", label="consistent backup manifest") + if ( + manifest.get("release_id") != release_id + or manifest.get("environment_id") != environment_id + or manifest.get("consistency_set_id") != consistency_set_id + or manifest.get("consistency_mode") != "quiesced" + or manifest.get("status") != "complete" + ): + raise SnapshotRestorePlanError("consistent backup manifest is bound to another recovery set") + databases = manifest.get("databases") + if ( + not isinstance(databases, dict) + or len(databases) != len(REQUIRED_DATABASES) + or set(databases) != set(REQUIRED_DATABASES) + ): + raise SnapshotRestorePlanError("consistent backup manifest database inventory is incomplete") + for name in REQUIRED_DATABASES: + record = _exact(databases[name], {"artifact_sha256", "size_bytes", "schema_revision"}, label=f"{name} backup") + _sha(record.get("artifact_sha256"), label=f"{name} backup SHA-256") + if not isinstance(record.get("size_bytes"), int) or isinstance(record.get("size_bytes"), bool) or record["size_bytes"] <= 0: + raise SnapshotRestorePlanError(f"{name} backup size is invalid") + _identifier(record.get("schema_revision"), label=f"{name} schema revision") + contract_files = _exact( + manifest.get("contract_files"), + {"archive_sha256", "size_bytes", "file_count"}, + label="contract_files backup", + ) + _sha(contract_files.get("archive_sha256"), label="contract_files archive SHA-256") + minio = _exact( + manifest.get("minio"), + {"content_index_sha256", "object_count", "total_size_bytes"}, + label="MinIO backup", + ) + _sha(minio.get("content_index_sha256"), label="MinIO content index SHA-256") + for record, fields, label in ( + (contract_files, ("size_bytes", "file_count"), "contract_files"), + (minio, ("object_count", "total_size_bytes"), "MinIO"), + ): + if any(not isinstance(record.get(field), int) or isinstance(record.get(field), bool) or record[field] < 0 for field in fields): + raise SnapshotRestorePlanError(f"{label} backup counters are invalid") + return manifest, digest, _timestamp(manifest.get("completed_at"), label="backup completed_at") + + +def _validate_bindings(value: object) -> dict[str, str]: + bindings = _exact( + value, + { + "production_env_sha256", + "tls_certificate_sha256", + "tls_private_key_sha256", + "tls_trusted_ca_sha256", + }, + label="release_environment_and_tls_bindings", + ) + return {key: _sha(bindings[key], label=key) for key in sorted(bindings)} + + +def _validate_restore_test( + value: object, + *, + release_id: str, + environment_id: str, + consistency_set_id: str, + snapshot_receipt_sha256: str, + backup_manifest_sha256: str, + bindings_sha256: str, +) -> tuple[dict[str, Any], str, datetime]: + payload, digest, _path = _external_record(value, label="independent restore test") + _exact( + payload, + { + "schema_version", + "release_id", + "source_environment_id", + "restore_environment_id", + "restore_environment_class", + "consistency_set_id", + "snapshot_receipt_sha256", + "backup_manifest_sha256", + "release_environment_and_tls_bindings_sha256", + "host_snapshot_restored", + "databases_restored", + "contract_files_restored", + "minio_restored", + "environment_tls_bindings_verified", + "database_downgrade_performed", + "status", + "tested_at", + "evidence_payload_sha256", + }, + label="independent restore test", + ) + if payload.get("schema_version") != RESTORE_TEST_SCHEMA: + raise SnapshotRestorePlanError("unsupported independent restore-test schema") + _embedded_hash(payload, "evidence_payload_sha256", label="independent restore test") + if ( + payload.get("release_id") != release_id + or payload.get("source_environment_id") != environment_id + or payload.get("restore_environment_id") == environment_id + or payload.get("restore_environment_class") != "isolated_uat" + or payload.get("consistency_set_id") != consistency_set_id + or payload.get("snapshot_receipt_sha256") != snapshot_receipt_sha256 + or payload.get("backup_manifest_sha256") != backup_manifest_sha256 + or payload.get("release_environment_and_tls_bindings_sha256") + != bindings_sha256 + or payload.get("host_snapshot_restored") is not True + or payload.get("databases_restored") != list(REQUIRED_DATABASES) + or payload.get("contract_files_restored") is not True + or payload.get("minio_restored") is not True + or payload.get("environment_tls_bindings_verified") is not True + or payload.get("database_downgrade_performed") is not False + or payload.get("status") != "passed" + ): + raise SnapshotRestorePlanError("independent restore test is incomplete or mismatched") + _identifier(payload.get("restore_environment_id"), label="restore environment_id") + return payload, digest, _timestamp(payload.get("tested_at"), label="restore tested_at") + + +def _validate_approvals( + value: object, + *, + evidence_payload_sha256: str, + trusted_keys: Mapping[str, Mapping[str, Any]], + evidence_not_before: datetime, + now: datetime, +) -> list[dict[str, Any]]: + if not isinstance(value, list) or len(value) != 2: + raise SnapshotRestorePlanError("exactly two recovery approvals are required") + roles: set[str] = set() + people: set[str] = set() + key_ids: set[str] = set() + approval_ids: set[str] = set() + approvals: list[dict[str, Any]] = [] + fields = { + "schema_version", + "approval_id", + "approver_id", + "role", + "key_id", + "signed_at", + "expires_at", + "evidence_payload_sha256", + "signature_base64", + } + for raw in value: + item = _exact(raw, fields, label="recovery approval") + if item.get("schema_version") != APPROVAL_SCHEMA: + raise SnapshotRestorePlanError("unsupported recovery approval schema") + approval_id = _identifier(item.get("approval_id"), label="approval_id") + if approval_id in approval_ids: + raise SnapshotRestorePlanError("recovery approvals repeat an approval_id") + approval_ids.add(approval_id) + person = _identifier(item.get("approver_id"), label="approver_id") + role = item.get("role") + if role not in REQUIRED_ROLES: + raise SnapshotRestorePlanError("recovery approval role is invalid") + key_id = _identifier(item.get("key_id"), label="approval key_id") + key = trusted_keys.get(key_id) + if key is None or key["approver_id"] != person or key["role"] != role: + raise SnapshotRestorePlanError("recovery approval does not match a trusted key") + if item.get("evidence_payload_sha256") != evidence_payload_sha256: + raise SnapshotRestorePlanError("recovery approval is bound to different evidence") + signed_at = _timestamp(item.get("signed_at"), label="approval signed_at") + expires_at = _timestamp(item.get("expires_at"), label="approval expires_at") + if ( + signed_at < evidence_not_before + or signed_at > now + MAX_CLOCK_SKEW + or expires_at <= signed_at + or now >= expires_at + or signed_at < key["not_before_parsed"] + or expires_at > key["not_after_parsed"] + ): + raise SnapshotRestorePlanError("recovery approval validity window is invalid") + signed = {field: item[field] for field in item if field != "signature_base64"} + try: + signature = base64.b64decode(item.get("signature_base64"), validate=True) + key["public_key"].verify(signature, canonical_json_bytes(signed)) + except (TypeError, ValueError, InvalidSignature) as exc: + raise SnapshotRestorePlanError("recovery approval signature is invalid") from exc + roles.add(str(role)) + people.add(person) + key_ids.add(key_id) + approvals.append(item) + if roles != REQUIRED_ROLES or len(people) != 2 or len(key_ids) != 2: + raise SnapshotRestorePlanError("recovery approvals require two roles, people, and keys") + return approvals + + +def validate_evidence( + evidence: Mapping[str, Any], + *, + package_binding: Mapping[str, Any], + keyring: Mapping[str, Any], + trusted_root_public_key: Ed25519PublicKey, + now: datetime, +) -> dict[str, Any]: + if now.tzinfo is None: + raise SnapshotRestorePlanError("validation time must include a timezone") + current = now.astimezone(UTC) + _exact( + evidence, + { + "schema_version", + "evidence_id", + "mode", + "release_binding", + "target", + "consistency_set_id", + "captured_at", + "expires_at", + "no_prior_successful_release_receipt", + "components", + "independent_restore_test", + "evidence_payload_sha256", + "approvals", + }, + label="first-release recovery evidence", + ) + if evidence.get("schema_version") != EVIDENCE_SCHEMA or evidence.get("mode") != "FIRST_RELEASE_PREINSTALL_SNAPSHOT_RESTORE": + raise SnapshotRestorePlanError("unsupported first-release recovery evidence") + evidence_id = _identifier(evidence.get("evidence_id"), label="evidence_id") + release_binding = _exact( + evidence.get("release_binding"), + {"release_id", "package_sha256", "source_commit", "image_identity_sha256"}, + label="release binding", + ) + if release_binding != { + key: package_binding[key] + for key in ("release_id", "package_sha256", "source_commit", "image_identity_sha256") + }: + raise SnapshotRestorePlanError("recovery evidence is bound to a different package") + target = _exact( + evidence.get("target"), + { + "environment_id", + "environment_class", + "host_id", + "host_identity_sha256", + "compose_project_name", + }, + label="recovery target", + ) + environment_id = _identifier(target.get("environment_id"), label="environment_id") + host_id = _identifier(target.get("host_id"), label="host_id") + if target.get("environment_class") not in {"isolated_uat", "production"}: + raise SnapshotRestorePlanError("recovery target environment class is invalid") + if target.get("compose_project_name") != package_binding.get("compose_project_name"): + raise SnapshotRestorePlanError("recovery target Compose project differs from the package") + _sha(target.get("host_identity_sha256"), label="host identity SHA-256") + consistency_set_id = _identifier( + evidence.get("consistency_set_id"), label="consistency_set_id" + ) + captured_at = _timestamp(evidence.get("captured_at"), label="captured_at") + expires_at = _timestamp(evidence.get("expires_at"), label="expires_at") + if ( + captured_at > current + MAX_CLOCK_SKEW + or current - captured_at > MAX_EVIDENCE_AGE + or expires_at <= captured_at + or current >= expires_at + or expires_at - captured_at > MAX_EVIDENCE_AGE + ): + raise SnapshotRestorePlanError("first-release recovery evidence is stale or expired") + + no_prior, no_prior_sha, no_prior_at = _validate_no_prior_release( + evidence.get("no_prior_successful_release_receipt"), + environment_id=environment_id, + release_id=package_binding["release_id"], + ) + components = _exact( + evidence.get("components"), set(REQUIRED_COMPONENTS), label="recovery components" + ) + snapshot, snapshot_sha, snapshot_at = _validate_snapshot_component( + components["host_runtime_state"], + host_id=host_id, + consistency_set_id=consistency_set_id, + ) + backup, backup_sha, backup_at = _validate_backup_components( + components["postgres_consistent_backup"], + components["object_storage_state"], + release_id=package_binding["release_id"], + environment_id=environment_id, + consistency_set_id=consistency_set_id, + ) + bindings = _validate_bindings(components["release_environment_and_tls_bindings"]) + bindings_sha = canonical_sha256(bindings) + restore, restore_sha, restored_at = _validate_restore_test( + evidence.get("independent_restore_test"), + release_id=package_binding["release_id"], + environment_id=environment_id, + consistency_set_id=consistency_set_id, + snapshot_receipt_sha256=snapshot_sha, + backup_manifest_sha256=backup_sha, + bindings_sha256=bindings_sha, + ) + evidence_times = (no_prior_at, snapshot_at, backup_at, restored_at) + if any(item > current + MAX_CLOCK_SKEW for item in evidence_times): + raise SnapshotRestorePlanError("recovery component timestamp is in the future") + if any(current - item > MAX_EVIDENCE_AGE for item in evidence_times): + raise SnapshotRestorePlanError("recovery component evidence is stale") + if restored_at < max(snapshot_at, backup_at) or captured_at < max(evidence_times): + raise SnapshotRestorePlanError("restore test or capture predates the bound snapshot/backup") + + unsigned = { + key: evidence[key] + for key in evidence + if key not in {"evidence_payload_sha256", "approvals"} + } + evidence_hash = _sha( + evidence.get("evidence_payload_sha256"), label="evidence payload hash" + ) + if canonical_sha256(unsigned) != evidence_hash: + raise SnapshotRestorePlanError("first-release evidence payload hash mismatch") + trusted_keys = _validate_keyring( + keyring, root_public_key=trusted_root_public_key, now=current + ) + approvals = _validate_approvals( + evidence.get("approvals"), + evidence_payload_sha256=evidence_hash, + trusted_keys=trusted_keys, + evidence_not_before=max(captured_at, no_prior_at, snapshot_at, backup_at, restored_at), + now=current, + ) + return { + "evidence_id": evidence_id, + "evidence_payload_sha256": evidence_hash, + "release_binding": release_binding, + "target": target, + "consistency_set_id": consistency_set_id, + "captured_at": evidence["captured_at"], + "expires_at": evidence["expires_at"], + "no_prior_release_receipt_sha256": no_prior_sha, + "snapshot": { + "provider": snapshot["provider"], + "control_plane_id": snapshot["control_plane_id"], + "snapshot_id": snapshot["snapshot_id"], + "provider_receipt_sha256": snapshot_sha, + }, + "backup": { + "backup_id": backup["backup_id"], + "manifest_sha256": backup_sha, + "databases": list(REQUIRED_DATABASES), + "storage_targets": list(REQUIRED_STORAGE_TARGETS), + }, + "bindings": {**bindings, "bindings_payload_sha256": bindings_sha}, + "restore_test": { + "restore_environment_id": restore["restore_environment_id"], + "evidence_sha256": restore_sha, + "tested_at": restore["tested_at"], + }, + "approvals": approvals, + "keyring": { + "epoch": keyring["epoch"], + "payload_sha256": keyring["keyring_payload_sha256"], + }, + } + + +def build_plan( + *, + package_binding: Mapping[str, Any], + validated_evidence: Mapping[str, Any], + trusted_root_public_key_sha256: str, + generated_at: datetime, +) -> dict[str, Any]: + if generated_at.tzinfo is None: + raise SnapshotRestorePlanError("plan time must include a timezone") + unsigned: dict[str, Any] = { + "schema_version": PLAN_SCHEMA, + "status": "signature_verified_against_frozen_pins_plan_only", + "generated_at": generated_at.astimezone(UTC).isoformat().replace("+00:00", "Z"), + "mode": "FIRST_RELEASE_PREINSTALL_SNAPSHOT_RESTORE", + "release_binding": dict(validated_evidence["release_binding"]), + "target": dict(validated_evidence["target"]), + "consistency_set_id": validated_evidence["consistency_set_id"], + "evidence_id": validated_evidence["evidence_id"], + "evidence_payload_sha256": validated_evidence["evidence_payload_sha256"], + "trusted_root_public_key_sha256": _sha( + trusted_root_public_key_sha256, label="trusted root public key SHA-256" + ), + "keyring": dict(validated_evidence["keyring"]), + "approval_ids": sorted( + item["approval_id"] for item in validated_evidence["approvals"] + ), + "snapshot": dict(validated_evidence["snapshot"]), + "backup": dict(validated_evidence["backup"]), + "bindings": dict(validated_evidence["bindings"]), + "independent_restore_test": dict(validated_evidence["restore_test"]), + "restore_sequence": [ + "obtain_separate_destructive_restore_authorization", + "confirm_no_public_traffic_or_confirm_old_service_switchback", + "isolate_and_power_off_exact_target_host", + "restore_immutable_host_runtime_snapshot_from_external_control_plane", + "verify_contract_ai_operations_database_state", + "verify_contract_files_and_minio_state", + "verify_release_environment_and_tls_hash_bindings", + "run_health_and_main_journey_acceptance", + ], + "required_components": list(REQUIRED_COMPONENTS), + "database_downgrade_allowed": False, + "fake_predecessor_package_allowed": False, + "execution_entrypoint_available": False, + "apply_allowed": False, + "writes_performed": False, + "docker_invoked": False, + "remote_connection_attempted": False, + "traffic_or_dns_change_performed": False, + "claims_not_made": [ + "No snapshot was created or restored by this plan-only command.", + "Temporary test keys do not establish a production trust anchor.", + "This plan is not UAT, production, pilot, public-release, or legal approval.", + ], + } + unsigned["plan_payload_sha256"] = canonical_sha256(unsigned) + return unsigned + + +def _read_root_public_key(path: Path) -> tuple[Ed25519PublicKey, str]: + if not SHA256_RE.fullmatch(TRUSTED_ROOT_PUBLIC_KEY_SHA256): + raise SnapshotRestorePlanError( + "organization first-release recovery root key is not provisioned" + ) + target = _regular_file(path, label="trusted root public key", max_bytes=32) + raw = target.read_bytes() + if len(raw) != 32: + raise SnapshotRestorePlanError("trusted root public key must contain 32 raw bytes") + actual = hashlib.sha256(raw).hexdigest() + if actual != TRUSTED_ROOT_PUBLIC_KEY_SHA256: + raise SnapshotRestorePlanError( + "trusted root public key does not match the frozen trust anchor" + ) + try: + return Ed25519PublicKey.from_public_bytes(raw), actual + except ValueError as exc: + raise SnapshotRestorePlanError("trusted root public key is invalid") from exc + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--package", type=Path, required=True) + parser.add_argument("--expected-package-sha256", required=True) + parser.add_argument("--evidence", type=Path, required=True) + parser.add_argument("--keyring", type=Path, required=True) + parser.add_argument("--trusted-root-public-key", type=Path, required=True) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + package_binding = _validate_package(args.package, args.expected_package_sha256) + evidence = load_json(args.evidence, label="first-release recovery evidence") + keyring = load_json(args.keyring, label="recovery keyring") + root_key, root_sha = _read_root_public_key(args.trusted_root_public_key) + now = datetime.now(UTC) + validated = validate_evidence( + evidence, + package_binding=package_binding, + keyring=keyring, + trusted_root_public_key=root_key, + now=now, + ) + plan = build_plan( + package_binding=package_binding, + validated_evidence=validated, + trusted_root_public_key_sha256=root_sha, + generated_at=now, + ) + except (OSError, SnapshotRestorePlanError) as exc: + print( + canonical_json_bytes( + {"schema_version": PLAN_SCHEMA, "status": "rejected", "error": str(exc)} + ).decode("utf-8"), + file=sys.stderr, + ) + return 2 + print(canonical_json_bytes(plan).decode("utf-8")) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/yuqei-infra/scripts/build-infrastructure-image-bundle.py b/yuqei-infra/scripts/build-infrastructure-image-bundle.py new file mode 100644 index 00000000..4691ecbc --- /dev/null +++ b/yuqei-infra/scripts/build-infrastructure-image-bundle.py @@ -0,0 +1,754 @@ +#!/usr/bin/env python3 +"""Build a deterministic, offline-verifiable infrastructure image bundle manifest. + +The tool is deliberately a pure file validator. It never invokes Docker and +never contacts a registry. Registry manifest bytes and Docker archives must +already be present below ``--artifact-root`` and are verified before a bundle +manifest is emitted. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import stat +import sys +import tarfile +from pathlib import Path, PurePosixPath +from typing import Any, Mapping + + +SOURCE_SET_SCHEMA = "yuqei.infrastructure-image-source-set.v1" +BUNDLE_SCHEMA = "yuqei.infrastructure-image-bundle.v1" +APPLICATION_MANIFEST_SCHEMA = "yuqei.installable-release-manifest.v1" +IMPORT_PLAN_SCHEMA = "yuqei.infrastructure-docker-image-import-plan.v1" +TARGET = {"os": "linux", "architecture": "amd64", "platform": "linux/amd64"} +DOCKER_CLI = "/usr/bin/docker" +DOCKER_HOST = "unix:///var/run/docker.sock" +MAX_JSON_BYTES = 8 * 1024 * 1024 +MAX_ARCHIVE_MEMBERS = 100_000 + +SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +REFERENCE_RE = re.compile( + r"^[a-z0-9](?:[a-z0-9._:/-]{0,510}[a-z0-9])?@sha256:[0-9a-f]{64}$" +) +TAG_RE = re.compile(r"^[a-z0-9](?:[a-z0-9._/-]{0,510}[a-z0-9])?:[^@\s]{1,128}$") + +APPLICATION_SPECS = ( + ("contract-web", "CONTRACT_WEB_IMAGE"), + ("contract-api", "CONTRACT_API_IMAGE"), + ("ai-platform-api", "AI_PLATFORM_API_IMAGE"), + ("ai-worker", "AI_WORKER_IMAGE"), +) +INFRASTRUCTURE_SPECS = ( + ("postgres-runtime", "POSTGRES_IMAGE", ("postgres",)), + ("postgres-client", "POSTGRES_CLIENT_IMAGE", ("postgres-init-databases",)), + ("redis", "REDIS_IMAGE", ("redis", "ai-redis")), + ("minio", "MINIO_IMAGE", ("minio",)), + ("minio-client", "MINIO_MC_IMAGE", ("minio-backup",)), + ("nginx-edge", "NGINX_IMAGE", ("edge",)), + ("backup-utility", "BACKUP_UTILITY_IMAGE", ("contract-files-backup",)), +) +ALL_IMAGE_KEYS = tuple(key for _role, key in APPLICATION_SPECS) + tuple( + key for _role, key, _services in INFRASTRUCTURE_SPECS +) + + +class BundleValidationError(RuntimeError): + """Raised when a staged image artifact is incomplete or ambiguous.""" + + +def canonical_json_bytes(value: object) -> bytes: + try: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise BundleValidationError("value cannot be encoded as canonical JSON") from exc + + +def canonical_sha256(value: object) -> str: + return hashlib.sha256(canonical_json_bytes(value)).hexdigest() + + +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _reject_constant(value: str) -> None: + raise BundleValidationError(f"non-finite JSON value is forbidden: {value}") + + +def _reject_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise BundleValidationError(f"duplicate JSON key is forbidden: {key}") + result[key] = value + return result + + +def load_json(path: Path, *, label: str) -> dict[str, Any]: + target = _regular_file(path, label=label, max_bytes=MAX_JSON_BYTES) + try: + payload = json.loads( + target.read_text(encoding="utf-8-sig"), + object_pairs_hook=_reject_duplicate_pairs, + parse_constant=_reject_constant, + ) + except BundleValidationError: + raise + except (OSError, UnicodeError, ValueError) as exc: + raise BundleValidationError(f"{label} is not strict UTF-8 JSON") from exc + if not isinstance(payload, dict): + raise BundleValidationError(f"{label} must be a JSON object") + return payload + + +def _absolute_lexical(path: Path) -> Path: + return Path(os.path.abspath(os.fspath(path))) + + +def _reject_symlink_components(path: Path, *, label: str) -> None: + absolute = _absolute_lexical(path) + current = Path(absolute.anchor) + for part in absolute.parts[1:]: + current /= part + try: + if current.exists() and current.is_symlink(): + raise BundleValidationError(f"{label} contains a symbolic link: {current}") + except OSError as exc: + raise BundleValidationError(f"cannot inspect {label} path") from exc + + +def _regular_file(path: Path, *, label: str, max_bytes: int | None = None) -> Path: + lexical = _absolute_lexical(path) + _reject_symlink_components(lexical, label=label) + try: + resolved = lexical.resolve(strict=True) + metadata = resolved.stat() + except OSError as exc: + raise BundleValidationError(f"{label} is missing") from exc + if lexical.is_symlink() or not stat.S_ISREG(metadata.st_mode): + raise BundleValidationError(f"{label} must be a regular file") + if metadata.st_nlink != 1: + raise BundleValidationError(f"{label} must not be hard linked") + if max_bytes is not None and (metadata.st_size <= 0 or metadata.st_size > max_bytes): + raise BundleValidationError(f"{label} has an invalid size") + return resolved + + +def _artifact_root(path: Path) -> Path: + lexical = _absolute_lexical(path) + _reject_symlink_components(lexical, label="artifact root") + try: + resolved = lexical.resolve(strict=True) + except OSError as exc: + raise BundleValidationError("artifact root is missing") from exc + if lexical.is_symlink() or not resolved.is_dir(): + raise BundleValidationError("artifact root must be a real directory") + return resolved + + +def _relative_path(value: object, *, label: str) -> str: + if not isinstance(value, str) or not value or "\\" in value or "\x00" in value: + raise BundleValidationError(f"{label} must be a canonical POSIX relative path") + pure = PurePosixPath(value) + if pure.is_absolute() or pure.as_posix() != value or any( + part in {"", ".", ".."} for part in pure.parts + ): + raise BundleValidationError(f"{label} is unsafe") + return value + + +def _artifact_file(root: Path, value: object, *, label: str) -> tuple[Path, str]: + relative = _relative_path(value, label=label) + lexical = root.joinpath(*PurePosixPath(relative).parts) + target = _regular_file(lexical, label=label) + try: + target.relative_to(root) + except ValueError as exc: + raise BundleValidationError(f"{label} escapes the artifact root") from exc + return target, relative + + +def _digest(value: object, *, label: str) -> str: + if not isinstance(value, str) or not DIGEST_RE.fullmatch(value): + raise BundleValidationError(f"{label} must be a sha256 digest") + return value + + +def _reference(value: object, *, label: str) -> str: + if not isinstance(value, str) or not REFERENCE_RE.fullmatch(value): + raise BundleValidationError(f"{label} must be an immutable image reference") + return value + + +def _tag(value: object, *, label: str) -> str: + if not isinstance(value, str) or not TAG_RE.fullmatch(value): + raise BundleValidationError(f"{label} must contain an explicit image tag") + tag = value.rsplit(":", 1)[1].casefold() + if tag == "latest": + raise BundleValidationError(f"{label} must not use latest") + return value + + +def _embedded_hash(payload: Mapping[str, Any], field: str, *, label: str) -> None: + expected = payload.get(field) + if not isinstance(expected, str) or not SHA256_RE.fullmatch(expected): + raise BundleValidationError(f"{label} has no valid {field}") + unsigned = {key: value for key, value in payload.items() if key != field} + if canonical_sha256(unsigned) != expected: + raise BundleValidationError(f"{label} payload hash mismatch") + + +def _file_record(root: Path, value: object, *, label: str) -> tuple[Path, dict[str, Any]]: + if not isinstance(value, dict) or set(value) != {"path", "sha256", "size_bytes"}: + raise BundleValidationError(f"{label} record fields are missing or unknown") + target, relative = _artifact_file(root, value.get("path"), label=label) + size = value.get("size_bytes") + digest = value.get("sha256") + if not isinstance(size, int) or isinstance(size, bool) or size <= 0: + raise BundleValidationError(f"{label} size is invalid") + if not isinstance(digest, str) or not SHA256_RE.fullmatch(digest): + raise BundleValidationError(f"{label} SHA-256 is invalid") + if target.stat().st_size != size or file_sha256(target) != digest: + raise BundleValidationError(f"{label} size or SHA-256 mismatch") + return target, {"path": relative, "sha256": digest, "size_bytes": size} + + +def _safe_tar_name(value: str, *, label: str) -> str: + if "\\" in value or "\x00" in value: + raise BundleValidationError(f"{label} contains an unsafe member path") + pure = PurePosixPath(value) + if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts): + raise BundleValidationError(f"{label} contains an unsafe member path") + return pure.as_posix() + + +def inspect_docker_archive( + path: Path, + *, + expected_repo_tag: str, + expected_config_digest: str, +) -> dict[str, Any]: + """Inspect a Docker save archive without extracting it or invoking Docker.""" + + try: + with tarfile.open(path, mode="r:*") as archive: + members = archive.getmembers() + if not members or len(members) > MAX_ARCHIVE_MEMBERS: + raise BundleValidationError("Docker archive has an invalid member count") + by_name: dict[str, tarfile.TarInfo] = {} + folded: set[str] = set() + for member in members: + name = _safe_tar_name(member.name, label="Docker archive") + if name in by_name or name.casefold() in folded: + raise BundleValidationError("Docker archive has duplicate member paths") + folded.add(name.casefold()) + if not (member.isfile() or member.isdir()): + raise BundleValidationError("Docker archive contains a link or special entry") + by_name[name] = member + manifest_member = by_name.get("manifest.json") + if manifest_member is None or not manifest_member.isfile(): + raise BundleValidationError("Docker archive has no manifest.json") + stream = archive.extractfile(manifest_member) + if stream is None: + raise BundleValidationError("Docker archive manifest cannot be read") + manifest_raw = stream.read(MAX_JSON_BYTES + 1) + if len(manifest_raw) > MAX_JSON_BYTES: + raise BundleValidationError("Docker archive manifest is too large") + try: + manifest = json.loads( + manifest_raw.decode("utf-8"), + object_pairs_hook=_reject_duplicate_pairs, + parse_constant=_reject_constant, + ) + except BundleValidationError: + raise + except (UnicodeError, ValueError) as exc: + raise BundleValidationError("Docker archive manifest is invalid") from exc + if not isinstance(manifest, list) or len(manifest) != 1 or not isinstance(manifest[0], dict): + raise BundleValidationError("Docker archive must contain exactly one image") + item = manifest[0] + if set(item) != {"Config", "RepoTags", "Layers"}: + raise BundleValidationError("Docker archive manifest fields are invalid") + repo_tags = item.get("RepoTags") + if repo_tags != [expected_repo_tag]: + raise BundleValidationError("Docker archive repository tag is not exact") + config_name = _safe_tar_name(item.get("Config"), label="Docker config") + config_member = by_name.get(config_name) + if config_member is None or not config_member.isfile(): + raise BundleValidationError("Docker archive config is missing") + config_stream = archive.extractfile(config_member) + if config_stream is None: + raise BundleValidationError("Docker archive config cannot be read") + config_raw = config_stream.read(MAX_JSON_BYTES + 1) + if len(config_raw) > MAX_JSON_BYTES: + raise BundleValidationError("Docker archive config is too large") + observed_config_digest = "sha256:" + hashlib.sha256(config_raw).hexdigest() + if observed_config_digest != expected_config_digest: + raise BundleValidationError("Docker archive config digest mismatch") + if config_name != expected_config_digest.removeprefix("sha256:") + ".json": + raise BundleValidationError("Docker archive config filename is not content addressed") + try: + config = json.loads( + config_raw.decode("utf-8"), + object_pairs_hook=_reject_duplicate_pairs, + parse_constant=_reject_constant, + ) + except BundleValidationError: + raise + except (UnicodeError, ValueError) as exc: + raise BundleValidationError("Docker archive config JSON is invalid") from exc + if not isinstance(config, dict) or config.get("os") != "linux" or config.get("architecture") != "amd64": + raise BundleValidationError("Docker archive platform must be linux/amd64") + layers = item.get("Layers") + if not isinstance(layers, list) or not layers or any(not isinstance(layer, str) for layer in layers): + raise BundleValidationError("Docker archive layer inventory is invalid") + normalized_layers = [_safe_tar_name(layer, label="Docker layer") for layer in layers] + if len(normalized_layers) != len(set(normalized_layers)): + raise BundleValidationError("Docker archive repeats a layer path") + if any(name not in by_name or not by_name[name].isfile() for name in normalized_layers): + raise BundleValidationError("Docker archive is missing a declared layer") + except (tarfile.TarError, OSError) as exc: + raise BundleValidationError("Docker archive cannot be inspected") from exc + return { + "config_digest": observed_config_digest, + "expected_repo_tag": expected_repo_tag, + "platform": TARGET["platform"], + "layer_count": len(normalized_layers), + } + + +def _validate_registry_manifest( + path: Path, + *, + expected_manifest_digest: str, + expected_config_digest: str, +) -> dict[str, Any]: + raw = path.read_bytes() + if "sha256:" + hashlib.sha256(raw).hexdigest() != expected_manifest_digest: + raise BundleValidationError("registry manifest digest mismatch") + try: + payload = json.loads( + raw.decode("utf-8"), + object_pairs_hook=_reject_duplicate_pairs, + parse_constant=_reject_constant, + ) + except BundleValidationError: + raise + except (UnicodeError, ValueError) as exc: + raise BundleValidationError("registry manifest is invalid JSON") from exc + if not isinstance(payload, dict): + raise BundleValidationError("registry manifest must be an object") + media_type = payload.get("mediaType") + if media_type not in { + "application/vnd.docker.distribution.manifest.v2+json", + "application/vnd.oci.image.manifest.v1+json", + }: + raise BundleValidationError("registry artifact is not a resolved image manifest") + config = payload.get("config") + if not isinstance(config, dict) or config.get("digest") != expected_config_digest: + raise BundleValidationError("registry manifest config digest mismatch") + layers = payload.get("layers") + if not isinstance(layers, list) or not layers: + raise BundleValidationError("registry manifest layer inventory is missing") + for layer in layers: + if not isinstance(layer, dict) or not DIGEST_RE.fullmatch(str(layer.get("digest", ""))): + raise BundleValidationError("registry manifest has an invalid layer digest") + return {"media_type": media_type, "layer_count": len(layers)} + + +def _validate_sbom( + path: Path, + *, + role: str, + expected_manifest_digest: str, +) -> dict[str, Any]: + payload = load_json(path, label=f"{role} SBOM") + if payload.get("spdxVersion") != "SPDX-2.3" or payload.get("SPDXID") != "SPDXRef-DOCUMENT": + raise BundleValidationError(f"{role} SBOM must be an SPDX-2.3 document") + described = payload.get("documentDescribes") + packages = payload.get("packages") + if not isinstance(described, list) or len(described) != 1 or not isinstance(packages, list) or not packages: + raise BundleValidationError(f"{role} SBOM must identify exactly one image root package") + package_ids: set[str] = set() + root_packages: list[Mapping[str, Any]] = [] + for package in packages: + if not isinstance(package, dict): + raise BundleValidationError(f"{role} SBOM package inventory is invalid") + spdx_id = package.get("SPDXID") + if not isinstance(spdx_id, str) or not spdx_id or spdx_id in package_ids: + raise BundleValidationError(f"{role} SBOM package identity is invalid") + package_ids.add(spdx_id) + if spdx_id == described[0]: + root_packages.append(package) + if len(root_packages) != 1 or root_packages[0].get("versionInfo") != expected_manifest_digest: + raise BundleValidationError(f"{role} SBOM root does not bind the image manifest digest") + creation = payload.get("creationInfo") + creators = creation.get("creators") if isinstance(creation, dict) else None + if not isinstance(creators, list) or not any("syft" in str(item).casefold() for item in creators): + raise BundleValidationError(f"{role} SBOM generator identity is missing") + return { + "spdx_version": "SPDX-2.3", + "package_count": len(packages), + "root_package_spdx_id": described[0], + "root_manifest_digest": expected_manifest_digest, + } + + +def _validate_trivy_scan( + path: Path, + *, + role: str, + scan_kind: str, + expected_config_digest: str, +) -> dict[str, Any]: + payload = load_json(path, label=f"{role} {scan_kind}") + if payload.get("SchemaVersion") != 2 or payload.get("ArtifactType") != "container_image": + raise BundleValidationError(f"{role} {scan_kind} must be a Trivy container-image report") + metadata = payload.get("Metadata") + if not isinstance(metadata, dict) or metadata.get("ImageID") != expected_config_digest: + raise BundleValidationError(f"{role} {scan_kind} config image ID mismatch") + results = payload.get("Results") + if not isinstance(results, list) or not results: + raise BundleValidationError(f"{role} {scan_kind} result inventory is missing") + finding_key = "Vulnerabilities" if scan_kind == "vulnerability scan" else "Licenses" + finding_count = 0 + for result in results: + if not isinstance(result, dict): + raise BundleValidationError(f"{role} {scan_kind} result is invalid") + findings = result.get(finding_key, []) + if findings is None: + findings = [] + if not isinstance(findings, list) or any(not isinstance(item, dict) for item in findings): + raise BundleValidationError(f"{role} {scan_kind} findings are invalid") + finding_count += len(findings) + return { + "scanner": "trivy", + "config_digest": expected_config_digest, + "result_count": len(results), + "finding_count": finding_count, + } + + +def _validate_application_manifest(root: Path, payload: Mapping[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]: + if payload.get("schema_version") != APPLICATION_MANIFEST_SCHEMA: + raise BundleValidationError("unsupported application installable manifest schema") + _embedded_hash(payload, "manifest_payload_sha256", label="application manifest") + candidate = payload.get("candidate") + if not isinstance(candidate, dict): + raise BundleValidationError("application manifest candidate is missing") + images = payload.get("images") + if not isinstance(images, list) or len(images) != len(APPLICATION_SPECS): + raise BundleValidationError("application manifest must contain four images") + overrides: dict[str, str] = {} + normalized: list[dict[str, Any]] = [] + for raw, (role, environment_key) in zip(images, APPLICATION_SPECS, strict=True): + if not isinstance(raw, dict) or raw.get("role") != role or raw.get("environment_key") != environment_key: + raise BundleValidationError("application image role or order drifted") + if raw.get("platform") != TARGET["platform"]: + raise BundleValidationError("application image platform must be linux/amd64") + config_digest = _digest(raw.get("config_digest"), label=f"{role} config digest") + manifest_digest = _digest(raw.get("manifest_digest"), label=f"{role} manifest digest") + archive_path, archive_record = _file_record( + root, raw.get("docker_archive"), label=f"{role} Docker archive" + ) + expected_tag = _tag(raw.get("image_ref"), label=f"{role} image tag") + inspected = inspect_docker_archive( + archive_path, + expected_repo_tag=expected_tag, + expected_config_digest=config_digest, + ) + overrides[environment_key] = config_digest + normalized.append( + { + "role": role, + "environment_key": environment_key, + "manifest_digest": manifest_digest, + "config_digest": config_digest, + "docker_archive": archive_record, + "expected_repo_tag": expected_tag, + "archive_inspection": inspected, + } + ) + return dict(candidate), normalized + + +def build_bundle( + *, + artifact_root: Path, + source_set: Mapping[str, Any], + application_manifest: Mapping[str, Any], +) -> dict[str, Any]: + root = _artifact_root(artifact_root) + if source_set.get("schema_version") != SOURCE_SET_SCHEMA: + raise BundleValidationError("unsupported infrastructure source-set schema") + _embedded_hash(source_set, "source_set_payload_sha256", label="infrastructure source set") + if source_set.get("target") != TARGET: + raise BundleValidationError("infrastructure source target must be linux/amd64") + source_candidate = source_set.get("candidate") + if not isinstance(source_candidate, dict) or set(source_candidate) != {"commit", "tag"}: + raise BundleValidationError("infrastructure source candidate is invalid") + app_candidate, application_images = _validate_application_manifest( + root, application_manifest + ) + if app_candidate.get("commit") != source_candidate.get("commit") or app_candidate.get("tag") != source_candidate.get("tag"): + raise BundleValidationError("application and infrastructure candidate bindings differ") + + raw_images = source_set.get("images") + if not isinstance(raw_images, list) or len(raw_images) != len(INFRASTRUCTURE_SPECS): + raise BundleValidationError("infrastructure source set must contain exactly seven images") + seen_archives: set[str] = set() + seen_configs: set[str] = set() + seen_security_artifacts: set[str] = set() + infrastructure_images: list[dict[str, Any]] = [] + import_images: list[dict[str, Any]] = [] + overrides = {item["environment_key"]: item["config_digest"] for item in application_images} + for sequence, (raw, (role, environment_key, services)) in enumerate( + zip(raw_images, INFRASTRUCTURE_SPECS, strict=True), start=1 + ): + if not isinstance(raw, dict): + raise BundleValidationError("infrastructure image entries must be objects") + expected_fields = { + "role", + "environment_key", + "services", + "canonical_source_tag", + "canonical_reference", + "transport_reference", + "manifest_digest", + "config_digest", + "registry_manifest", + "docker_archive", + "expected_repo_tag", + "sbom", + "vulnerability_scan", + "license_scan", + } + if set(raw) != expected_fields: + raise BundleValidationError(f"{role} fields are missing or unknown") + if raw.get("role") != role or raw.get("environment_key") != environment_key or raw.get("services") != list(services): + raise BundleValidationError("infrastructure image role, service mapping, or order drifted") + source_tag = _tag(raw.get("canonical_source_tag"), label=f"{role} source tag") + expected_tag = _tag(raw.get("expected_repo_tag"), label=f"{role} offline tag") + manifest_digest = _digest(raw.get("manifest_digest"), label=f"{role} manifest digest") + config_digest = _digest(raw.get("config_digest"), label=f"{role} config digest") + canonical_reference = _reference(raw.get("canonical_reference"), label=f"{role} canonical reference") + transport_reference = _reference(raw.get("transport_reference"), label=f"{role} transport reference") + if canonical_reference.rsplit("@", 1)[1] != manifest_digest or transport_reference.rsplit("@", 1)[1] != manifest_digest: + raise BundleValidationError(f"{role} references do not match the resolved manifest digest") + if canonical_reference.split("@", 1)[0] != source_tag.rsplit(":", 1)[0]: + raise BundleValidationError(f"{role} canonical source repository drifted") + manifest_path, manifest_record = _file_record( + root, raw.get("registry_manifest"), label=f"{role} registry manifest" + ) + registry = _validate_registry_manifest( + manifest_path, + expected_manifest_digest=manifest_digest, + expected_config_digest=config_digest, + ) + archive_path, archive_record = _file_record( + root, raw.get("docker_archive"), label=f"{role} Docker archive" + ) + if archive_record["path"] in seen_archives or config_digest in seen_configs: + raise BundleValidationError("infrastructure archives and config digests must be unique") + seen_archives.add(archive_record["path"]) + seen_configs.add(config_digest) + inspected = inspect_docker_archive( + archive_path, + expected_repo_tag=expected_tag, + expected_config_digest=config_digest, + ) + sbom_path, sbom_record = _file_record( + root, raw.get("sbom"), label=f"{role} SBOM" + ) + vulnerability_path, vulnerability_record = _file_record( + root, + raw.get("vulnerability_scan"), + label=f"{role} vulnerability scan", + ) + license_path, license_record = _file_record( + root, raw.get("license_scan"), label=f"{role} license scan" + ) + security_paths = { + sbom_record["path"], + vulnerability_record["path"], + license_record["path"], + } + if len(security_paths) != 3 or seen_security_artifacts.intersection(security_paths): + raise BundleValidationError("infrastructure security artifact paths must be unique") + seen_security_artifacts.update(security_paths) + sbom_summary = _validate_sbom( + sbom_path, + role=role, + expected_manifest_digest=manifest_digest, + ) + vulnerability_summary = _validate_trivy_scan( + vulnerability_path, + role=role, + scan_kind="vulnerability scan", + expected_config_digest=config_digest, + ) + license_summary = _validate_trivy_scan( + license_path, + role=role, + scan_kind="license scan", + expected_config_digest=config_digest, + ) + image = { + "sequence": sequence, + "role": role, + "environment_key": environment_key, + "services": list(services), + "platform": TARGET["platform"], + "canonical_source_tag": source_tag, + "canonical_reference": canonical_reference, + "transport_reference": transport_reference, + "manifest_digest": manifest_digest, + "config_digest": config_digest, + "registry_manifest": manifest_record, + "docker_archive": archive_record, + "expected_repo_tag": expected_tag, + "archive_inspection": inspected, + "sbom": sbom_record, + "sbom_summary": sbom_summary, + "vulnerability_scan": vulnerability_record, + "vulnerability_scan_summary": vulnerability_summary, + "license_scan": license_record, + "license_scan_summary": license_summary, + } + infrastructure_images.append(image) + import_images.append( + { + "sequence": sequence, + "role": role, + "environment_key": environment_key, + "archive": archive_record, + "load_command_argv": [ + DOCKER_CLI, + "--host", + DOCKER_HOST, + "image", + "load", + "--input", + archive_record["path"], + ], + "tag_command_argv": [ + DOCKER_CLI, + "--host", + DOCKER_HOST, + "image", + "tag", + config_digest, + expected_tag, + ], + "inspect_command_argv": [ + DOCKER_CLI, + "--host", + DOCKER_HOST, + "image", + "inspect", + "--format", + "{{json .}}", + expected_tag, + ], + "expected_loaded_image_id": config_digest, + "expected_manifest_digest": manifest_digest, + "expected_platform": TARGET["platform"], + "offline_compose_override": config_digest, + } + ) + overrides[environment_key] = config_digest + + if tuple(overrides) != ALL_IMAGE_KEYS or set(overrides) != set(ALL_IMAGE_KEYS): + raise BundleValidationError("offline override map does not cover exactly all eleven images") + unsigned: dict[str, Any] = { + "schema_version": BUNDLE_SCHEMA, + "candidate": {"commit": source_candidate["commit"], "tag": source_candidate["tag"]}, + "target": TARGET, + "application_images": application_images, + "infrastructure_images": infrastructure_images, + "offline_compose_overrides": overrides, + "all_release_image_keys": list(ALL_IMAGE_KEYS), + "docker_import_plan": { + "schema_version": IMPORT_PLAN_SCHEMA, + "implicit_preloaded_images_allowed": False, + "network_access_required": False, + "docker_required_for_validation": False, + "images": import_images, + }, + "claims_not_made": [ + "This file-only validator did not contact a registry or invoke Docker.", + "This bundle does not prove installation, snapshot capture, restore, UAT, production, or public-release approval.", + ], + } + unsigned["bundle_payload_sha256"] = canonical_sha256(unsigned) + return unsigned + + +def _write_exclusive(path: Path, payload: Mapping[str, Any]) -> None: + target = _absolute_lexical(path) + if target.exists() or target.is_symlink(): + raise BundleValidationError("refusing to overwrite bundle output") + target.parent.mkdir(parents=True, exist_ok=True) + data = canonical_json_bytes(payload) + b"\n" + descriptor = os.open(target, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + try: + with os.fdopen(descriptor, "wb") as stream: + descriptor = -1 + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + finally: + if descriptor >= 0: + os.close(descriptor) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--artifact-root", type=Path, required=True) + parser.add_argument("--source-set", type=Path, required=True) + parser.add_argument("--application-manifest", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + payload = build_bundle( + artifact_root=args.artifact_root, + source_set=load_json(args.source_set, label="infrastructure source set"), + application_manifest=load_json( + args.application_manifest, label="application installable manifest" + ), + ) + _write_exclusive(args.output, payload) + except (BundleValidationError, OSError) as exc: + print( + canonical_json_bytes( + {"schema_version": BUNDLE_SCHEMA, "status": "rejected", "error": str(exc)} + ).decode("utf-8"), + file=sys.stderr, + ) + return 2 + print(canonical_json_bytes({"status": "verified", "output": str(args.output)}).decode("utf-8")) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/yuqei-infra/tests/test_first_release_snapshot_restore_planner.py b/yuqei-infra/tests/test_first_release_snapshot_restore_planner.py new file mode 100644 index 00000000..84453e09 --- /dev/null +++ b/yuqei-infra/tests/test_first_release_snapshot_restore_planner.py @@ -0,0 +1,570 @@ +from __future__ import annotations + +import base64 +import hashlib +import importlib.util +import json +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + +MODULE_PATH = ( + Path(__file__).resolve().parents[1] + / "deploy" + / "plan-first-release-snapshot-restore.py" +) +SPEC = importlib.util.spec_from_file_location( + "first_release_snapshot_restore_planner", MODULE_PATH +) +assert SPEC is not None and SPEC.loader is not None +planner = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = planner +SPEC.loader.exec_module(planner) + + +NOW = datetime(2026, 9, 27, 1, 0, tzinfo=UTC) + + +def _sha(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _write_json(path: Path, payload: object) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(planner.canonical_json_bytes(payload)) + return path + + +def _embedded(payload: dict[str, Any], field: str) -> dict[str, Any]: + result = dict(payload) + result[field] = planner.canonical_sha256(result) + return result + + +def _external(path: Path) -> dict[str, str]: + return {"path": str(path.resolve()), "sha256": _sha(path)} + + +def _raw_public(private_key: Ed25519PrivateKey) -> bytes: + return private_key.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + + +def _approval( + *, + approval_id: str, + approver_id: str, + role: str, + key_id: str, + private_key: Ed25519PrivateKey, + evidence_hash: str, +) -> dict[str, Any]: + unsigned = { + "schema_version": planner.APPROVAL_SCHEMA, + "approval_id": approval_id, + "approver_id": approver_id, + "role": role, + "key_id": key_id, + "signed_at": "2026-09-27T00:40:00Z", + "expires_at": "2026-09-27T06:00:00Z", + "evidence_payload_sha256": evidence_hash, + } + return { + **unsigned, + "signature_base64": base64.b64encode( + private_key.sign(planner.canonical_json_bytes(unsigned)) + ).decode("ascii"), + } + + +def _case( + tmp_path: Path, + *, + recovery_storage_targets: list[str] | None = None, + backup_databases: list[str] | None = None, + snapshot_created_at: str = "2026-09-27T00:00:00Z", + database_downgrade_performed: bool = False, + successful_release_count: int = 0, + same_approver_person: bool = False, + restore_bindings_match: bool = True, +) -> dict[str, Any]: + release_id = "release-first-1" + environment_id = "production-cn-1" + host_id = "contract-host-1" + consistency_set_id = "consistency-set-1" + source_commit = "1" * 40 + package_root = tmp_path / "package" + + recovery = { + "strategy": planner.RECOVERY_STRATEGY, + "evidence_status": planner.RECOVERY_EVIDENCE_STATUS, + "first_release_only": True, + "predecessor_package_required": False, + "database_downgrade_allowed": False, + "required_restore_scopes": list(planner.REQUIRED_COMPONENTS), + "required_postgres_databases": list(planner.REQUIRED_DATABASES), + "required_storage_targets": ( + list(planner.REQUIRED_STORAGE_TARGETS) + if recovery_storage_targets is None + else recovery_storage_targets + ), + } + identity = _embedded( + { + "schema_version": planner.IMAGE_IDENTITY_SCHEMA, + "release_id": release_id, + "source_commit": source_commit, + "application_images": {}, + "infrastructure_images": {}, + "recovery": recovery, + }, + "identity_payload_sha256", + ) + identity_path = _write_json(package_root / "image-identity.json", identity) + package = _embedded( + { + "schema_version": "yuqei.release-package.v1", + "release_id": release_id, + "compose_project_name": "yuqei-contract", + "artifacts": { + "image_identity": { + "path": "image-identity.json", + "sha256": _sha(identity_path), + } + }, + }, + "package_payload_sha256", + ) + package_path = _write_json(package_root / "release-package.json", package) + + evidence_root = tmp_path / "evidence" + no_prior = _embedded( + { + "schema_version": planner.NO_PRIOR_RELEASE_SCHEMA, + "environment_id": environment_id, + "candidate_release_id": release_id, + "successful_release_count": successful_release_count, + "status": "confirmed_no_prior_successful_release", + "observed_at": "2026-09-27T00:10:00Z", + }, + "receipt_payload_sha256", + ) + no_prior_path = _write_json(evidence_root / "no-prior-release.json", no_prior) + + snapshot = _embedded( + { + "schema_version": planner.SNAPSHOT_RECEIPT_SCHEMA, + "provider": "approved-snapshot-provider", + "control_plane_id": "independent-control-plane-1", + "snapshot_id": "snapshot-immutable-1", + "source_host_id": host_id, + "consistency_set_id": consistency_set_id, + "off_host": True, + "immutable": True, + "status": "available", + "created_at": snapshot_created_at, + }, + "receipt_payload_sha256", + ) + snapshot_path = _write_json(evidence_root / "host-snapshot.json", snapshot) + + database_names = ( + list(planner.REQUIRED_DATABASES) + if backup_databases is None + else backup_databases + ) + database_records = { + name: { + "artifact_sha256": hashlib.sha256(name.encode("utf-8")).hexdigest(), + "size_bytes": 1024, + "schema_revision": f"{name}-schema-1", + } + for name in database_names + } + backup = _embedded( + { + "schema_version": planner.BACKUP_MANIFEST_SCHEMA, + "backup_id": "backup-consistent-1", + "release_id": release_id, + "environment_id": environment_id, + "consistency_set_id": consistency_set_id, + "consistency_mode": "quiesced", + "databases": database_records, + "contract_files": { + "archive_sha256": "a" * 64, + "size_bytes": 2048, + "file_count": 2, + }, + "minio": { + "content_index_sha256": "b" * 64, + "object_count": 3, + "total_size_bytes": 4096, + }, + "completed_at": "2026-09-27T00:05:00Z", + "status": "complete", + }, + "manifest_payload_sha256", + ) + backup_path = _write_json(evidence_root / "consistent-backup.json", backup) + + release_bindings = { + "production_env_sha256": "d" * 64, + "tls_certificate_sha256": "e" * 64, + "tls_private_key_sha256": "f" * 64, + "tls_trusted_ca_sha256": "0" * 64, + } + + restore_test = _embedded( + { + "schema_version": planner.RESTORE_TEST_SCHEMA, + "release_id": release_id, + "source_environment_id": environment_id, + "restore_environment_id": "isolated-uat-restore-1", + "restore_environment_class": "isolated_uat", + "consistency_set_id": consistency_set_id, + "snapshot_receipt_sha256": _sha(snapshot_path), + "backup_manifest_sha256": _sha(backup_path), + "release_environment_and_tls_bindings_sha256": planner.canonical_sha256( + release_bindings + ) if restore_bindings_match else "9" * 64, + "host_snapshot_restored": True, + "databases_restored": list(planner.REQUIRED_DATABASES), + "contract_files_restored": True, + "minio_restored": True, + "environment_tls_bindings_verified": True, + "database_downgrade_performed": database_downgrade_performed, + "status": "passed", + "tested_at": "2026-09-27T00:20:00Z", + }, + "evidence_payload_sha256", + ) + restore_path = _write_json(evidence_root / "independent-restore-test.json", restore_test) + + root_private = Ed25519PrivateKey.generate() + role_keys = { + "release_manager": Ed25519PrivateKey.generate(), + "database_owner": Ed25519PrivateKey.generate(), + } + approver_ids = { + "release_manager": "approver-release-manager", + "database_owner": ( + "approver-release-manager" + if same_approver_person + else "approver-database-owner" + ), + } + key_ids = { + "release_manager": "key-release-manager-1", + "database_owner": "key-database-owner-1", + } + key_records = [] + for role in ("release_manager", "database_owner"): + key_records.append( + { + "key_id": key_ids[role], + "approver_id": approver_ids[role], + "role": role, + "public_key_base64": base64.b64encode( + _raw_public(role_keys[role]) + ).decode("ascii"), + "not_before": "2026-09-26T00:00:00Z", + "not_after": "2026-10-01T00:00:00Z", + "revoked": False, + } + ) + keyring_unsigned = { + "schema_version": planner.KEYRING_SCHEMA, + "root_key_id": planner.TRUSTED_ROOT_KEY_ID, + "epoch": 3, + "issued_at": "2026-09-26T00:00:00Z", + "expires_at": "2026-10-01T00:00:00Z", + "keys": key_records, + } + keyring = { + **keyring_unsigned, + "keyring_payload_sha256": planner.canonical_sha256(keyring_unsigned), + "root_signature_base64": base64.b64encode( + root_private.sign(planner.canonical_json_bytes(keyring_unsigned)) + ).decode("ascii"), + } + keyring_path = _write_json(evidence_root / "keyring.json", keyring) + root_path = evidence_root / "root-public-key.raw" + root_path.write_bytes(_raw_public(root_private)) + + evidence_unsigned = { + "schema_version": planner.EVIDENCE_SCHEMA, + "evidence_id": "first-release-recovery-evidence-1", + "mode": "FIRST_RELEASE_PREINSTALL_SNAPSHOT_RESTORE", + "release_binding": { + "release_id": release_id, + "package_sha256": _sha(package_path), + "source_commit": source_commit, + "image_identity_sha256": _sha(identity_path), + }, + "target": { + "environment_id": environment_id, + "environment_class": "production", + "host_id": host_id, + "host_identity_sha256": "c" * 64, + "compose_project_name": "yuqei-contract", + }, + "consistency_set_id": consistency_set_id, + "captured_at": "2026-09-27T00:30:00Z", + "expires_at": "2026-09-27T12:00:00Z", + "no_prior_successful_release_receipt": _external(no_prior_path), + "components": { + "host_runtime_state": { + "provider_receipt": _external(snapshot_path), + "off_host": True, + "immutable": True, + "status": "available", + }, + "postgres_consistent_backup": { + "backup_manifest": _external(backup_path), + "consistency_mode": "quiesced", + "databases": database_names, + }, + "object_storage_state": { + "backup_manifest": _external(backup_path), + "stores": list(planner.REQUIRED_STORAGE_TARGETS), + }, + "release_environment_and_tls_bindings": release_bindings, + }, + "independent_restore_test": _external(restore_path), + } + evidence_hash = planner.canonical_sha256(evidence_unsigned) + approvals = [ + _approval( + approval_id=f"approval-{role}", + approver_id=approver_ids[role], + role=role, + key_id=key_ids[role], + private_key=role_keys[role], + evidence_hash=evidence_hash, + ) + for role in ("release_manager", "database_owner") + ] + evidence = { + **evidence_unsigned, + "evidence_payload_sha256": evidence_hash, + "approvals": approvals, + } + evidence_path = _write_json(evidence_root / "recovery-evidence.json", evidence) + return { + "package_path": package_path, + "package_sha256": _sha(package_path), + "evidence": evidence, + "evidence_path": evidence_path, + "keyring": keyring, + "keyring_path": keyring_path, + "root_path": root_path, + "root_sha256": _sha(root_path), + } + + +def _provision_test_pins( + monkeypatch: pytest.MonkeyPatch, + case: dict[str, Any], + *, + minimum_epoch: int = 3, +) -> None: + monkeypatch.setattr( + planner, "TRUSTED_ROOT_PUBLIC_KEY_SHA256", case["root_sha256"] + ) + monkeypatch.setattr( + planner, + "TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256", + case["keyring"]["keyring_payload_sha256"], + ) + monkeypatch.setattr(planner, "MIN_KEYRING_EPOCH", minimum_epoch) + + +def _validate( + case: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any], str]: + package_binding = planner._validate_package( + case["package_path"], case["package_sha256"] + ) + root_key, root_sha = planner._read_root_public_key(case["root_path"]) + validated = planner.validate_evidence( + case["evidence"], + package_binding=package_binding, + keyring=case["keyring"], + trusted_root_public_key=root_key, + now=NOW, + ) + return package_binding, validated, root_sha + + +def test_valid_first_release_evidence_emits_only_a_non_writing_plan( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path) + _provision_test_pins(monkeypatch, case) + + package_binding, validated, root_sha = _validate(case) + plan = planner.build_plan( + package_binding=package_binding, + validated_evidence=validated, + trusted_root_public_key_sha256=root_sha, + generated_at=NOW, + ) + + assert validated["backup"]["databases"] == ["contract", "ai", "operations"] + assert validated["backup"]["storage_targets"] == [ + "contract_files_volume", + "minio_object_store", + ] + assert set(item["role"] for item in validated["approvals"]) == { + "release_manager", + "database_owner", + } + assert len({item["approver_id"] for item in validated["approvals"]}) == 2 + assert len({item["key_id"] for item in validated["approvals"]}) == 2 + assert plan["status"] == "signature_verified_against_frozen_pins_plan_only" + assert plan["apply_allowed"] is False + assert plan["execution_entrypoint_available"] is False + assert plan["writes_performed"] is False + assert plan["docker_invoked"] is False + assert plan["remote_connection_attempted"] is False + assert plan["database_downgrade_allowed"] is False + assert plan["fake_predecessor_package_allowed"] is False + assert plan["plan_payload_sha256"] == planner.canonical_sha256( + {key: value for key, value in plan.items() if key != "plan_payload_sha256"} + ) + + +def test_trust_anchor_and_current_revocation_state_fail_closed_until_frozen( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path) + with pytest.raises(planner.SnapshotRestorePlanError, match="root key is not provisioned"): + planner._read_root_public_key(case["root_path"]) + + monkeypatch.setattr( + planner, "TRUSTED_ROOT_PUBLIC_KEY_SHA256", case["root_sha256"] + ) + root_key, _ = planner._read_root_public_key(case["root_path"]) + binding = planner._validate_package(case["package_path"], case["package_sha256"]) + with pytest.raises(planner.SnapshotRestorePlanError, match="keyring pin is not provisioned"): + planner.validate_evidence( + case["evidence"], + package_binding=binding, + keyring=case["keyring"], + trusted_root_public_key=root_key, + now=NOW, + ) + + +def test_old_or_unpinned_keyring_revocation_state_is_rejected( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path) + _provision_test_pins(monkeypatch, case, minimum_epoch=4) + with pytest.raises(planner.SnapshotRestorePlanError, match="older than the trusted minimum"): + _validate(case) + + monkeypatch.setattr(planner, "MIN_KEYRING_EPOCH", 3) + monkeypatch.setattr( + planner, "TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256", "9" * 64 + ) + with pytest.raises(planner.SnapshotRestorePlanError, match="currently trusted revocation state"): + _validate(case) + + +@pytest.mark.parametrize( + "targets", + ( + ["contract_files_volume"], + ["contract-files", "minio_object_store"], + ["contract_files_volume", "minio_bucket"], + ), +) +def test_package_must_bind_exact_approved_storage_targets( + tmp_path: Path, + targets: list[str], +) -> None: + case = _case(tmp_path, recovery_storage_targets=targets) + with pytest.raises(planner.SnapshotRestorePlanError, match="recovery policy is unsafe"): + planner._validate_package(case["package_path"], case["package_sha256"]) + + +def test_backup_must_cover_contract_ai_and_operations_databases( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path, backup_databases=["contract", "ai"]) + _provision_test_pins(monkeypatch, case) + with pytest.raises(planner.SnapshotRestorePlanError, match="contract/ai/operations"): + _validate(case) + + +def test_no_prior_successful_release_receipt_is_mandatory( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path, successful_release_count=1) + _provision_test_pins(monkeypatch, case) + with pytest.raises(planner.SnapshotRestorePlanError, match="does not prove a first release"): + _validate(case) + + +def test_snapshot_and_component_evidence_must_be_fresh( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path, snapshot_created_at="2026-09-25T00:00:00Z") + _provision_test_pins(monkeypatch, case) + with pytest.raises(planner.SnapshotRestorePlanError, match="component evidence is stale"): + _validate(case) + + +def test_restore_evidence_forbids_database_downgrade( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path, database_downgrade_performed=True) + _provision_test_pins(monkeypatch, case) + with pytest.raises(planner.SnapshotRestorePlanError, match="restore test is incomplete"): + _validate(case) + + +def test_restore_test_is_bound_to_the_exact_environment_and_tls_hashes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path, restore_bindings_match=False) + _provision_test_pins(monkeypatch, case) + with pytest.raises(planner.SnapshotRestorePlanError, match="restore test is incomplete"): + _validate(case) + + +def test_approvals_require_two_different_people_and_keys( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case(tmp_path, same_approver_person=True) + _provision_test_pins(monkeypatch, case) + with pytest.raises(planner.SnapshotRestorePlanError, match="two roles, people, and keys"): + _validate(case) + + +def test_cli_has_no_apply_or_caller_supplied_trust_pin() -> None: + options = { + option + for action in planner.build_parser()._actions + for option in action.option_strings + } + assert "--apply" not in options + assert "--expected-trusted-root-public-key-sha256" not in options diff --git a/yuqei-infra/tests/test_infrastructure_image_bundle.py b/yuqei-infra/tests/test_infrastructure_image_bundle.py new file mode 100644 index 00000000..ed1a0e07 --- /dev/null +++ b/yuqei-infra/tests/test_infrastructure_image_bundle.py @@ -0,0 +1,451 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import io +import json +import sys +import tarfile +from pathlib import Path + +import pytest + + +MODULE_PATH = ( + Path(__file__).resolve().parents[1] + / "scripts" + / "build-infrastructure-image-bundle.py" +) +SPEC = importlib.util.spec_from_file_location("build_infrastructure_image_bundle", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +bundle = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = bundle +SPEC.loader.exec_module(bundle) + + +def _sha(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _record(path: Path, root: Path) -> dict[str, object]: + return { + "path": path.relative_to(root).as_posix(), + "sha256": _sha(path), + "size_bytes": path.stat().st_size, + } + + +def _write_canonical(path: Path, payload: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(bundle.canonical_json_bytes(payload)) + + +def _with_hash(payload: dict[str, object], field: str) -> dict[str, object]: + result = dict(payload) + result[field] = bundle.canonical_sha256(result) + return result + + +def _make_image_artifacts( + root: Path, + *, + role: str, + repo_tag: str, + marker: str, +) -> tuple[str, str, dict[str, object], dict[str, object]]: + config = { + "architecture": "amd64", + "os": "linux", + "rootfs": {"type": "layers", "diff_ids": [f"sha256:{marker * 64}"]}, + } + config_raw = bundle.canonical_json_bytes(config) + config_digest = "sha256:" + hashlib.sha256(config_raw).hexdigest() + layer_raw = (role + "-layer").encode("utf-8") + registry_manifest = { + "schemaVersion": 2, + "mediaType": "application/vnd.oci.image.manifest.v1+json", + "config": { + "mediaType": "application/vnd.oci.image.config.v1+json", + "digest": config_digest, + "size": len(config_raw), + }, + "layers": [ + { + "mediaType": "application/vnd.oci.image.layer.v1.tar", + "digest": "sha256:" + hashlib.sha256(layer_raw).hexdigest(), + "size": len(layer_raw), + } + ], + } + manifest_path = root / "registry-manifests" / f"{role}.json" + _write_canonical(manifest_path, registry_manifest) + manifest_digest = "sha256:" + _sha(manifest_path) + + archive_path = root / "docker-archives" / f"{role}.docker.tar" + archive_path.parent.mkdir(parents=True, exist_ok=True) + config_name = config_digest.removeprefix("sha256:") + ".json" + layer_name = f"layers/{role}/layer.tar" + docker_manifest = [ + {"Config": config_name, "RepoTags": [repo_tag], "Layers": [layer_name]} + ] + with tarfile.open(archive_path, mode="w") as archive: + for name, raw in ( + (config_name, config_raw), + (layer_name, layer_raw), + ("manifest.json", bundle.canonical_json_bytes(docker_manifest)), + ): + item = tarfile.TarInfo(name) + item.size = len(raw) + item.mode = 0o644 + archive.addfile(item, io.BytesIO(raw)) + return ( + manifest_digest, + config_digest, + _record(manifest_path, root), + _record(archive_path, root), + ) + + +def _make_security_artifacts( + root: Path, + *, + role: str, + manifest_digest: str, + config_digest: str, +) -> tuple[dict[str, object], dict[str, object], dict[str, object]]: + sbom_path = root / "sbom" / f"{role}.spdx.json" + root_package = f"SPDXRef-Package-{role}" + _write_canonical( + sbom_path, + { + "spdxVersion": "SPDX-2.3", + "SPDXID": "SPDXRef-DOCUMENT", + "name": f"{role}-sbom", + "documentNamespace": f"https://example.invalid/spdx/{role}", + "creationInfo": {"creators": ["Tool: syft-1.18.1"]}, + "documentDescribes": [root_package], + "packages": [ + { + "SPDXID": root_package, + "name": role, + "versionInfo": manifest_digest, + } + ], + }, + ) + vulnerability_path = root / "scans" / f"{role}.vulnerability.json" + license_path = root / "scans" / f"{role}.license.json" + common = { + "SchemaVersion": 2, + "ArtifactName": role, + "ArtifactType": "container_image", + "Metadata": {"ImageID": config_digest}, + } + _write_canonical( + vulnerability_path, + { + **common, + "Results": [ + { + "Target": role, + "Class": "os-pkgs", + "Type": "alpine", + "Vulnerabilities": [], + } + ], + }, + ) + _write_canonical( + license_path, + { + **common, + "Results": [ + { + "Target": role, + "Class": "license", + "Type": "image", + "Licenses": [{"Name": "Apache-2.0", "Category": "notice"}], + } + ], + }, + ) + return ( + _record(sbom_path, root), + _record(vulnerability_path, root), + _record(license_path, root), + ) + + +def _fixture(root: Path) -> tuple[dict[str, object], dict[str, object]]: + commit = "1" * 40 + tag = "v2-test-1" + app_images: list[dict[str, object]] = [] + for index, (role, environment_key) in enumerate(bundle.APPLICATION_SPECS, start=1): + repo_tag = f"yuqei/{role}:{commit}" + manifest_digest, config_digest, _manifest, archive = _make_image_artifacts( + root, role=role, repo_tag=repo_tag, marker=hex(index)[-1] + ) + app_images.append( + { + "role": role, + "environment_key": environment_key, + "platform": "linux/amd64", + "manifest_digest": manifest_digest, + "config_digest": config_digest, + "image_ref": repo_tag, + "docker_archive": archive, + } + ) + application = _with_hash( + { + "schema_version": bundle.APPLICATION_MANIFEST_SCHEMA, + "candidate": {"commit": commit, "tag": tag}, + "images": app_images, + }, + "manifest_payload_sha256", + ) + + repositories = ( + "docker.io/pgvector/pgvector", + "docker.io/library/postgres", + "docker.io/library/redis", + "docker.io/minio/minio", + "docker.io/minio/mc", + "docker.io/library/nginx", + "docker.io/library/alpine", + ) + tags = ( + "pg16", + "16-alpine", + "7-alpine", + "RELEASE.2023-06-23T20-26-00Z", + "RELEASE.2023-06-23T18-12-07Z", + "1.27-alpine", + "3.21", + ) + infra_images: list[dict[str, object]] = [] + for index, ((role, environment_key, services), repository, image_tag) in enumerate( + zip(bundle.INFRASTRUCTURE_SPECS, repositories, tags, strict=True), start=8 + ): + offline_tag = f"yuqei-infra/{role}:offline" + manifest_digest, config_digest, manifest, archive = _make_image_artifacts( + root, role=role, repo_tag=offline_tag, marker=hex(index)[-1] + ) + sbom, vulnerability_scan, license_scan = _make_security_artifacts( + root, + role=role, + manifest_digest=manifest_digest, + config_digest=config_digest, + ) + infra_images.append( + { + "role": role, + "environment_key": environment_key, + "services": list(services), + "canonical_source_tag": f"{repository}:{image_tag}", + "canonical_reference": f"{repository}@{manifest_digest}", + "transport_reference": ( + f"docker.m.daocloud.io/{repository.removeprefix('docker.io/')}@{manifest_digest}" + ), + "manifest_digest": manifest_digest, + "config_digest": config_digest, + "registry_manifest": manifest, + "docker_archive": archive, + "expected_repo_tag": offline_tag, + "sbom": sbom, + "vulnerability_scan": vulnerability_scan, + "license_scan": license_scan, + } + ) + source_set = _with_hash( + { + "schema_version": bundle.SOURCE_SET_SCHEMA, + "candidate": {"commit": commit, "tag": tag}, + "target": bundle.TARGET, + "images": infra_images, + }, + "source_set_payload_sha256", + ) + return source_set, application + + +def _rehash_source(source: dict[str, object]) -> dict[str, object]: + source = dict(source) + source.pop("source_set_payload_sha256", None) + return _with_hash(source, "source_set_payload_sha256") + + +def test_bundle_is_deterministic_file_only_and_covers_all_eleven_images( + tmp_path: Path, +) -> None: + source, application = _fixture(tmp_path) + first = bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + ) + second = bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + ) + + assert first == second + assert first["target"] == bundle.TARGET + assert first["all_release_image_keys"] == list(bundle.ALL_IMAGE_KEYS) + assert list(first["offline_compose_overrides"]) == list(bundle.ALL_IMAGE_KEYS) + assert all( + value.startswith("sha256:") + for value in first["offline_compose_overrides"].values() + ) + plan = first["docker_import_plan"] + assert plan["network_access_required"] is False + assert plan["docker_required_for_validation"] is False + assert len(plan["images"]) == 7 + assert [item["role"] for item in plan["images"]] == [ + role for role, _key, _services in bundle.INFRASTRUCTURE_SPECS + ] + assert all(item["load_command_argv"][0] == "/usr/bin/docker" for item in plan["images"]) + assert all( + set(item) >= { + "sbom", + "sbom_summary", + "vulnerability_scan", + "vulnerability_scan_summary", + "license_scan", + "license_scan_summary", + } + for item in first["infrastructure_images"] + ) + assert all( + item["sbom_summary"]["root_manifest_digest"] == item["manifest_digest"] + and item["vulnerability_scan_summary"]["config_digest"] == item["config_digest"] + and item["license_scan_summary"]["config_digest"] == item["config_digest"] + for item in first["infrastructure_images"] + ) + assert first["bundle_payload_sha256"] == bundle.canonical_sha256( + {key: value for key, value in first.items() if key != "bundle_payload_sha256"} + ) + + +@pytest.mark.parametrize( + ("mutation", "message"), + ( + (lambda source: source["images"].pop(), "exactly seven"), + (lambda source: source.update({"target": {"os": "linux", "architecture": "arm64", "platform": "linux/arm64"}}), "linux/amd64"), + (lambda source: source["images"][4].update({"canonical_source_tag": "docker.io/minio/mc:latest"}), "latest"), + (lambda source: source["images"][0].update({"services": ["postgres", "unexpected"]}), "mapping"), + ), +) +def test_source_set_shape_and_platform_fail_closed( + tmp_path: Path, mutation, message: str +) -> None: + source, application = _fixture(tmp_path) + mutation(source) + source = _rehash_source(source) + + with pytest.raises(bundle.BundleValidationError, match=message): + bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + ) + + +def test_archive_content_change_is_rejected_even_if_outer_source_hash_is_refreshed( + tmp_path: Path, +) -> None: + source, application = _fixture(tmp_path) + record = source["images"][2]["docker_archive"] + archive = tmp_path / record["path"] + archive.write_bytes(archive.read_bytes() + b"tampered") + source = _rehash_source(source) + + with pytest.raises(bundle.BundleValidationError, match="size or SHA-256 mismatch"): + bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + ) + + +def test_registry_manifest_and_docker_archive_config_identity_must_match( + tmp_path: Path, +) -> None: + source, application = _fixture(tmp_path) + source["images"][0]["config_digest"] = "sha256:" + "f" * 64 + source = _rehash_source(source) + + with pytest.raises(bundle.BundleValidationError, match="config digest mismatch"): + bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + ) + + +def test_candidate_binding_between_application_and_infrastructure_is_required( + tmp_path: Path, +) -> None: + source, application = _fixture(tmp_path) + application = dict(application) + application.pop("manifest_payload_sha256") + application["candidate"] = {"commit": "2" * 40, "tag": "v2-other"} + application = _with_hash(application, "manifest_payload_sha256") + + with pytest.raises(bundle.BundleValidationError, match="candidate bindings differ"): + bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + ) + + +@pytest.mark.parametrize("field", ("sbom", "vulnerability_scan", "license_scan")) +def test_every_infrastructure_security_artifact_is_required( + tmp_path: Path, + field: str, +) -> None: + source, application = _fixture(tmp_path) + source["images"][0].pop(field) + source = _rehash_source(source) + + with pytest.raises(bundle.BundleValidationError, match="fields are missing or unknown"): + bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + ) + + +@pytest.mark.parametrize( + ("field", "message"), + ( + ("sbom", "SBOM root does not bind"), + ("vulnerability_scan", "vulnerability scan config image ID mismatch"), + ("license_scan", "license scan config image ID mismatch"), + ), +) +def test_security_artifact_image_identity_drift_is_rejected( + tmp_path: Path, + field: str, + message: str, +) -> None: + source, application = _fixture(tmp_path) + record = source["images"][0][field] + path = tmp_path / record["path"] + payload = json.loads(path.read_text(encoding="utf-8")) + if field == "sbom": + payload["packages"][0]["versionInfo"] = "sha256:" + "f" * 64 + else: + payload["Metadata"]["ImageID"] = "sha256:" + "f" * 64 + _write_canonical(path, payload) + source["images"][0][field] = _record(path, tmp_path) + source = _rehash_source(source) + + with pytest.raises(bundle.BundleValidationError, match=message): + bundle.build_bundle( + artifact_root=tmp_path, + source_set=source, + application_manifest=application, + )