#!/usr/bin/env python3 """Build and validate the immutable GCP leoclean no-send service package.""" from __future__ import annotations import argparse import hashlib import json import re import shutil import stat import subprocess import sys from collections.abc import Mapping from pathlib import Path from typing import Any IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v1" RELEASE_SCHEMA = "livingip.leocleanNoSendRelease.v1" ARTIFACT_MANIFEST_SCHEMA = "livingip.leocleanNoSendArtifactManifest.v1" ARTIFACT_RECEIPT_SCHEMA = "livingip.leocleanNoSendArtifactVerification.v1" IDENTITY_MANIFEST_SCHEMA = "livingip.leoIdentityManifest.v1" IDENTITY_LOCK_SCHEMA = "livingip.leoIdentityProfileLock.v1" PROJECT = "teleo-501523" ZONE = "europe-west6-a" INSTANCE = "teleo-staging-1" SERVICE = "leoclean-gcp-nosend.service" SERVICE_ACCOUNT = "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com" PLATFORM = "linux/amd64" IMAGE_REPOSITORY = "europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging" CONTAINER_NAME = "livingip-leoclean-nosend" HERMES_COMMIT = "b2f477a30b3c05d0f383c543af98496ae8a96070" PYTHON_VERSION = "3.11.9" UV_VERSION = "0.9.30" POSTGRES_VERSION = "16.14" PYTHON_IMAGE = ( "docker.io/library/python:3.11.9-slim-bookworm@" "sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317" ) UV_IMAGE = "ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4" POSTGRES_IMAGE = ( "docker.io/library/postgres:16.14-bookworm@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55" ) CA_SHA256 = "80701e768f0e1f6b9d621aa0b53f6e851daaa276c6d9a8e51a300fbc015539cb" CA_DESTINATION = "/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem" RUNTIME_ROOT = "/opt/livingip/leoclean-nosend" PROFILE_ROOT = "/run/leoclean-profile" RUNTIME_UID = 65532 RUNTIME_GID = 65532 HEX_40 = re.compile(r"^[0-9a-f]{40}$") HEX_64 = re.compile(r"^[0-9a-f]{64}$") IMAGE_REFERENCE = re.compile(rf"^{re.escape(IMAGE_REPOSITORY)}@sha256:([0-9a-f]{{64}})$") FORBIDDEN_DESCRIPTOR_MARKERS = ( "77.42.65.182", "gcp-teleo-pgvector-standby-postgres-password", "leoclean-gateway.service", "sa-teleo-prod-vm", "telegram_bot_token", "teleo-prod-1", ) IDENTITY_FILES = ("SOUL.md", "identity-lock.json", "identity-manifest.json") class PackageError(RuntimeError): """The image or service package does not satisfy the staging contract.""" def _require(condition: bool, message: str) -> None: if not condition: raise PackageError(message) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def canonical_sha256(value: Any) -> str: encoded = json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def load_json(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise PackageError(f"cannot read {label}: {type(exc).__name__}") from exc _require(isinstance(value, dict), f"{label} must be a JSON object") return value def write_json(path: Path, value: Mapping[str, Any], mode: int = 0o644) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") path.chmod(mode) def _is_sha256(value: Any) -> bool: return isinstance(value, str) and bool(HEX_64.fullmatch(value)) def _is_revision(value: Any) -> bool: return isinstance(value, str) and bool(HEX_40.fullmatch(value)) def _safe_tree(root: Path, label: str) -> None: _require(root.is_dir() and not root.is_symlink(), f"{label} root is missing or unsafe") for path in root.rglob("*"): _require(not path.is_symlink(), f"{label} contains a symlink") _require(path.is_dir() or path.is_file(), f"{label} contains an unsupported entry") def tree_manifest(root: Path, *, excluded: frozenset[str] = frozenset()) -> dict[str, Any]: _safe_tree(root, "tree") files: list[dict[str, Any]] = [] for path in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()): relative = path.relative_to(root).as_posix() if relative in excluded or path.is_dir(): continue mode = stat.S_IMODE(path.stat().st_mode) files.append( { "path": relative, "bytes": path.stat().st_size, "mode": f"0o{mode:03o}", "sha256": sha256_file(path), } ) stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} return {**stable, "sha256": canonical_sha256(stable)} def identity_bundle_manifest(root: Path) -> dict[str, Any]: _safe_tree(root, "identity bundle") actual = sorted(path.name for path in root.iterdir()) _require(actual == sorted(IDENTITY_FILES), "identity bundle entry set is not exact") files = [] for name in IDENTITY_FILES: path = root / name _require(path.is_file() and not path.is_symlink(), f"identity file is unsafe: {name}") _require(path.stat().st_size <= 1024 * 1024, f"identity file is too large: {name}") _require(not stat.S_IMODE(path.stat().st_mode) & 0o022, f"identity file is group/world writable: {name}") files.append({"path": name, "bytes": path.stat().st_size, "sha256": sha256_file(path)}) stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} return {**stable, "sha256": canonical_sha256(stable)} def validate_identity_bundle(root: Path) -> dict[str, Any]: bundle = identity_bundle_manifest(root) manifest = load_json(root / "identity-manifest.json", "identity manifest") lock = load_json(root / "identity-lock.json", "identity lock") _require( set(manifest) == {"schema", "identity_inputs", "identity_inputs_sha256", "compiled_views", "manifest_sha256"}, "identity manifest fields are not exact", ) _require(manifest.get("schema") == IDENTITY_MANIFEST_SCHEMA, "identity manifest schema is invalid") _require(_is_sha256(manifest.get("manifest_sha256")), "identity manifest hash is invalid") _require(_is_sha256(manifest.get("identity_inputs_sha256")), "identity input hash is invalid") stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"} _require(canonical_sha256(stable) == manifest["manifest_sha256"], "identity manifest self-hash drifted") views = manifest.get("compiled_views") _require( isinstance(views, dict) and set(views) == {"SOUL.md", "identity.json"}, "compiled identity views are incomplete" ) soul = views["SOUL.md"] if isinstance(views, dict) else None _require(isinstance(soul, dict) and _is_sha256(soul.get("sha256")), "SOUL identity binding is invalid") _require(soul["sha256"] == sha256_file(root / "SOUL.md"), "SOUL content differs from identity manifest") _require( set(lock) == {"schema", "manifest_sha256", "identity_inputs_sha256", "compiled_views", "session_boundary"}, "identity lock fields are not exact", ) _require(lock.get("schema") == IDENTITY_LOCK_SCHEMA, "identity lock schema is invalid") _require(lock.get("manifest_sha256") == manifest["manifest_sha256"], "identity lock manifest binding drifted") _require( lock.get("identity_inputs_sha256") == manifest["identity_inputs_sha256"], "identity lock input binding drifted", ) _require( isinstance(lock.get("compiled_views"), dict) and (lock["compiled_views"].get("SOUL.md") or {}).get("sha256") == soul["sha256"], "identity lock SOUL binding drifted", ) inputs = manifest.get("identity_inputs") database = inputs.get("canonical_database") if isinstance(inputs, dict) else None _require(isinstance(database, dict), "identity database authority is missing") authority = database.get("authority") canonical = database.get("canonical_authority_granted") _require( (authority, canonical) in { ("synthetic_noncanonical_fixture", False), ("canonical_database_same_transaction_export", True), }, "identity database authority is invalid", ) return { "bundle_sha256": bundle["sha256"], "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], "soul_sha256": soul["sha256"], "authority": authority, "canonical_authority_granted": canonical, } def validate_artifact(artifact: Path, receipt_path: Path, expected_revision: str) -> dict[str, Any]: _require(_is_revision(expected_revision), "expected Teleo revision is invalid") _safe_tree(artifact, "runtime artifact") manifest = load_json(artifact / "artifact-manifest.json", "artifact manifest") receipt = load_json(receipt_path, "artifact verification receipt") _require(manifest.get("schema") == ARTIFACT_MANIFEST_SCHEMA, "artifact manifest schema is invalid") _require(receipt.get("schema") == ARTIFACT_RECEIPT_SCHEMA, "artifact receipt schema is invalid") _require(receipt.get("status") == "pass", "artifact receipt did not pass") _require(receipt.get("verification_mode") == "release", "artifact receipt is not release evidence") _require(manifest.get("teleo_git_head") == expected_revision, "artifact revision differs from source revision") _require(receipt.get("teleo_git_head") == expected_revision, "receipt revision differs from source revision") content = tree_manifest(artifact, excluded=frozenset({"artifact-manifest.json"})) _require(manifest.get("content") == content, "artifact content manifest drifted") _require(receipt.get("artifact_sha256") == content["sha256"], "artifact receipt content hash drifted") _require(manifest.get("hermes_commit") == HERMES_COMMIT, "artifact Hermes commit drifted") _require(receipt.get("hermes_commit") == HERMES_COMMIT, "receipt Hermes commit drifted") runtime_lock = load_json(artifact / "runtime-lock.json", "artifact runtime lock") _require( runtime_lock.get("base_images") == {"python": PYTHON_IMAGE.removeprefix("docker.io/library/"), "uv": UV_IMAGE}, "artifact base-image lock drifted", ) hermes = runtime_lock.get("hermes") _require(isinstance(hermes, dict) and hermes.get("commit") == HERMES_COMMIT, "artifact Hermes lock drifted") claim_ceiling = manifest.get("claim_ceiling") _require(isinstance(claim_ceiling, str) and claim_ceiling == receipt.get("claim_ceiling"), "claim ceiling drifted") return { "teleo_git_head": expected_revision, "artifact_sha256": content["sha256"], "artifact_manifest_sha256": sha256_file(artifact / "artifact-manifest.json"), "artifact_receipt_sha256": sha256_file(receipt_path), "hermes_commit": HERMES_COMMIT, "hermes_source_sha256": receipt.get("hermes_source_sha256"), "uv_lock_sha256": hermes.get("uv_lock_sha256"), "claim_ceiling": claim_ceiling, } def _target() -> dict[str, str]: return { "project": PROJECT, "zone": ZONE, "instance": INSTANCE, "service": SERVICE, "service_account": SERVICE_ACCOUNT, "platform": PLATFORM, } def _base_images() -> dict[str, str]: return {"python": PYTHON_IMAGE, "uv": UV_IMAGE, "postgres": POSTGRES_IMAGE} def _assert_no_forbidden_descriptor_markers(value: Mapping[str, Any]) -> None: encoded = json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).casefold() for marker in FORBIDDEN_DESCRIPTOR_MARKERS: _require( marker.casefold() not in encoded, "release descriptor contains a forbidden production/transport marker" ) def validate_image_input(value: dict[str, Any]) -> None: _require( set(value) == { "schema", "target", "runtime", "identity", "base_images", "source_bindings", "claim_ceiling", "input_sha256", }, "image input fields are not exact", ) _require(value.get("schema") == IMAGE_INPUT_SCHEMA, "image input schema is invalid") _require(value.get("target") == _target(), "image input target is not exact GCP staging") _require(value.get("base_images") == _base_images(), "image input base-image closure drifted") runtime = value.get("runtime") _require(isinstance(runtime, dict), "image runtime binding is missing") _require( set(runtime) == { "teleo_git_head", "artifact_sha256", "artifact_manifest_sha256", "artifact_receipt_sha256", "hermes_commit", "hermes_source_sha256", "uv_lock_sha256", "python_version", "uv_version", "postgres_version", "cloudsql_ca_sha256", }, "image runtime binding fields are not exact", ) _require(_is_revision(runtime.get("teleo_git_head")), "image Teleo revision is invalid") for field in ( "artifact_sha256", "artifact_manifest_sha256", "artifact_receipt_sha256", "hermes_source_sha256", "uv_lock_sha256", "cloudsql_ca_sha256", ): _require(_is_sha256(runtime.get(field)), f"image runtime {field} is invalid") _require(runtime.get("hermes_commit") == HERMES_COMMIT, "image Hermes commit drifted") _require(runtime.get("python_version") == PYTHON_VERSION, "image Python version drifted") _require(runtime.get("uv_version") == UV_VERSION, "image uv version drifted") _require(runtime.get("postgres_version") == POSTGRES_VERSION, "image PostgreSQL version drifted") _require(runtime.get("cloudsql_ca_sha256") == CA_SHA256, "image Cloud SQL CA drifted") identity = value.get("identity") _require( isinstance(identity, dict) and set(identity) == { "bundle_sha256", "manifest_sha256", "identity_inputs_sha256", "soul_sha256", "authority", "canonical_authority_granted", }, "image identity binding fields are not exact", ) for field in ("bundle_sha256", "manifest_sha256", "identity_inputs_sha256", "soul_sha256"): _require(_is_sha256(identity.get(field)), f"image identity {field} is invalid") _require( (identity.get("authority"), identity.get("canonical_authority_granted")) in { ("synthetic_noncanonical_fixture", False), ("canonical_database_same_transaction_export", True), }, "image identity authority is invalid", ) bindings = value.get("source_bindings") _require( isinstance(bindings, dict) and set(bindings) == {"Dockerfile", "entrypoint.py", "package_contract.py"} and all(_is_sha256(item) for item in bindings.values()), "image packaging source bindings are invalid", ) _require( isinstance(value.get("claim_ceiling"), str) and bool(value["claim_ceiling"]), "image claim ceiling is missing" ) stable = {key: item for key, item in value.items() if key != "input_sha256"} _require(value.get("input_sha256") == canonical_sha256(stable), "image input self-hash drifted") _assert_no_forbidden_descriptor_markers(value) def build_image_input( *, artifact: Path, receipt: Path, identity: Path, ca: Path, source_revision: str, dockerfile: Path, entrypoint: Path, package_contract: Path, ) -> dict[str, Any]: _require(sha256_file(ca) == CA_SHA256, "Cloud SQL CA does not match the reviewed certificate") runtime = validate_artifact(artifact, receipt, source_revision) runtime.update( { "python_version": PYTHON_VERSION, "uv_version": UV_VERSION, "postgres_version": POSTGRES_VERSION, "cloudsql_ca_sha256": CA_SHA256, } ) identity_binding = validate_identity_bundle(identity) stable = { "schema": IMAGE_INPUT_SCHEMA, "target": _target(), "runtime": {key: runtime[key] for key in runtime if key != "claim_ceiling"}, "identity": identity_binding, "base_images": _base_images(), "source_bindings": { "Dockerfile": sha256_file(dockerfile), "entrypoint.py": sha256_file(entrypoint), "package_contract.py": sha256_file(package_contract), }, "claim_ceiling": runtime["claim_ceiling"], } value = {**stable, "input_sha256": canonical_sha256(stable)} validate_image_input(value) return value def prepare_context( *, repo_root: Path, artifact: Path, receipt: Path, identity: Path, output: Path, source_revision: str, ) -> dict[str, Any]: _require(not output.exists() and not output.is_symlink(), "output context already exists") dockerfile = repo_root / "docker" / "leoclean-nosend" / "Dockerfile" entrypoint = repo_root / "docker" / "leoclean-nosend" / "entrypoint.py" package_contract = repo_root / "ops" / "gcp_leoclean_nosend_package.py" ca = repo_root / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem" for path, label in ( (dockerfile, "Dockerfile"), (entrypoint, "entrypoint"), (package_contract, "package contract"), (ca, "Cloud SQL CA"), ): _require(path.is_file() and not path.is_symlink(), f"{label} source is missing or unsafe") image_input = build_image_input( artifact=artifact, receipt=receipt, identity=identity, ca=ca, source_revision=source_revision, dockerfile=dockerfile, entrypoint=entrypoint, package_contract=package_contract, ) output.mkdir(parents=True, mode=0o700) try: shutil.copytree(artifact, output / "artifact", symlinks=False) shutil.copytree(identity, output / "identity", symlinks=False) for name in IDENTITY_FILES: (output / "identity" / name).chmod(0o600) shutil.copyfile(receipt, output / "artifact-receipt.json") shutil.copyfile(ca, output / "cloudsql-server-ca.pem") shutil.copyfile(dockerfile, output / "Dockerfile") shutil.copyfile(entrypoint, output / "entrypoint.py") shutil.copyfile(package_contract, output / "package_contract.py") for name in ( "artifact-receipt.json", "cloudsql-server-ca.pem", "Dockerfile", "entrypoint.py", "package_contract.py", ): (output / name).chmod(0o644) write_json(output / "image-input.json", image_input) write_json( output / "build-args.json", { "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"], "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"], "INPUT_SHA256": image_input["input_sha256"], "TELEO_REVISION": source_revision, }, ) _safe_tree(output, "prepared image context") return image_input except BaseException: shutil.rmtree(output, ignore_errors=True) raise def build_release_descriptor(image_input: dict[str, Any], image_reference: str) -> dict[str, Any]: validate_image_input(image_input) match = IMAGE_REFERENCE.fullmatch(image_reference) _require(match is not None, "release image must be the exact staging repository at a sha256 digest") digest = f"sha256:{match.group(1)}" labels = expected_image_labels(image_input) stable = { "schema": RELEASE_SCHEMA, "target": _target(), "image": { "reference": image_reference, "digest": digest, "platform": PLATFORM, "labels": labels, "input_sha256": image_input["input_sha256"], }, "image_input": image_input, } value = {**stable, "release_sha256": canonical_sha256(stable)} validate_release_descriptor(value) return value def expected_image_labels(image_input: dict[str, Any]) -> dict[str, str]: validate_image_input(image_input) return { "livingip.artifact.sha256": image_input["runtime"]["artifact_sha256"], "livingip.identity.sha256": image_input["identity"]["bundle_sha256"], "livingip.input.sha256": image_input["input_sha256"], "livingip.runtime": "leoclean-nosend", "livingip.target": "gcp-staging", "org.opencontainers.image.revision": image_input["runtime"]["teleo_git_head"], "org.opencontainers.image.source": "https://github.com/living-ip/teleo-infrastructure", } def validate_release_descriptor(value: dict[str, Any]) -> None: _require( set(value) == {"schema", "target", "image", "image_input", "release_sha256"}, "release descriptor fields are not exact", ) _require(value.get("schema") == RELEASE_SCHEMA, "release descriptor schema is invalid") _require(value.get("target") == _target(), "release target is not exact GCP staging") image_input = value.get("image_input") _require(isinstance(image_input, dict), "release image input is missing") validate_image_input(image_input) image = value.get("image") _require( isinstance(image, dict) and set(image) == {"reference", "digest", "platform", "labels", "input_sha256"}, "release image fields are not exact", ) reference = image.get("reference") match = IMAGE_REFERENCE.fullmatch(reference) if isinstance(reference, str) else None _require(match is not None, "release image is not digest-pinned in the staging repository") _require(image.get("digest") == f"sha256:{match.group(1)}", "release image digest differs from its reference") _require(image.get("platform") == PLATFORM, "release platform is not exact") _require(image.get("input_sha256") == image_input["input_sha256"], "release image input binding drifted") _require(image.get("labels") == expected_image_labels(image_input), "release image labels are not exact") stable = {key: item for key, item in value.items() if key != "release_sha256"} _require(value.get("release_sha256") == canonical_sha256(stable), "release descriptor self-hash drifted") _assert_no_forbidden_descriptor_markers(value) def render_systemd_unit(release: dict[str, Any]) -> str: validate_release_descriptor(release) image = release["image"]["reference"] release_sha = release["release_sha256"] profile_tmpfs = f"{PROFILE_ROOT}:rw,noexec,nosuid,nodev,size=64m,uid=0,gid=0,mode=0700" temp_tmpfs = f"/tmp:rw,noexec,nosuid,nodev,size=64m,uid={RUNTIME_UID},gid={RUNTIME_GID},mode=0700" postgres_data_tmpfs = "/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000" exec_start = " ".join( [ "/usr/bin/docker run", "--rm", f"--name={CONTAINER_NAME}", "--pull=never", "--read-only", "--cap-drop=ALL", "--cap-add=CHOWN", "--cap-add=SETGID", "--cap-add=SETUID", "--cap-add=SETPCAP", "--security-opt=no-new-privileges:true", "--pids-limit=512", "--network=bridge", "--stop-timeout=30", f"--tmpfs={profile_tmpfs}", f"--tmpfs={temp_tmpfs}", f"--tmpfs={postgres_data_tmpfs}", f"--label=livingip.release.sha256={release_sha}", image, "service", ] ) unit = f"""[Unit] Description=LivingIP leoclean-gcp-nosend staging Documentation=https://github.com/living-ip/teleo-infrastructure After=docker.service network-online.target Requires=docker.service Wants=network-online.target [Service] Type=simple ExecStart={exec_start} ExecStop=/usr/bin/docker stop --time=30 {CONTAINER_NAME} ExecStopPost=-/usr/bin/docker rm --force {CONTAINER_NAME} Restart=on-failure RestartSec=5s TimeoutStartSec=120s TimeoutStopSec=45s UMask=0077 NoNewPrivileges=yes CapabilityBoundingSet= AmbientCapabilities= PrivateDevices=yes PrivateTmp=yes ProtectSystem=strict ProtectHome=yes ProtectKernelTunables=yes ProtectKernelModules=yes ProtectKernelLogs=yes ProtectControlGroups=yes LockPersonality=yes RestrictSUIDSGID=yes RestrictRealtime=yes RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 SystemCallArchitectures=native [Install] WantedBy=multi-user.target """ _require("--publish" not in unit and " -p " not in unit, "service unit publishes a port") _require("docker.sock" not in unit, "service unit mounts the Docker socket") _require( "EnvironmentFile" not in unit and "LoadCredential" not in unit, "service unit adds an unreviewed credential source", ) _require("telegram" not in unit.casefold(), "service unit contains a messaging marker") return unit def _git_head(repo_root: Path) -> str: try: return subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() except (OSError, subprocess.CalledProcessError) as exc: raise PackageError("cannot determine source revision") from exc def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) prepare = subparsers.add_parser("prepare") prepare.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) prepare.add_argument("--artifact", type=Path, required=True) prepare.add_argument("--receipt", type=Path, required=True) prepare.add_argument("--identity", type=Path, required=True) prepare.add_argument("--output", type=Path, required=True) finalize = subparsers.add_parser("finalize") finalize.add_argument("--image-input", type=Path, required=True) finalize.add_argument("--image-reference", required=True) finalize.add_argument("--output-release", type=Path, required=True) finalize.add_argument("--output-unit", type=Path, required=True) validate = subparsers.add_parser("validate") validate.add_argument("--release", type=Path, required=True) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: try: args = _parse_args(argv) if args.command == "prepare": repo_root = args.repo_root.resolve() result = prepare_context( repo_root=repo_root, artifact=args.artifact.resolve(), receipt=args.receipt.resolve(), identity=args.identity.resolve(), output=args.output.absolute(), source_revision=_git_head(repo_root), ) elif args.command == "finalize": image_input = load_json(args.image_input, "image input") result = build_release_descriptor(image_input, args.image_reference) for path in (args.output_release, args.output_unit): _require(not path.exists() and not path.is_symlink(), "finalize output already exists") path.parent.mkdir(parents=True, exist_ok=True) write_json(args.output_release, result) args.output_unit.write_text(render_systemd_unit(result), encoding="utf-8") args.output_unit.chmod(0o644) else: result = load_json(args.release, "release descriptor") validate_release_descriptor(result) print(json.dumps({"status": "pass", "schema": result["schema"]}, sort_keys=True)) return 0 except PackageError as exc: print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True), file=sys.stderr) return 65 if __name__ == "__main__": raise SystemExit(main())