"""Strict, dependency-free validation helpers for the Linux release pipeline. The release scripts deliberately keep this module in the standard library so a clean builder can validate all release inputs before it installs or runs any third-party tooling. """ from __future__ import annotations import hashlib import gzip import json import os import re import shutil import stat import subprocess import sys import tarfile from datetime import datetime, timedelta, timezone from pathlib import Path, PurePosixPath from typing import Any, BinaryIO, Iterable POLICY_SCHEMA = "yuqei.linux-release-security-policy.v1" SOURCE_MANIFEST_SCHEMA = "yuqei.release-source-manifest.v1" BUILD_RECEIPT_SCHEMA = "yuqei.linux-release-build-receipt.v1" INSTALLABLE_MANIFEST_SCHEMA = "yuqei.installable-release-manifest.v1" TARGET = {"os": "linux", "architecture": "amd64", "platform": "linux/amd64"} BUILDER_POLICY = { "allowed_drivers": ["docker-container"], "maximum_nodes": 1, "required_buildkit_version": "0.13.2", "required_driver_image": ( "docker.m.daocloud.io/moby/buildkit@" "sha256:9194b5ec1be368f41c516df7f93f7f540630ea06136056b2ffebb62226ed4ad6" ), "required_driver_image_digest": ( "sha256:9194b5ec1be368f41c516df7f93f7f540630ea06136056b2ffebb62226ed4ad6" ), "required_exporters": ["oci", "docker"], "required_local_driver_image_id": ( "sha256:1a21071780b069d68bd7ef277675f1c5273fee488cec145494cf072777260797" ), "required_platform": "linux/amd64", } COMPOSE_GATE_POLICY = { "case_count": 8, "compose_inputs": [ "yuqei-infra/compose/docker-compose.yml", "yuqei-infra/compose/docker-compose.production.yml", "yuqei-infra/compose/docker-compose.production.offline.yml", ], "production_secrets_forbidden": True, "schema_version": "yuqei.production-compose-merge-gate.v1", "synthetic_environment_required": True, "validator": "yuqei-infra/deploy/verify-production-compose.py", } PINNED_EXECUTABLES = { "docker": "/usr/bin/docker", "git": "/usr/bin/git", "syft": "/usr/bin/syft", "trivy": "/usr/bin/trivy", } LOCAL_DOCKER_HOST = "unix:///var/run/docker.sock" CONTROLLED_SUBPROCESS_ENV = { "BUILDX_CONFIG": "/root/.docker/buildx", "DOCKER_CONFIG": "/var/empty/docker", "HOME": "/var/empty", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PATH": "/usr/bin:/bin", } COMMAND_TIMEOUTS_SECONDS = { "base_image_inspect": 120, "builder_inspect": 120, "compose_gate": 600, "container_inspect": 30, "git_validation": 30, "image_build": 3600, "sbom_generation": 600, "tool_version": 30, "trivy_db_download": 600, "trivy_db_inspect": 120, "vulnerability_scan": 900, } TRIVY_DB_REPOSITORY_TRANSPORT = { "allowed_media_types": ["application/vnd.oci.image.manifest.v1+json"], "reference": "docker.m.daocloud.io/aquasec/trivy-db:2", "schema_version": 2, "transport": "domestic_registry_mirror", } GIT_VALIDATION_ENV = { "GIT_CONFIG_NOSYSTEM": "1", "GIT_NO_REPLACE_OBJECTS": "1", "GIT_OPTIONAL_LOCKS": "0", "GIT_TERMINAL_PROMPT": "0", } # Windows supports read-only plan/assembly validation, while real release # execution is Linux-only. Resolve Git once while this self-bound module is # imported; a caller cannot redirect later validation by mutating ambient PATH. _WINDOWS_GIT_EXECUTABLE = ( None if sys.platform.startswith("linux") else ( str(Path(discovered).resolve()) if (discovered := shutil.which("git")) is not None else None ) ) VULNERABILITY_WAIVERS = [ { "id": "RC04.6-DEBIAN13-PYTHON-UNFIXED-HIGH-001", "expires_at": "2026-10-20T00:00:00Z", "scope": { "roles": ["contract-api", "ai-platform-api", "ai-worker"], "class": "os-pkgs", "type": "debian", "os_family": "debian", "severity": "HIGH", "fixed_version_must_be_empty": True, }, "findings": [ { "vulnerability_id": "CVE-2025-69720", "package_names": ["libncursesw6", "libtinfo6", "ncurses-base", "ncurses-bin"], "installed_version": "6.5+20250216-2", }, { "vulnerability_id": "CVE-2026-16742", "package_names": ["libsystemd0", "libudev1"], "installed_version": "257.13-1~deb13u1", }, { "vulnerability_id": "CVE-2026-54369", "package_names": ["libacl1"], "installed_version": "2.3.2-2+b1", }, { "vulnerability_id": "CVE-2026-76642", "package_names": ["bsdutils"], "installed_version": "1:2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-76642", "package_names": [ "libblkid1", "liblastlog2-2", "libmount1", "libsmartcols1", "libuuid1", "mount", "util-linux", ], "installed_version": "2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-76642", "package_names": ["login"], "installed_version": "1:4.16.0-2+really2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78408", "package_names": ["bsdutils"], "installed_version": "1:2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78408", "package_names": [ "libblkid1", "liblastlog2-2", "libmount1", "libsmartcols1", "libuuid1", "mount", "util-linux", ], "installed_version": "2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78408", "package_names": ["login"], "installed_version": "1:4.16.0-2+really2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78409", "package_names": ["bsdutils"], "installed_version": "1:2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78409", "package_names": [ "libblkid1", "liblastlog2-2", "libmount1", "libsmartcols1", "libuuid1", "mount", "util-linux", ], "installed_version": "2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78409", "package_names": ["login"], "installed_version": "1:4.16.0-2+really2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78410", "package_names": ["bsdutils"], "installed_version": "1:2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78410", "package_names": [ "libblkid1", "liblastlog2-2", "libmount1", "libsmartcols1", "libuuid1", "mount", "util-linux", ], "installed_version": "2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-78410", "package_names": ["login"], "installed_version": "1:4.16.0-2+really2.41.5-0+deb13u1", }, { "vulnerability_id": "CVE-2026-9538", "package_names": ["perl-base"], "installed_version": "5.40.1-6+deb13u1", }, ], } ] PYTHON_BASE = { "build_arg": "PYTHON_BASE_IMAGE", "content_digest": "sha256:44ff437bba879d4941b710a369a8f19266aea34b29002807f0c487fabc9eec9b", "source_reference": ( "python:3.12-slim@sha256:44ff437bba879d4941b710a369a8f19266aea34b29002807f0c487fabc9eec9b" ), "transport": "domestic_registry_mirror", "transport_reference": ( "docker.m.daocloud.io/library/python:3.12-slim@" "sha256:44ff437bba879d4941b710a369a8f19266aea34b29002807f0c487fabc9eec9b" ), } NODE_BASE = { "build_arg": "NODE_BASE_IMAGE", "content_digest": "sha256:b64da1de5a51067ab8e75f0bc8dbd0905d8894baa22261f439a4572f41291e50", "source_reference": ( "node:22-alpine@sha256:b64da1de5a51067ab8e75f0bc8dbd0905d8894baa22261f439a4572f41291e50" ), "transport": "domestic_registry_mirror", "transport_reference": ( "docker.m.daocloud.io/library/node:22-alpine@" "sha256:b64da1de5a51067ab8e75f0bc8dbd0905d8894baa22261f439a4572f41291e50" ), } REQUIRED_TOOL_VERSIONS = { "docker_engine": "26.1.3", "docker_cli": "26.1.3", "buildx": "0.14.0", "compose": "2.27.0", "syft": "1.52.0", "trivy": "0.74.0", } REQUIRED_IMAGE_SPECS = ( { "role": "contract-web", "base_image": NODE_BASE, "name": "yuqei/contract-web", "environment_key": "CONTRACT_WEB_IMAGE", "dockerfile": "yuqei-infra/docker/contract-web.Dockerfile", }, { "role": "contract-api", "base_image": PYTHON_BASE, "name": "yuqei/contract-api", "environment_key": "CONTRACT_API_IMAGE", "dockerfile": "yuqei-infra/docker/contract-api.Dockerfile", }, { "role": "ai-platform-api", "base_image": PYTHON_BASE, "name": "yuqei/ai-platform-api", "environment_key": "AI_PLATFORM_API_IMAGE", "dockerfile": "yuqei-infra/docker/ai-platform-api.Dockerfile", }, { "role": "ai-worker", "base_image": PYTHON_BASE, "name": "yuqei/ai-worker", "environment_key": "AI_WORKER_IMAGE", "dockerfile": "yuqei-infra/docker/worker.Dockerfile", }, ) REQUIRED_ROLES = tuple(item["role"] for item in REQUIRED_IMAGE_SPECS) SHA256_RE = re.compile(r"[0-9a-f]{64}") FULL_COMMIT_RE = re.compile(r"[0-9a-f]{40}") GIT_OBJECT_RE = re.compile(r"(?:[0-9a-f]{40}|[0-9a-f]{64})") TAG_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,127}") RFC3339_RE = re.compile( r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}" r"(?:\.[0-9]+)?(?:Z|[+-][0-9]{2}:[0-9]{2})" ) class ReleaseValidationError(RuntimeError): """An input or generated artifact is unsafe, ambiguous, or inconsistent.""" def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ReleaseValidationError(f"duplicate JSON key: {key}") result[key] = value return result def load_json_bytes(data: bytes, *, label: str) -> Any: try: text = data.decode("utf-8", "strict") except UnicodeDecodeError as error: raise ReleaseValidationError(f"{label} is not UTF-8: {error}") from error try: return json.loads( text, object_pairs_hook=_reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw( ReleaseValidationError(f"non-finite JSON value: {value}") ), ) except (json.JSONDecodeError, ReleaseValidationError) as error: raise ReleaseValidationError(f"{label} is not valid unambiguous JSON: {error}") from error def load_json_file(path: Path, *, label: str) -> Any: try: data = path.read_bytes() except OSError as error: raise ReleaseValidationError(f"cannot read {label} {path}: {error}") from error return load_json_bytes(data, label=label) def canonical_json_bytes(payload: Any) -> bytes: return ( json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ) + "\n" ).encode("utf-8") def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def sha256_file(path: Path) -> str: digest = hashlib.sha256() try: with path.open("rb") as handle: while chunk := handle.read(1024 * 1024): digest.update(chunk) except OSError as error: raise ReleaseValidationError(f"cannot hash {path}: {error}") from error return digest.hexdigest() def controlled_subprocess_env(extra: dict[str, str] | None = None) -> dict[str, str]: """Return the allowlisted child environment used by release commands.""" if sys.platform.startswith("linux"): result = dict(CONTROLLED_SUBPROCESS_ENV) else: # Plan/assembly validation remains usable on Windows, but real builds # are Linux-only. Keep only OS bootstrap variables and a fixed Git # directory; never forward Docker routing/configuration variables. result = { key: os.environ[key] for key in ("SYSTEMROOT", "WINDIR", "TEMP", "TMP") if key in os.environ } if _WINDOWS_GIT_EXECUTABLE: result["PATH"] = str(Path(_WINDOWS_GIT_EXECUTABLE).parent) result.update({"LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"}) if extra: result.update(extra) return result def release_git_executable() -> str: if sys.platform.startswith("linux"): return PINNED_EXECUTABLES["git"] if not _WINDOWS_GIT_EXECUTABLE: raise ReleaseValidationError("Git executable is unavailable for plan validation") return _WINDOWS_GIT_EXECUTABLE def validate_pinned_executables() -> dict[str, dict[str, Any]]: """Validate root ownership/mode and hash every pinned Linux executable.""" if not sys.platform.startswith("linux"): raise ReleaseValidationError("pinned release executables can be validated only on Linux") result: dict[str, dict[str, Any]] = {} for name, raw_path in PINNED_EXECUTABLES.items(): path = Path(raw_path) if path.is_symlink() or not path.is_file(): raise ReleaseValidationError(f"pinned {name} executable is missing or a symlink: {path}") metadata = path.stat() mode = stat.S_IMODE(metadata.st_mode) if metadata.st_uid != 0 or mode != 0o755: raise ReleaseValidationError( f"pinned {name} executable must be root-owned mode 0755: {path}" ) result[name] = { "path": raw_path, "sha256": sha256_file(path), "size_bytes": metadata.st_size, "owner_uid": metadata.st_uid, "mode": "0755", } return result def validate_buildx_config_root(builder_name: str) -> dict[str, Any]: """Validate the one fixed Buildx state tree without trusting ambient Docker config.""" if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,63}", builder_name): raise ReleaseValidationError("Buildx builder name is missing or unsafe") root = Path(CONTROLLED_SUBPROCESS_ENV["BUILDX_CONFIG"]) required_directories = [Path("/root"), Path("/root/.docker"), root] for path in required_directories: if path.is_symlink() or not path.is_dir(): raise ReleaseValidationError(f"fixed Buildx config component is missing or unsafe: {path}") metadata = path.stat() if metadata.st_uid != 0 or stat.S_IMODE(metadata.st_mode) & 0o022: raise ReleaseValidationError( f"fixed Buildx config component is not root-owned/non-writable: {path}" ) instance = root / "instances" / builder_name if instance.is_symlink() or not instance.is_file(): raise ReleaseValidationError( f"fixed Buildx config has no regular selected-builder instance: {instance}" ) inventory: list[str] = [] for path in sorted(root.rglob("*"), key=lambda item: item.as_posix().encode("utf-8")): relative = path.relative_to(root).as_posix() if path.is_symlink(): raise ReleaseValidationError(f"fixed Buildx config contains a symlink: {relative}") metadata = path.stat() mode = stat.S_IMODE(metadata.st_mode) if metadata.st_uid != 0 or mode & 0o022: raise ReleaseValidationError( f"fixed Buildx config entry is not root-owned/non-writable: {relative}" ) if path.is_dir(): kind = "directory" elif path.is_file(): kind = "file" else: raise ReleaseValidationError( f"fixed Buildx config contains a special entry: {relative}" ) inventory.append(f"{relative}\0{kind}\0{metadata.st_uid}\0{mode:04o}") return { "path": root.as_posix(), "selected_builder_instance": instance.relative_to(root).as_posix(), "entry_count": len(inventory), "metadata_sha256": sha256_bytes("\n".join(inventory).encode("utf-8")), "root_owned": True, "group_or_other_writable": False, "symlinks_allowed": False, } def file_record(path: Path, relative_path: str, **extra: Any) -> dict[str, Any]: validate_relative_path(relative_path, label="artifact path") record = { "path": relative_path, "sha256": sha256_file(path), "size_bytes": path.stat().st_size, } record.update(extra) return record def validate_relative_path(value: str, *, label: str) -> PurePosixPath: if not isinstance(value, str) or not value: raise ReleaseValidationError(f"{label} must be a non-empty POSIX relative path") if "\\" in value or "\x00" in value or ":" in value: raise ReleaseValidationError(f"{label} contains a forbidden character: {value!r}") path = PurePosixPath(value) if path.is_absolute() or value.startswith("/"): raise ReleaseValidationError(f"{label} must be relative: {value!r}") if any(part in {"", ".", ".."} for part in path.parts): raise ReleaseValidationError(f"{label} contains traversal or ambiguous components: {value!r}") if path.as_posix() != value: raise ReleaseValidationError(f"{label} is not canonical POSIX form: {value!r}") return path def _require_mapping(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict): raise ReleaseValidationError(f"{label} must be a JSON object") return value def _require_list(value: Any, label: str) -> list[Any]: if not isinstance(value, list): raise ReleaseValidationError(f"{label} must be a JSON array") return value def _require_exact_keys(value: dict[str, Any], required: set[str], label: str) -> None: actual = set(value) if actual != required: raise ReleaseValidationError( f"{label} keys differ: missing={sorted(required - actual)}, unknown={sorted(actual - required)}" ) def validate_policy(payload: Any) -> dict[str, Any]: policy = _require_mapping(payload, "release policy") _require_exact_keys( policy, { "schema_version", "target", "builder", "compose_gate", "execution", "toolchain", "images", "sbom", "vulnerability_scan", "unsupported_tools", }, "release policy", ) if policy["schema_version"] != POLICY_SCHEMA: raise ReleaseValidationError(f"unsupported release policy schema: {policy['schema_version']!r}") if policy["target"] != TARGET: raise ReleaseValidationError(f"release target must be exactly {TARGET}") if policy["builder"] != BUILDER_POLICY: raise ReleaseValidationError( "builder policy must require one linux/amd64 docker-container node with OCI and Docker exporters" ) if policy["compose_gate"] != COMPOSE_GATE_POLICY: raise ReleaseValidationError( "Compose gate policy must require the eight synthetic online/offline cases" ) execution = _require_mapping(policy["execution"], "execution") if execution != { "command_timeouts_seconds": COMMAND_TIMEOUTS_SECONDS, "controlled_environment": CONTROLLED_SUBPROCESS_ENV, "docker_host": LOCAL_DOCKER_HOST, "pinned_executables": PINNED_EXECUTABLES, }: raise ReleaseValidationError( "execution policy must pin the complete external-command timeout contract" ) toolchain = _require_mapping(policy["toolchain"], "toolchain") if set(toolchain) != set(REQUIRED_TOOL_VERSIONS): raise ReleaseValidationError("toolchain entries do not match the required pinned tool set") for tool, required_version in REQUIRED_TOOL_VERSIONS.items(): entry = _require_mapping(toolchain[tool], f"toolchain.{tool}") _require_exact_keys(entry, {"allowed_versions"}, f"toolchain.{tool}") if entry["allowed_versions"] != [required_version]: raise ReleaseValidationError( f"toolchain.{tool} must allow only {required_version}, got {entry['allowed_versions']!r}" ) images = _require_list(policy["images"], "images") if images != list(REQUIRED_IMAGE_SPECS): raise ReleaseValidationError( "image roles, order, names, and Dockerfiles must match the four-image release contract" ) for index, image in enumerate(images): _require_exact_keys( _require_mapping(image, f"images[{index}]"), {"role", "name", "environment_key", "dockerfile", "base_image"}, f"images[{index}]", ) validate_relative_path(image["dockerfile"], label=f"images[{index}].dockerfile") sbom = _require_mapping(policy["sbom"], "sbom") if sbom != {"generator": "syft", "format": "spdx-json", "spdx_version": "SPDX-2.3"}: raise ReleaseValidationError("SBOM policy must require Syft SPDX-2.3 JSON") scan = _require_mapping(policy["vulnerability_scan"], "vulnerability_scan") _require_exact_keys( scan, { "database", "exemptions", "fail_closed_on", "format", "ignore_unfixed", "scanner", "scanners", "thresholds", }, "vulnerability_scan", ) if scan["scanner"] != "trivy" or scan["format"] != "json" or scan["scanners"] != ["vuln"]: raise ReleaseValidationError("vulnerability scanner must be Trivy JSON in vuln-only mode") if scan["ignore_unfixed"] is not False: raise ReleaseValidationError("unfixed vulnerabilities must not be ignored") if scan["thresholds"] != {"CRITICAL": 0, "HIGH": 0}: raise ReleaseValidationError("vulnerability thresholds must be CRITICAL=0 and HIGH=0") if scan["exemptions"] != VULNERABILITY_WAIVERS: raise ReleaseValidationError( "release vulnerability waivers must exactly match the time-limited package allowlist" ) fail_closed = _require_list(scan["fail_closed_on"], "vulnerability_scan.fail_closed_on") mandatory_failures = { "database_download_failure", "database_metadata_missing", "database_expired", "database_too_old", "scanner_execution_failure", "scan_output_missing", "json_parse_failure", "schema_validation_failure", "waiver_expired", "waiver_scope_mismatch", "waiver_finding_mismatch", } if set(fail_closed) != mandatory_failures or len(fail_closed) != len(mandatory_failures): raise ReleaseValidationError("fail-closed conditions are incomplete or duplicated") database = _require_mapping(scan["database"], "vulnerability_scan.database") _require_exact_keys( database, { "fail_closed", "maximum_age_hours", "repository_transport", "require_next_update_after_scan", }, "vulnerability_scan.database", ) if database != { "fail_closed": True, "maximum_age_hours": 24, "repository_transport": TRIVY_DB_REPOSITORY_TRANSPORT, "require_next_update_after_scan": True, }: raise ReleaseValidationError("Trivy database must fail closed at a 24-hour maximum age") if policy["unsupported_tools"] != { "skopeo": "not_required_and_must_not_be_used_by_this_pipeline" }: raise ReleaseValidationError("Skopeo must remain explicitly outside this pipeline") return policy def load_policy(path: Path) -> dict[str, Any]: return validate_policy(load_json_file(path, label="release policy")) def synthetic_compose_gate_env_bytes() -> bytes: """Return the deterministic, non-secret environment for structural Compose validation.""" image_values = { "CONTRACT_API_IMAGE": ("contract-api", "1"), "CONTRACT_WEB_IMAGE": ("contract-web", "2"), "AI_PLATFORM_API_IMAGE": ("ai-platform-api", "3"), "AI_WORKER_IMAGE": ("ai-worker", "4"), "POSTGRES_IMAGE": ("postgres", "5"), "POSTGRES_CLIENT_IMAGE": ("postgres-client", "6"), "REDIS_IMAGE": ("redis", "7"), "MINIO_IMAGE": ("minio", "8"), "MINIO_MC_IMAGE": ("minio-mc", "9"), "NGINX_IMAGE": ("nginx", "a"), "BACKUP_UTILITY_IMAGE": ("backup-utility", "b"), } values = { key: f"synthetic.invalid/yuqei/{name}@sha256:{digit * 64}" for key, (name, digit) in image_values.items() } values.update( { "AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256": "c" * 64, "AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST": ( "/synthetic/provider-acceptance/receipt.json" ), "AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST": ( "/synthetic/provider-acceptance/trusted-keys.json" ), "AI_PLATFORM_RELEASE_IMAGE_DIGEST": "sha256:" + "d" * 64, "AI_PROVIDER_API_KEY_FILE": "/synthetic/not-a-secret/provider-api-key", "BACKEND_SERVER_NAME": "synthetic.invalid", "BACKEND_TLS_CERT_DIR": "/synthetic/tls", "BACKUP_HOST_DIR": "/synthetic/backups", "COMPOSE_PROJECT_NAME": "yuqei-rc04-structural-gate", "CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST": "sha256:" + "e" * 64, "MINIO_ROOT_PASSWORD": "synthetic-not-a-production-secret", "MINIO_ROOT_USER": "synthetic-not-a-production-user", "PUBLIC_POLICY_CANDIDATE_DIR_HOST": "/synthetic/public-policy/candidate", "PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST": ( "/synthetic/public-policy/legal-approval.json" ), "PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST": ( "/synthetic/public-policy/trusted-keyring.json" ), "RELEASE_ID": "synthetic-rc04-structure-only", "YUQEI_COMPOSE_GATE_SYNTHETIC": "true", } ) header = ( "# SYNTHETIC STRUCTURAL VALIDATION ONLY.\n" "# Contains no production credentials and must never be used for deployment.\n" ) body = "".join(f"{key}={values[key]}\n" for key in sorted(values)) return (header + body).encode("utf-8") def validate_compose_gate_report( payload: Any, policy: dict[str, Any], *, expected_env_sha256: str, source_file_sha256: dict[str, str], ) -> dict[str, Any]: report = _require_mapping(payload, "production Compose gate report") _require_exact_keys( report, { "schema_version", "status", "env_file_sha256", "compose_inputs", "application_service_count", "production_service_count", "cases", }, "production Compose gate report", ) gate_policy = policy["compose_gate"] if report["schema_version"] != gate_policy["schema_version"] or report["status"] != "passed": raise ReleaseValidationError("production Compose gate did not report a supported pass") if report["env_file_sha256"] != expected_env_sha256: raise ReleaseValidationError("production Compose gate environment digest drift") compose_inputs = _require_list(report["compose_inputs"], "Compose gate inputs") if len(compose_inputs) != len(gate_policy["compose_inputs"]): raise ReleaseValidationError("production Compose gate input count drift") validated_inputs: list[dict[str, str]] = [] for index, (raw_input, expected_path) in enumerate( zip(compose_inputs, gate_policy["compose_inputs"], strict=True) ): item = _require_mapping(raw_input, f"Compose gate input {index}") _require_exact_keys(item, {"path", "sha256"}, f"Compose gate input {index}") expected_digest = source_file_sha256.get(expected_path) if item.get("path") != expected_path or item.get("sha256") != expected_digest: raise ReleaseValidationError(f"production Compose gate input drift: {expected_path}") if not isinstance(expected_digest, str) or not SHA256_RE.fullmatch(expected_digest): raise ReleaseValidationError(f"source manifest lacks Compose input: {expected_path}") validated_inputs.append({"path": expected_path, "sha256": expected_digest}) if report["application_service_count"] != 9 or report["production_service_count"] != 16: raise ReleaseValidationError("production Compose gate service inventory drift") applications_without_web = [ "contract-api-migrate", "contract-api", "contract-review-worker", "contract-playbook-governance-worker", "ai-platform-api-migrate", "ai-platform-prompt-assets-import", "ai-platform-api", "worker", ] expected_cases = [] for profile_name, profiles, service_count, includes_web in ( ("default", [], 13, False), ("backend-hosted-web", ["backend-hosted-web"], 14, True), ("ops", ["ops"], 15, False), ("backend-hosted-web+ops", ["backend-hosted-web", "ops"], 16, True), ): applications = applications_without_web + (["contract-web"] if includes_web else []) for mode in ("online", "offline"): expected_cases.append((profile_name, profiles, mode, service_count, applications)) cases = _require_list(report["cases"], "Compose gate cases") if len(cases) != gate_policy["case_count"] or len(cases) != len(expected_cases): raise ReleaseValidationError("production Compose gate did not run exactly eight cases") config_digests: list[str] = [] for index, (raw_case, expected) in enumerate(zip(cases, expected_cases, strict=True)): case = _require_mapping(raw_case, f"Compose gate case {index}") _require_exact_keys( case, { "mode", "profile_case", "profiles", "service_count", "validated_application_services", "config_sha256", }, f"Compose gate case {index}", ) profile_name, profiles, mode, service_count, applications = expected if ( case["profile_case"] != profile_name or case["profiles"] != profiles or case["mode"] != mode or case["service_count"] != service_count or case["validated_application_services"] != applications ): raise ReleaseValidationError(f"production Compose gate case {index} drift") config_digest = case["config_sha256"] if not isinstance(config_digest, str) or not SHA256_RE.fullmatch(config_digest): raise ReleaseValidationError(f"production Compose gate case {index} digest is invalid") config_digests.append(config_digest) if len(set(config_digests)) != len(config_digests): raise ReleaseValidationError("production Compose gate cases produced duplicate models") return { "schema_version": report["schema_version"], "case_count": len(cases), "compose_inputs": validated_inputs, "config_sha256s": config_digests, "synthetic_environment": True, "production_secrets_used": False, } def parse_rfc3339(value: Any, *, label: str) -> datetime: if not isinstance(value, str) or not RFC3339_RE.fullmatch(value): raise ReleaseValidationError(f"{label} must be a non-empty RFC3339 timestamp") normalized = value[:-1] + "+00:00" if value.endswith("Z") else value try: parsed = datetime.fromisoformat(normalized) except ValueError as error: raise ReleaseValidationError(f"{label} is not RFC3339: {value!r}") from error if parsed.tzinfo is None: raise ReleaseValidationError(f"{label} must include a timezone") return parsed.astimezone(timezone.utc) def format_rfc3339(value: datetime) -> str: return value.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def validate_source_manifest(payload: Any, *, commit: str) -> dict[str, Any]: manifest = _require_mapping(payload, "source manifest") if manifest.get("schema_version") != SOURCE_MANIFEST_SCHEMA: raise ReleaseValidationError("source manifest schema is not the immutable RC source schema") if not FULL_COMMIT_RE.fullmatch(commit): raise ReleaseValidationError("commit must be a full lowercase 40-character Git object id") claimed_payload_digest = manifest.get("manifest_payload_sha256") if not isinstance(claimed_payload_digest, str) or not SHA256_RE.fullmatch(claimed_payload_digest): raise ReleaseValidationError("source manifest payload digest is missing or invalid") unsigned = dict(manifest) del unsigned["manifest_payload_sha256"] actual_payload_digest = sha256_bytes( json.dumps(unsigned, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") ) if actual_payload_digest != claimed_payload_digest: raise ReleaseValidationError("source manifest payload digest mismatch") candidate = _require_mapping(manifest.get("candidate"), "source manifest candidate") for key in ("commit", "tree", "commit_timestamp", "worktree_was_clean", "identity_policy"): if key not in candidate: raise ReleaseValidationError(f"source manifest candidate is missing {key}") if candidate["commit"] != commit: raise ReleaseValidationError( f"source manifest commit {candidate['commit']!r} does not match explicit commit {commit}" ) if not GIT_OBJECT_RE.fullmatch(str(candidate["tree"])): raise ReleaseValidationError("source manifest tree object id is invalid") parse_rfc3339(candidate["commit_timestamp"], label="candidate.commit_timestamp") if candidate["worktree_was_clean"] is not True: raise ReleaseValidationError("source manifest does not attest a clean candidate") if candidate["identity_policy"] != "full_commit_oid_equals_HEAD": raise ReleaseValidationError("source manifest identity policy is not strict") files = _require_mapping(manifest.get("files"), "source manifest files") for key in ("count", "total_size", "content_index_sha256", "items"): if key not in files: raise ReleaseValidationError(f"source manifest files is missing {key}") items = _require_list(files["items"], "source manifest files.items") if ( not isinstance(files["count"], int) or isinstance(files["count"], bool) or files["count"] < 0 ): raise ReleaseValidationError("source manifest file count is invalid") if ( not isinstance(files["total_size"], int) or isinstance(files["total_size"], bool) or files["total_size"] < 0 ): raise ReleaseValidationError("source manifest total size is invalid") if not isinstance(files["content_index_sha256"], str) or not SHA256_RE.fullmatch( files["content_index_sha256"] ): raise ReleaseValidationError("source manifest content index digest is invalid") records: list[dict[str, Any]] = [] seen: set[str] = set() seen_casefold: set[str] = set() for index, raw_record in enumerate(items): record = _require_mapping(raw_record, f"source manifest files.items[{index}]") _require_exact_keys( record, {"path", "mode", "git_blob_oid", "size", "sha256"}, f"source manifest files.items[{index}]", ) path = record["path"] validate_relative_path(path, label=f"source file path {index}") folded = path.casefold() if path in seen or folded in seen_casefold: raise ReleaseValidationError(f"duplicate or case-colliding source path: {path}") seen.add(path) seen_casefold.add(folded) if record["mode"] not in {"100644", "100755"}: raise ReleaseValidationError(f"unsupported source file mode for {path}: {record['mode']!r}") if not GIT_OBJECT_RE.fullmatch(str(record["git_blob_oid"])): raise ReleaseValidationError(f"invalid Git blob object id for {path}") if not isinstance(record["size"], int) or isinstance(record["size"], bool) or record["size"] < 0: raise ReleaseValidationError(f"invalid source file size for {path}") if not isinstance(record["sha256"], str) or not SHA256_RE.fullmatch(record["sha256"]): raise ReleaseValidationError(f"invalid source file SHA-256 for {path}") records.append(record) sorted_paths = sorted((record["path"] for record in records), key=lambda item: item.encode("utf-8")) if [record["path"] for record in records] != sorted_paths: raise ReleaseValidationError("source manifest files are not in canonical UTF-8 path order") if files["count"] != len(records): raise ReleaseValidationError("source manifest file count mismatch") if files["total_size"] != sum(record["size"] for record in records): raise ReleaseValidationError("source manifest total size mismatch") content_index = "\n".join( f"{record['path']}\0{record['mode']}\0{record['size']}\0{record['sha256']}" for record in records ).encode("utf-8") if files["content_index_sha256"] != sha256_bytes(content_index): raise ReleaseValidationError("source manifest content index digest mismatch") return manifest def load_source_manifest(path: Path, *, commit: str) -> dict[str, Any]: return validate_source_manifest(load_json_file(path, label="source manifest"), commit=commit) def _git_blob_oid(content: bytes, expected_length: int) -> str: framed = b"blob " + str(len(content)).encode("ascii") + b"\0" + content if expected_length == 40: return hashlib.sha1(framed).hexdigest() # noqa: S324 - Git SHA-1 object identity, not security. if expected_length == 64: return hashlib.sha256(framed).hexdigest() raise ReleaseValidationError("unsupported Git object id length") def _archive_member_path(name: str, *, label: str) -> str: value = name[:-1] if name.endswith("/") else name validate_relative_path(value, label=label) return value def _expected_parent_directories(paths: Iterable[str]) -> set[str]: result: set[str] = set() for value in paths: parent = PurePosixPath(value).parent while parent != PurePosixPath("."): result.add(parent.as_posix()) parent = parent.parent return result def _validate_tar_stream_termination(path: Path) -> None: """Reject truncation and non-zero data hidden after the tar end marker.""" with path.open("rb") as probe: magic = probe.read(6) if magic.startswith(b"\x1f\x8b"): handle: BinaryIO = gzip.open(path, "rb") elif magic.startswith((b"BZh", b"\xfd7zXZ")): raise ReleaseValidationError("source archives must be uncompressed tar or gzip-compressed tar") else: handle = path.open("rb") with handle: while True: header = handle.read(512) if len(header) != 512: raise ReleaseValidationError("source archive lacks a complete tar end marker") if header == b"\0" * 512: second = handle.read(512) if second != b"\0" * 512: raise ReleaseValidationError("source archive has an incomplete tar end marker") while remainder := handle.read(1024 * 1024): if any(remainder): raise ReleaseValidationError("source archive contains non-zero trailing data") if len(remainder) % 512: raise ReleaseValidationError("source archive trailing padding is not block aligned") return raw_size = header[124:136] try: if raw_size[0] & 0x80: size = int.from_bytes(bytes([raw_size[0] & 0x7F]) + raw_size[1:], "big") else: text_size = raw_size.rstrip(b"\0 ").lstrip(b" ") or b"0" size = int(text_size, 8) except (ValueError, IndexError) as error: raise ReleaseValidationError("source archive contains an invalid tar size field") from error if size < 0: raise ReleaseValidationError("source archive contains a negative tar member size") remaining = ((size + 511) // 512) * 512 while remaining: chunk = handle.read(min(remaining, 1024 * 1024)) if not chunk: raise ReleaseValidationError("source archive is truncated within a member") remaining -= len(chunk) def verify_source_archive( archive_path: Path, manifest: dict[str, Any], *, extract_to: Path | None = None, ) -> dict[str, Any]: if archive_path.is_symlink() or not archive_path.is_file(): raise ReleaseValidationError("source archive must be a regular non-symlink file") _validate_tar_stream_termination(archive_path) records = {item["path"]: item for item in manifest["files"]["items"]} expected_paths = set(records) expected_mtime = int(parse_rfc3339( manifest["candidate"]["commit_timestamp"], label="candidate.commit_timestamp" ).timestamp()) try: archive = tarfile.open(archive_path, mode="r:*") except (OSError, tarfile.TarError) as error: raise ReleaseValidationError(f"source archive is not a readable tar archive: {error}") from error with archive: members = archive.getmembers() if not members: raise ReleaseValidationError("source archive is empty") seen_raw: set[str] = set() seen_folded: set[str] = set() normalized: list[tuple[tarfile.TarInfo, str]] = [] file_names: list[str] = [] for index, member in enumerate(members): name = _archive_member_path(member.name, label=f"archive member {index}") folded = name.casefold() if name in seen_raw or folded in seen_folded: raise ReleaseValidationError(f"duplicate or case-colliding archive member: {name}") seen_raw.add(name) seen_folded.add(folded) if not (member.isfile() or member.isdir()): raise ReleaseValidationError( f"source archive contains a link or special entry: {name} ({member.type!r})" ) if member.uid != 0 or member.gid != 0 or member.uname not in {"", "root"} or member.gname not in {"", "root"}: raise ReleaseValidationError(f"archive ownership is not deterministic root:root for {name}") if member.mtime != expected_mtime: raise ReleaseValidationError( f"archive timestamp mismatch for {name}: {member.mtime} != {expected_mtime}" ) normalized.append((member, name)) if member.isfile(): file_names.append(name) prefix = "" if set(file_names) != expected_paths: roots = {name.split("/", 1)[0] for name in file_names if "/" in name} if len(roots) != 1: raise ReleaseValidationError("archive paths do not exactly match the source manifest") candidate_prefix = next(iter(roots)) + "/" stripped_files = [ name[len(candidate_prefix) :] if name.startswith(candidate_prefix) else "" for name in file_names ] if "" in stripped_files or set(stripped_files) != expected_paths: raise ReleaseValidationError("archive paths do not exactly match the source manifest") prefix = candidate_prefix stripped_file_order = [name[len(prefix) :] for name in file_names] if stripped_file_order != sorted(stripped_file_order, key=lambda item: item.encode("utf-8")): raise ReleaseValidationError("archive files are not in canonical UTF-8 path order") expected_directories = _expected_parent_directories(expected_paths) extracted: list[tuple[Path, int]] = [] if extract_to is not None: if extract_to.exists() or extract_to.is_symlink(): raise ReleaseValidationError("archive extraction destination must not already exist") extract_to.mkdir(parents=True) observed_files: set[str] = set() for member, raw_name in normalized: if prefix: if raw_name == prefix[:-1]: if not member.isdir(): raise ReleaseValidationError("archive root prefix is not a directory") continue if not raw_name.startswith(prefix): raise ReleaseValidationError(f"archive member escapes the single root prefix: {raw_name}") relative_name = raw_name[len(prefix) :] else: relative_name = raw_name if not relative_name: continue validate_relative_path(relative_name, label="stripped archive member") if member.isdir(): if relative_name not in expected_directories: raise ReleaseValidationError(f"unexpected directory in source archive: {relative_name}") if member.mode not in {0o755, 0o775}: raise ReleaseValidationError(f"unexpected directory mode in source archive: {relative_name}") if extract_to is not None: destination = extract_to.joinpath(*PurePosixPath(relative_name).parts) destination.mkdir(parents=True, exist_ok=True) continue record = records.get(relative_name) if record is None: raise ReleaseValidationError(f"unexpected file in source archive: {relative_name}") if member.size != record["size"]: raise ReleaseValidationError(f"source archive size mismatch for {relative_name}") expected_executable = record["mode"] == "100755" if bool(member.mode & 0o111) != expected_executable or member.mode not in ( {0o755, 0o775} if expected_executable else {0o644, 0o664} ): raise ReleaseValidationError(f"source archive mode mismatch for {relative_name}") source = archive.extractfile(member) if source is None: raise ReleaseValidationError(f"cannot read archive member: {relative_name}") content = source.read() if len(content) != member.size or sha256_bytes(content) != record["sha256"]: raise ReleaseValidationError(f"source archive content digest mismatch for {relative_name}") expected_oid = record["git_blob_oid"] if _git_blob_oid(content, len(expected_oid)) != expected_oid: raise ReleaseValidationError(f"source archive Git blob identity mismatch for {relative_name}") observed_files.add(relative_name) if extract_to is not None: destination = extract_to.joinpath(*PurePosixPath(relative_name).parts) destination.parent.mkdir(parents=True, exist_ok=True) with destination.open("xb") as output: output.write(content) mode = 0o755 if expected_executable else 0o644 os.chmod(destination, mode) os.utime(destination, (expected_mtime, expected_mtime)) extracted.append((destination, mode)) if observed_files != expected_paths: raise ReleaseValidationError( f"source archive file set mismatch: missing={sorted(expected_paths - observed_files)}" ) if extract_to is not None: for directory in sorted( (path for path in extract_to.rglob("*") if path.is_dir()), key=lambda path: len(path.parts), reverse=True, ): os.chmod(directory, 0o755) os.utime(directory, (expected_mtime, expected_mtime)) os.utime(extract_to, (expected_mtime, expected_mtime)) return { "archive_sha256": sha256_file(archive_path), "archive_size_bytes": archive_path.stat().st_size, "file_count": len(expected_paths), "root_prefix": prefix, } def _run_git( repo: Path, *arguments: str, binary: bool = False, ) -> str | bytes: command = [ release_git_executable(), "--no-replace-objects", "-C", str(repo), *arguments, ] run_kwargs: dict[str, Any] = { "check": False, "capture_output": True, "env": controlled_subprocess_env(GIT_VALIDATION_ENV), "timeout": COMMAND_TIMEOUTS_SECONDS["git_validation"], } if not binary: run_kwargs.update( { "text": True, "encoding": "utf-8", "errors": "replace", } ) try: completed = subprocess.run(command, **run_kwargs) except subprocess.TimeoutExpired as error: raise ReleaseValidationError( "Git validation timed out after " f"{COMMAND_TIMEOUTS_SECONDS['git_validation']} seconds: {' '.join(arguments)}" ) from error except OSError as error: raise ReleaseValidationError(f"cannot execute Git validation: {error}") from error if completed.returncode != 0: stderr = completed.stderr stdout = completed.stdout if isinstance(stderr, bytes): stderr = stderr.decode("utf-8", "replace") if isinstance(stdout, bytes): stdout = stdout.decode("utf-8", "replace") message = str(stderr or "").strip() or str(stdout or "").strip() raise ReleaseValidationError( f"Git validation failed for {' '.join(arguments)}: {message}" ) if binary: return bytes(completed.stdout) return str(completed.stdout).strip() def validate_candidate_checkout(repo: Path, *, commit: str) -> dict[str, Any]: """Require the exact clean checkout and reject every replacement ref. All object reads already use ``--no-replace-objects``. Rejecting the refs as well makes the release evidence explicit and prevents a replacement namespace from being mistaken for part of the approved candidate state. """ if repo.is_symlink() or not repo.is_dir(): raise ReleaseValidationError(f"Git repository does not exist or is a symlink: {repo}") if not FULL_COMMIT_RE.fullmatch(commit): raise ReleaseValidationError("commit must be a full lowercase 40-character Git object id") head = _run_git(repo, "rev-parse", "--verify", "HEAD") assert isinstance(head, str) if head != commit: raise ReleaseValidationError(f"repository HEAD {head} does not equal candidate commit {commit}") replacement_refs = _run_git( repo, "for-each-ref", "--format=%(refname)", "refs/replace/", ) assert isinstance(replacement_refs, str) if replacement_refs: raise ReleaseValidationError("candidate repository contains forbidden refs/replace entries") dirty = _run_git(repo, "status", "--porcelain=v1", "-z", "-uall", binary=True) assert isinstance(dirty, bytes) if dirty: raise ReleaseValidationError("candidate repository working tree is dirty") return { "head": head, "worktree_clean": True, "replace_refs_present": False, "status_sha256": sha256_bytes(dirty), } def validate_manifest_commit_tree( repo: Path, *, commit: str, manifest: dict[str, Any], ) -> dict[str, Any]: """Bind every manifest file record to the immutable Git commit tree.""" if repo.is_symlink() or not repo.is_dir(): raise ReleaseValidationError(f"Git repository does not exist or is a symlink: {repo}") if not FULL_COMMIT_RE.fullmatch(commit): raise ReleaseValidationError("commit must be a full lowercase 40-character Git object id") raw = _run_git(repo, "ls-tree", "-r", "-z", "-l", commit, binary=True) assert isinstance(raw, bytes) actual: list[dict[str, Any]] = [] for index, record in enumerate(raw.split(b"\0")): if not record: continue try: metadata, raw_path = record.split(b"\t", 1) mode, object_type, object_id, raw_size = metadata.decode("ascii").split() path = raw_path.decode("utf-8", "strict") except (UnicodeError, ValueError) as error: raise ReleaseValidationError( f"Git tree entry {index} is malformed or not UTF-8" ) from error if object_type != "blob" or raw_size == "-": raise ReleaseValidationError( f"Git tree contains unsupported entry {path!r}: {object_type} {raw_size}" ) validate_relative_path(path, label=f"Git tree path {index}") try: size = int(raw_size) except ValueError as error: raise ReleaseValidationError(f"Git tree size is invalid for {path}") from error actual.append( { "path": path, "mode": mode, "git_blob_oid": object_id, "size": size, } ) actual.sort(key=lambda item: item["path"].encode("utf-8")) expected = [ { "path": item["path"], "mode": item["mode"], "git_blob_oid": item["git_blob_oid"], "size": item["size"], } for item in manifest["files"]["items"] ] if actual != expected: actual_by_path = {item["path"]: item for item in actual} expected_by_path = {item["path"]: item for item in expected} missing = sorted(set(actual_by_path) - set(expected_by_path)) extra = sorted(set(expected_by_path) - set(actual_by_path)) changed = sorted( path for path in set(actual_by_path) & set(expected_by_path) if actual_by_path[path] != expected_by_path[path] ) raise ReleaseValidationError( "source manifest is not the candidate Git tree: " f"missing_from_manifest={missing}, extra_in_manifest={extra}, changed={changed}" ) return { "entry_count": len(actual), "ls_tree_sha256": sha256_bytes(raw), } def validate_tag_binding(repo: Path, *, tag: str, commit: str, expected_tree: str) -> dict[str, str]: if repo.is_symlink() or not repo.is_dir(): raise ReleaseValidationError(f"Git repository does not exist: {repo}") if not FULL_COMMIT_RE.fullmatch(commit): raise ReleaseValidationError("commit must be a full lowercase 40-character Git object id") if ( not TAG_RE.fullmatch(tag) or tag.endswith((".", "/")) or ".." in tag or "//" in tag or "@{" in tag or "/." in tag ): raise ReleaseValidationError(f"tag name is unsafe or ambiguous: {tag!r}") def git(*arguments: str) -> str: result = _run_git(repo, *arguments) assert isinstance(result, str) return result ref = f"refs/tags/{tag}" tag_object_oid = git("rev-parse", "--verify", ref) if not GIT_OBJECT_RE.fullmatch(tag_object_oid): raise ReleaseValidationError("annotated tag object id is invalid") if git("show-ref", "--verify", ref) != f"{tag_object_oid} {ref}": raise ReleaseValidationError("tag ref identity is ambiguous") object_type = git("cat-file", "-t", ref) if object_type != "tag": raise ReleaseValidationError("release tag must be an annotated tag object") actual_commit = git("rev-parse", "--verify", f"{ref}^{{commit}}") if actual_commit != commit: raise ReleaseValidationError(f"tag {tag} resolves to {actual_commit}, not {commit}") direct_commit = git("rev-parse", "--verify", f"{commit}^{{commit}}") if direct_commit != commit: raise ReleaseValidationError("explicit commit does not resolve exactly") tree = git("rev-parse", "--verify", f"{commit}^{{tree}}") if tree != expected_tree: raise ReleaseValidationError(f"Git tree {tree} does not match source manifest tree {expected_tree}") return { "tag": tag, "tag_object_oid": tag_object_oid, "commit": actual_commit, "tree": tree, "tag_object_type": object_type, } def ensure_regular_file_under(root: Path, relative_path: str) -> Path: pure = validate_relative_path(relative_path, label="artifact path") if root.is_symlink() or not root.is_dir(): raise ReleaseValidationError("artifact root must be a real directory, not a symlink") current = root for part in pure.parts: current = current / part if current.is_symlink(): raise ReleaseValidationError(f"artifact path traverses a symlink: {relative_path}") if not current.is_file(): raise ReleaseValidationError(f"artifact is missing or not a regular file: {relative_path}") try: current.resolve(strict=True).relative_to(root.resolve(strict=True)) except (OSError, ValueError) as error: raise ReleaseValidationError(f"artifact escapes its root: {relative_path}") from error return current def verify_file_record(root: Path, record: Any, *, label: str) -> Path: item = _require_mapping(record, label) for key in ("path", "sha256", "size_bytes"): if key not in item: raise ReleaseValidationError(f"{label} is missing {key}") if not isinstance(item["sha256"], str) or not SHA256_RE.fullmatch(item["sha256"]): raise ReleaseValidationError(f"{label} has an invalid SHA-256") if not isinstance(item["size_bytes"], int) or isinstance(item["size_bytes"], bool) or item["size_bytes"] < 0: raise ReleaseValidationError(f"{label} has an invalid size") path = ensure_regular_file_under(root, item["path"]) if path.stat().st_size != item["size_bytes"]: raise ReleaseValidationError(f"{label} size drift: {item['path']}") if sha256_file(path) != item["sha256"]: raise ReleaseValidationError(f"{label} digest drift: {item['path']}") return path def validate_spdx(payload: Any, policy: dict[str, Any]) -> dict[str, Any]: document = _require_mapping(payload, "SPDX SBOM") if document.get("spdxVersion") != policy["sbom"]["spdx_version"]: raise ReleaseValidationError("SBOM SPDX version mismatch") if document.get("SPDXID") != "SPDXRef-DOCUMENT": raise ReleaseValidationError("SBOM is missing SPDXRef-DOCUMENT identity") namespace = document.get("documentNamespace") if not isinstance(namespace, str) or not namespace: raise ReleaseValidationError("SBOM document namespace is missing") creation = _require_mapping(document.get("creationInfo"), "SPDX creationInfo") creators = _require_list(creation.get("creators"), "SPDX creationInfo.creators") if not any(isinstance(item, str) and item.casefold().startswith("tool: syft") for item in creators): raise ReleaseValidationError("SBOM does not identify Syft as a creator") packages = _require_list(document.get("packages"), "SPDX packages") if not packages: raise ReleaseValidationError("SBOM package inventory is empty") return {"spdx_version": document["spdxVersion"], "package_count": len(packages)} def validate_trivy_scan( payload: Any, policy: dict[str, Any], *, expected_config_digest: str | None = None, expected_artifact_basename: str | None = None, role: str | None = None, at: datetime | None = None, ) -> dict[str, Any]: report = _require_mapping(payload, "Trivy scan report") schema_version = report.get("SchemaVersion") if not isinstance(schema_version, int) or isinstance(schema_version, bool) or schema_version < 2: raise ReleaseValidationError("Trivy scan SchemaVersion is missing or unsupported") results = _require_list(report.get("Results"), "Trivy scan Results") if not results: raise ReleaseValidationError("Trivy scan has no Results and cannot prove an image was scanned") artifact_name = report.get("ArtifactName") if not isinstance(artifact_name, str) or not artifact_name: raise ReleaseValidationError("Trivy scan ArtifactName is missing") if expected_artifact_basename is not None and PurePosixPath( artifact_name.replace("\\", "/") ).name != expected_artifact_basename: raise ReleaseValidationError("Trivy scan ArtifactName is not bound to the OCI archive") metadata = _require_mapping(report.get("Metadata"), "Trivy scan Metadata") image_id = metadata.get("ImageID") if ( not isinstance(image_id, str) or not image_id.startswith("sha256:") or not SHA256_RE.fullmatch(image_id[7:]) ): raise ReleaseValidationError("Trivy scan Metadata.ImageID is missing or invalid") if expected_config_digest is not None and image_id != expected_config_digest: raise ReleaseValidationError("Trivy scan image identity differs from the OCI config digest") thresholds = policy["vulnerability_scan"]["thresholds"] evaluated_at = (at or datetime.now(timezone.utc)).astimezone(timezone.utc) os_metadata = metadata.get("OS") os_family = os_metadata.get("Family") if isinstance(os_metadata, dict) else None waiver_matchers: list[tuple[dict[str, Any], set[tuple[str, str, str]]]] = [] for waiver in policy["vulnerability_scan"]["exemptions"]: finding_keys = { ( finding["vulnerability_id"], package_name, finding["installed_version"], ) for finding in waiver["findings"] for package_name in finding["package_names"] } waiver_matchers.append((waiver, finding_keys)) total = {severity: 0 for severity in thresholds} exempted = {severity: 0 for severity in thresholds} blocking = {severity: 0 for severity in thresholds} applied_waiver_counts: dict[str, int] = {} consumed_waiver_findings: set[tuple[str, str, str, str]] = set() for result_index, raw_result in enumerate(results): result = _require_mapping(raw_result, f"Trivy Results[{result_index}]") vulnerabilities = result.get("Vulnerabilities", []) if vulnerabilities is None: vulnerabilities = [] vulnerabilities = _require_list(vulnerabilities, f"Trivy Results[{result_index}].Vulnerabilities") for vulnerability_index, raw_vulnerability in enumerate(vulnerabilities): vulnerability = _require_mapping( raw_vulnerability, f"Trivy vulnerability {result_index}/{vulnerability_index}", ) severity = vulnerability.get("Severity") if severity not in thresholds: raise ReleaseValidationError( f"Trivy report contains severity outside the requested fail-closed set: {severity!r}" ) total[severity] += 1 vulnerability_id = vulnerability.get("VulnerabilityID") package_name = vulnerability.get("PkgName") installed_version = vulnerability.get("InstalledVersion") finding_key = ( (vulnerability_id, package_name, installed_version) if all( isinstance(value, str) for value in (vulnerability_id, package_name, installed_version) ) else None ) fixed_version = vulnerability.get("FixedVersion") matched_waiver: dict[str, Any] | None = None for waiver, allowed_findings in waiver_matchers: scope = waiver["scope"] expiry = parse_rfc3339( waiver["expires_at"], label=f"vulnerability waiver {waiver['id']} expiry" ) consumed_key = ( (waiver["id"], *finding_key) if finding_key is not None else None ) if ( evaluated_at < expiry and role in scope["roles"] and result.get("Class") == scope["class"] and result.get("Type") == scope["type"] and os_family == scope["os_family"] and severity == scope["severity"] and scope["fixed_version_must_be_empty"] is True and (fixed_version is None or fixed_version == "") and finding_key is not None and finding_key in allowed_findings and consumed_key is not None and consumed_key not in consumed_waiver_findings ): matched_waiver = waiver consumed_waiver_findings.add(consumed_key) break if matched_waiver is None: blocking[severity] += 1 else: exempted[severity] += 1 waiver_id = matched_waiver["id"] applied_waiver_counts[waiver_id] = applied_waiver_counts.get(waiver_id, 0) + 1 passed = all(blocking[severity] <= limit for severity, limit in thresholds.items()) if not passed: raise ReleaseValidationError( "Trivy vulnerability gate failed: " f"total={total}, exempted={exempted}, blocking={blocking}" ) return { "schema_version": schema_version, "artifact_name": artifact_name, "image_id": image_id, "counts": { "total": total, "exempted": exempted, "blocking": blocking, }, "waiver_ids": sorted(applied_waiver_counts), "waiver_exempted_counts": { waiver_id: applied_waiver_counts[waiver_id] for waiver_id in sorted(applied_waiver_counts) }, "waiver_evaluated_at": format_rfc3339(evaluated_at), "passed": True, } def validate_trivy_db_metadata( payload: Any, policy: dict[str, Any], *, at: datetime, ) -> dict[str, Any]: metadata = _require_mapping(payload, "Trivy version/database metadata") version = metadata.get("Version") if version != REQUIRED_TOOL_VERSIONS["trivy"]: raise ReleaseValidationError( f"Trivy metadata version {version!r} is not pinned {REQUIRED_TOOL_VERSIONS['trivy']}" ) database = metadata.get("VulnerabilityDB") database = _require_mapping(database, "Trivy VulnerabilityDB metadata") database_version = database.get("Version") expected_database_version = policy["vulnerability_scan"]["database"][ "repository_transport" ]["schema_version"] if database_version != expected_database_version or isinstance(database_version, bool): raise ReleaseValidationError( "Trivy VulnerabilityDB.Version is missing or differs from the repository schema" ) updated = parse_rfc3339(database.get("UpdatedAt"), label="Trivy VulnerabilityDB.UpdatedAt") next_update = parse_rfc3339(database.get("NextUpdate"), label="Trivy VulnerabilityDB.NextUpdate") downloaded = parse_rfc3339(database.get("DownloadedAt"), label="Trivy VulnerabilityDB.DownloadedAt") moment = at.astimezone(timezone.utc) if updated > moment + timedelta(minutes=5) or downloaded > moment + timedelta(minutes=5): raise ReleaseValidationError("Trivy database metadata is implausibly in the future") maximum_age = timedelta(hours=policy["vulnerability_scan"]["database"]["maximum_age_hours"]) if moment - updated > maximum_age: raise ReleaseValidationError("Trivy vulnerability database is older than the allowed maximum age") if policy["vulnerability_scan"]["database"]["require_next_update_after_scan"] and next_update <= moment: raise ReleaseValidationError("Trivy vulnerability database is expired") if downloaded < updated - timedelta(minutes=5): raise ReleaseValidationError("Trivy database download predates its update metadata") return { "version": database_version, "updated_at": format_rfc3339(updated), "next_update": format_rfc3339(next_update), "downloaded_at": format_rfc3339(downloaded), "validated_at": format_rfc3339(moment), } def _read_tar_member(archive: tarfile.TarFile, name: str, *, label: str) -> bytes: try: member = archive.getmember(name) except KeyError as error: raise ReleaseValidationError(f"{label} is missing from OCI archive: {name}") from error if not member.isfile(): raise ReleaseValidationError(f"{label} is not a regular OCI archive member: {name}") source = archive.extractfile(member) if source is None: raise ReleaseValidationError(f"cannot read {label}: {name}") return source.read() def _hash_tar_member(archive: tarfile.TarFile, name: str, *, label: str) -> tuple[int, str]: """Hash one regular member without buffering an image layer in memory.""" try: member = archive.getmember(name) except KeyError as error: raise ReleaseValidationError(f"{label} is missing from archive: {name}") from error if not member.isfile(): raise ReleaseValidationError(f"{label} is not a regular archive member: {name}") source = archive.extractfile(member) if source is None: raise ReleaseValidationError(f"cannot read {label}: {name}") digest = hashlib.sha256() size = 0 while chunk := source.read(1024 * 1024): digest.update(chunk) size += len(chunk) if size != member.size: raise ReleaseValidationError(f"{label} is truncated: {name}") return size, digest.hexdigest() def _docker_layer_diff_id( archive: tarfile.TarFile, name: str, *, label: str, raw_digest: str, ) -> str: """Return a Docker layer's uncompressed digest for config diff-id binding.""" member = archive.getmember(name) source = archive.extractfile(member) if source is None: raise ReleaseValidationError(f"cannot read {label}: {name}") magic = source.read(4) source.close() if magic.startswith(b"\x28\xb5\x2f\xfd"): raise ReleaseValidationError( f"{label} uses unsupported zstd compression and cannot be verified fail closed" ) if not magic.startswith(b"\x1f\x8b"): return raw_digest source = archive.extractfile(member) if source is None: raise ReleaseValidationError(f"cannot re-read {label}: {name}") digest = hashlib.sha256() try: with gzip.GzipFile(fileobj=source, mode="rb") as uncompressed: while chunk := uncompressed.read(1024 * 1024): digest.update(chunk) except (EOFError, OSError) as error: raise ReleaseValidationError(f"{label} gzip payload is invalid: {name}") from error return digest.hexdigest() def _verify_oci_descriptor( archive: tarfile.TarFile, descriptor: Any, *, label: str, read_content: bool = True, ) -> tuple[dict[str, Any], bytes | None]: item = _require_mapping(descriptor, label) digest = item.get("digest") if not isinstance(digest, str) or not digest.startswith("sha256:") or not SHA256_RE.fullmatch(digest[7:]): raise ReleaseValidationError(f"{label} has an invalid digest") size = item.get("size") if not isinstance(size, int) or isinstance(size, bool) or size < 0: raise ReleaseValidationError(f"{label} has an invalid size") name = f"blobs/sha256/{digest[7:]}" actual_size, actual_digest = _hash_tar_member(archive, name, label=label) if actual_size != size or actual_digest != digest[7:]: raise ReleaseValidationError(f"{label} content digest or size mismatch") data = _read_tar_member(archive, name, label=label) if read_content else None return item, data def _image_config_summary(config_data: bytes, *, label: str) -> dict[str, Any]: config_payload = _require_mapping(load_json_bytes(config_data, label=label), label) if config_payload.get("os") != "linux" or config_payload.get("architecture") != "amd64": raise ReleaseValidationError(f"{label} platform is not linux/amd64") runtime_config = _require_mapping(config_payload.get("config"), f"{label} runtime config") labels = _require_mapping(runtime_config.get("Labels"), f"{label} labels") required_labels = { "revision": labels.get("org.opencontainers.image.revision"), "version": labels.get("org.opencontainers.image.version"), "release_role": labels.get("io.yuqei.release.role"), "title": labels.get("org.opencontainers.image.title"), } if not all(isinstance(value, str) and value for value in required_labels.values()): raise ReleaseValidationError(f"{label} lacks required candidate/role labels") rootfs = _require_mapping(config_payload.get("rootfs"), f"{label} rootfs") if rootfs.get("type") != "layers": raise ReleaseValidationError(f"{label} rootfs type is not layers") diff_ids = _require_list(rootfs.get("diff_ids"), f"{label} rootfs.diff_ids") if not diff_ids or not all( isinstance(value, str) and value.startswith("sha256:") and SHA256_RE.fullmatch(value[7:]) for value in diff_ids ): raise ReleaseValidationError(f"{label} rootfs.diff_ids is empty or invalid") return { **required_labels, "rootfs_diff_ids": diff_ids, "rootfs_identity_sha256": sha256_bytes(canonical_json_bytes(diff_ids)), } def inspect_oci_archive(path: Path) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise ReleaseValidationError("OCI archive must be a regular non-symlink file") _validate_tar_stream_termination(path) try: archive = tarfile.open(path, mode="r:*") except (OSError, tarfile.TarError) as error: raise ReleaseValidationError(f"OCI archive is unreadable: {error}") from error with archive: seen: set[str] = set() seen_folded: set[str] = set() for index, member in enumerate(archive.getmembers()): name = _archive_member_path(member.name, label=f"OCI archive member {index}") if name in seen or name.casefold() in seen_folded: raise ReleaseValidationError(f"duplicate or case-colliding OCI member: {name}") seen.add(name) seen_folded.add(name.casefold()) if not (member.isfile() or member.isdir()): raise ReleaseValidationError(f"OCI archive contains a link or special entry: {name}") layout = _require_mapping( load_json_bytes(_read_tar_member(archive, "oci-layout", label="OCI layout"), label="OCI layout"), "OCI layout", ) if layout != {"imageLayoutVersion": "1.0.0"}: raise ReleaseValidationError("unsupported OCI image layout version") index = _require_mapping( load_json_bytes(_read_tar_member(archive, "index.json", label="OCI index"), label="OCI index"), "OCI index", ) descriptors = _require_list(index.get("manifests"), "OCI index manifests") if len(descriptors) != 1: raise ReleaseValidationError("OCI index must contain exactly one top-level descriptor") descriptor, manifest_data = _verify_oci_descriptor( archive, descriptors[0], label="OCI top-level descriptor" ) assert manifest_data is not None media_type = descriptor.get("mediaType") if media_type in { "application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.list.v2+json", }: raise ReleaseValidationError( "OCI top-level descriptor must reference one image manifest directly, not an image index" ) platform = descriptor.get("platform") if isinstance(platform, dict) and ( platform.get("os") != "linux" or platform.get("architecture") != "amd64" ): raise ReleaseValidationError("OCI descriptor platform is not linux/amd64") if media_type not in { "application/vnd.oci.image.manifest.v1+json", "application/vnd.docker.distribution.manifest.v2+json", }: raise ReleaseValidationError(f"unsupported OCI image manifest media type: {media_type!r}") manifest = _require_mapping(load_json_bytes(manifest_data, label="OCI image manifest"), "OCI image manifest") config, config_data = _verify_oci_descriptor( archive, manifest.get("config"), label="OCI config descriptor" ) assert config_data is not None config_summary = _image_config_summary(config_data, label="OCI image config") layers = _require_list(manifest.get("layers"), "OCI image layers") if not layers: raise ReleaseValidationError("OCI image has no layers") if len(layers) != len(config_summary["rootfs_diff_ids"]): raise ReleaseValidationError("OCI layer count differs from config rootfs.diff_ids") for index, (layer, expected_diff_id) in enumerate( zip(layers, config_summary["rootfs_diff_ids"], strict=True) ): verified_layer, _unused = _verify_oci_descriptor( archive, layer, label=f"OCI layer {index}", read_content=False ) layer_digest = verified_layer["digest"] layer_name = f"blobs/sha256/{layer_digest[7:]}" actual_diff_id = _docker_layer_diff_id( archive, layer_name, label=f"OCI layer {index}", raw_digest=layer_digest[7:], ) if f"sha256:{actual_diff_id}" != expected_diff_id: raise ReleaseValidationError( f"OCI layer {index} content differs from config rootfs.diff_ids" ) return { "manifest_digest": descriptor["digest"], "manifest_media_type": media_type, "config_digest": config["digest"], **config_summary, "layer_count": len(layers), } def inspect_docker_archive(path: Path, *, expected_repo_tag: str | None = None) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise ReleaseValidationError("Docker archive must be a regular non-symlink file") _validate_tar_stream_termination(path) try: archive = tarfile.open(path, mode="r:") except (OSError, tarfile.TarError) as error: raise ReleaseValidationError(f"Docker archive is unreadable: {error}") from error with archive: seen: set[str] = set() seen_folded: set[str] = set() for index, member in enumerate(archive.getmembers()): name = _archive_member_path(member.name, label=f"Docker archive member {index}") if name in seen or name.casefold() in seen_folded: raise ReleaseValidationError(f"duplicate or case-colliding Docker archive member: {name}") seen.add(name) seen_folded.add(name.casefold()) if not (member.isfile() or member.isdir()): raise ReleaseValidationError( f"Docker archive contains a link or special entry: {name}" ) manifest_payload = load_json_bytes( _read_tar_member(archive, "manifest.json", label="Docker archive manifest"), label="Docker archive manifest", ) manifests = _require_list(manifest_payload, "Docker archive manifest") if len(manifests) != 1: raise ReleaseValidationError("Docker archive must contain exactly one image") manifest = _require_mapping(manifests[0], "Docker archive image manifest") config_name = manifest.get("Config") if not isinstance(config_name, str): raise ReleaseValidationError("Docker archive manifest has no Config path") validate_relative_path(config_name, label="Docker archive Config path") config_data = _read_tar_member(archive, config_name, label="Docker image config") config_sha256 = sha256_bytes(config_data) if config_name not in { f"{config_sha256}.json", f"blobs/sha256/{config_sha256}", }: raise ReleaseValidationError("Docker archive config filename does not match its digest") config_summary = _image_config_summary(config_data, label="Docker image config") repo_tags = _require_list(manifest.get("RepoTags"), "Docker archive RepoTags") if not repo_tags or not all(isinstance(item, str) and item for item in repo_tags): raise ReleaseValidationError("Docker archive has no usable repository tag") if len(repo_tags) != len(set(repo_tags)): raise ReleaseValidationError("Docker archive has duplicate repository tags") if expected_repo_tag is not None and repo_tags != [expected_repo_tag]: raise ReleaseValidationError( f"Docker archive repository tag drift: {repo_tags!r} != {[expected_repo_tag]!r}" ) layers = _require_list(manifest.get("Layers"), "Docker archive Layers") if not layers or not all(isinstance(item, str) for item in layers): raise ReleaseValidationError("Docker archive has no layers") if len(layers) != len(set(layers)): raise ReleaseValidationError("Docker archive has duplicate layer paths") if len(layers) != len(config_summary["rootfs_diff_ids"]): raise ReleaseValidationError( "Docker archive layer count differs from config rootfs.diff_ids" ) for index, (layer_name, expected_diff_id) in enumerate( zip(layers, config_summary["rootfs_diff_ids"], strict=True) ): validate_relative_path(layer_name, label=f"Docker layer path {index}") _, raw_digest = _hash_tar_member( archive, layer_name, label=f"Docker layer {index}" ) layer_path = PurePosixPath(layer_name) is_digest_blob = ( len(layer_path.parts) == 3 and layer_path.parts[:2] == ("blobs", "sha256") ) if not is_digest_blob and layer_path.name != "layer.tar": raise ReleaseValidationError( f"Docker layer {index} path is not a supported loadable archive layout" ) if is_digest_blob and layer_path.name != raw_digest: raise ReleaseValidationError( f"Docker layer {index} blob path differs from its compressed digest" ) actual_diff_id = _docker_layer_diff_id( archive, layer_name, label=f"Docker layer {index}", raw_digest=raw_digest, ) if f"sha256:{actual_diff_id}" != expected_diff_id: raise ReleaseValidationError( f"Docker layer {index} content differs from config rootfs.diff_ids" ) return { "config_digest": f"sha256:{config_sha256}", "repo_tags": repo_tags, **config_summary, "layer_count": len(layers), } def validate_receipt_payload_digest(receipt: Any) -> dict[str, Any]: result = _require_mapping(receipt, "build receipt") if result.get("schema_version") != BUILD_RECEIPT_SCHEMA: raise ReleaseValidationError("unsupported build receipt schema") digest = result.get("receipt_payload_sha256") if not isinstance(digest, str) or not SHA256_RE.fullmatch(digest): raise ReleaseValidationError("build receipt payload digest is invalid") unsigned = dict(result) del unsigned["receipt_payload_sha256"] if sha256_bytes(canonical_json_bytes(unsigned)[:-1]) != digest: raise ReleaseValidationError("build receipt payload digest mismatch") return result def with_payload_digest(payload: dict[str, Any], field: str) -> dict[str, Any]: result = dict(payload) result[field] = sha256_bytes(canonical_json_bytes(payload)[:-1]) return result