release: validate infrastructure bundle and first-release recovery
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -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())
|
||||
@@ -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
|
||||
@@ -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,
|
||||
)
|
||||
Reference in New Issue
Block a user