#!/usr/bin/env python3 """Read-only control-plane and host preflight for leoclean GCP staging.""" from __future__ import annotations import argparse import ast import hashlib import ipaddress import json import os import re import shlex import stat import subprocess import sys from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path from typing import Any if __package__: from ops import gcp_leoclean_nosend_package as package else: import gcp_leoclean_nosend_package as package SCHEMA = "livingip.leocleanNoSendGcpPreflight.v1" SOURCE_ROOT = Path(__file__).resolve().parents[1] RUNTIME_CLOUDSQL_SOURCE = SOURCE_ROOT / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" CRITICAL_TRACKED_PATHS = ( "ops/check_gcp_leoclean_nosend_preflight.py", "ops/gcp_leoclean_nosend_package.py", "hermes-agent/leoclean-bin/cloudsql_memory_tool.py", ) REGION = package.ZONE.rsplit("-", 1)[0] MACHINE_TYPE = "e2-standard-4" MINIMUM_BOOT_DISK_GB = 80 MINIMUM_DOCKER_FREE_BYTES = 10 * 1024**3 MINIMUM_DOCKER_MAJOR = 24 SSH_KEY_LIFETIME = "5m" NETWORK = "teleo-staging-net" SUBNET = "teleo-staging-europe-west6" SUBNET_CIDR = ipaddress.ip_network("10.60.0.0/20") CLOUDSQL_INSTANCE = "teleo-pgvector-standby" ARTIFACT_REPOSITORY = "teleo" Runner = Callable[[list[str]], subprocess.CompletedProcess[str]] HEX_40 = re.compile(r"^[0-9a-f]{40}$") VERSION = re.compile(r"^(\d+)\.(\d+)(?:\.(\d+))?(?:[-+].*)?$") SAFE_VALUE = re.compile(r"^[A-Za-z0-9@._/+:-]{1,512}$") class PreflightError(RuntimeError): """A prerequisite is missing or the live readback is unavailable.""" def __init__(self, category: str, message: str) -> None: super().__init__(message) self.category = category def _default_runner(argv: list[str]) -> subprocess.CompletedProcess[str]: environment = os.environ.copy() environment["CLOUDSDK_PYTHON"] = sys.executable return subprocess.run(argv, text=True, capture_output=True, check=False, timeout=60, env=environment) def _assert_read_only_command(argv: list[str]) -> None: allowed_prefixes = ( ("gcloud", "projects", "describe"), ("gcloud", "compute", "instances", "describe"), ("gcloud", "compute", "disks", "describe"), ("gcloud", "compute", "networks", "subnets", "describe"), ("gcloud", "sql", "instances", "describe"), ("gcloud", "artifacts", "repositories", "describe"), ("gcloud", "compute", "ssh"), ) if not any(tuple(argv[: len(prefix)]) == prefix for prefix in allowed_prefixes): raise PreflightError("unsafe_command", "preflight attempted an unreviewed command") if argv[:3] == ["gcloud", "compute", "ssh"]: remote = [item.removeprefix("--command=") for item in argv if item.startswith("--command=")] if len(remote) != 1: raise PreflightError("unsafe_command", "preflight SSH command is not exact") private_ip = re.search(r"/dev/tcp/([0-9.]+)/5432", remote[0]) if private_ip is None or argv != _ssh_argv(private_ip.group(1)): raise PreflightError("unsafe_command", "preflight SSH command is not exact") lowered = " ".join(argv).casefold() for marker in ( " add-iam-policy-binding", " remove-iam-policy-binding", " secrets versions access", " compute scp", " docker run", " docker pull", " docker push", " docker rm", " systemctl start", " systemctl stop", " systemctl restart", " systemctl enable", " systemctl disable", " systemctl daemon-reload", ): if marker in lowered: raise PreflightError("unsafe_command", "preflight command contains a mutation marker") def _classify_command_failure(completed: subprocess.CompletedProcess[str]) -> PreflightError: combined = f"{completed.stdout}\n{completed.stderr}".casefold() if "reauthentication failed" in combined or "gcloud auth login" in combined: return PreflightError("authentication_required", "Google Cloud authentication requires interactive refresh") if "permission_denied" in combined or "does not have permission" in combined or "forbidden" in combined: return PreflightError("permission_denied", "the active Google Cloud principal lacks read-only preflight access") if "connection timed out" in combined or "failed to connect" in combined: return PreflightError("connection_unavailable", "the bounded IAP/SSH readback was unavailable") return PreflightError("command_failed", f"a read-only preflight command exited {completed.returncode}") def _invoke(argv: list[str], runner: Runner) -> str: _assert_read_only_command(argv) try: completed = runner(argv) except (OSError, subprocess.SubprocessError) as exc: raise PreflightError("command_unavailable", f"cannot run read-only preflight: {type(exc).__name__}") from exc if completed.returncode != 0: raise _classify_command_failure(completed) return completed.stdout def _json(argv: list[str], runner: Runner) -> dict[str, Any]: try: value = json.loads(_invoke(argv, runner)) except json.JSONDecodeError as exc: raise PreflightError("invalid_readback", "a control-plane readback was not valid JSON") from exc if not isinstance(value, dict): raise PreflightError("invalid_readback", "a control-plane readback was not a JSON object") return value def _require(condition: bool, message: str) -> None: if not condition: raise PreflightError("prerequisite_mismatch", message) def _suffix(value: object) -> str: return str(value or "").rstrip("/").rsplit("/", 1)[-1] def _runtime_string_constant(name: str) -> str: try: tree = ast.parse(RUNTIME_CLOUDSQL_SOURCE.read_text(encoding="utf-8")) except (OSError, SyntaxError) as exc: raise PreflightError("invalid_runtime_contract", "cannot read the tracked runtime database contract") from exc for node in tree.body: if not isinstance(node, ast.Assign) or len(node.targets) != 1: continue target = node.targets[0] if ( isinstance(target, ast.Name) and target.id == name and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str) ): return node.value.value raise PreflightError("invalid_runtime_contract", f"runtime database contract lacks {name}") def _project_readback(runner: Runner) -> dict[str, Any]: value = _json( [ "gcloud", "projects", "describe", package.PROJECT, "--format=json(projectId,lifecycleState)", ], runner, ) _require(value == {"lifecycleState": "ACTIVE", "projectId": package.PROJECT}, "GCP project identity is not exact") return {"project_id": package.PROJECT, "lifecycle_state": "ACTIVE"} def _instance_readback(runner: Runner) -> dict[str, Any]: value = _json( [ "gcloud", "compute", "instances", "describe", package.INSTANCE, "--project", package.PROJECT, "--zone", package.ZONE, "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts,disks,networkInterfaces,tags.items)", ], runner, ) accounts = value.get("serviceAccounts") emails = [item.get("email") for item in accounts] if isinstance(accounts, list) else [] _require(value.get("name") == package.INSTANCE, "staging instance name drifted") _require(value.get("status") == "RUNNING", "staging instance is not running") _require(_suffix(value.get("zone")) == package.ZONE, "staging instance zone drifted") _require(_suffix(value.get("machineType")) == MACHINE_TYPE, "staging instance machine type drifted") _require(emails == [package.SERVICE_ACCOUNT], "staging instance service account drifted") disks = value.get("disks") _require(isinstance(disks, list) and len(disks) == 1, "staging instance disk attachment drifted") _require(disks[0].get("boot") is True, "staging instance boot-disk attachment drifted") _require(_suffix(disks[0].get("source")) == package.INSTANCE, "staging instance boot disk drifted") interfaces = value.get("networkInterfaces") _require(isinstance(interfaces, list) and len(interfaces) == 1, "staging instance network interface drifted") interface = interfaces[0] _require(_suffix(interface.get("network")) == NETWORK, "staging instance network drifted") _require(_suffix(interface.get("subnetwork")) == SUBNET, "staging instance subnet drifted") _require(not interface.get("accessConfigs"), "staging instance has a public network address") _require(not interface.get("ipv6AccessConfigs"), "staging instance has public IPv6 access configuration") _require(not interface.get("externalIpv6"), "staging instance has a public IPv6 address") _require(interface.get("stackType") == "IPV4_ONLY", "staging instance IP stack drifted") try: network_ip = ipaddress.ip_address(str(interface.get("networkIP") or "")) except ValueError as exc: raise PreflightError("invalid_readback", "staging instance private address is invalid") from exc _require(network_ip in SUBNET_CIDR, "staging instance private address is outside the reviewed subnet") tags = value.get("tags") tag_items = tags.get("items") if isinstance(tags, dict) else None _require(tag_items == ["teleo-staging-ssh"], "staging instance network tag drifted") return { "name": package.INSTANCE, "status": "RUNNING", "zone": package.ZONE, "machine_type": MACHINE_TYPE, "cpu_platform": str(value.get("cpuPlatform") or "unknown"), "service_account": package.SERVICE_ACCOUNT, } def _disk_readback(runner: Runner) -> dict[str, Any]: value = _json( [ "gcloud", "compute", "disks", "describe", package.INSTANCE, "--project", package.PROJECT, "--zone", package.ZONE, "--format=json(name,status,sizeGb,zone,architecture,users)", ], runner, ) try: size_gb = int(value.get("sizeGb")) except (TypeError, ValueError) as exc: raise PreflightError("invalid_readback", "boot disk size was not an integer") from exc _require(value.get("name") == package.INSTANCE, "staging boot disk name drifted") _require(value.get("status") == "READY", "staging boot disk is not ready") _require(_suffix(value.get("zone")) == package.ZONE, "staging boot disk zone drifted") _require(size_gb >= MINIMUM_BOOT_DISK_GB, "staging boot disk is below the reviewed minimum") _require(value.get("architecture") == "X86_64", "staging boot disk architecture drifted") users = value.get("users") _require( isinstance(users, list) and len(users) == 1 and _suffix(users[0]) == package.INSTANCE, "staging boot disk attachment drifted", ) return { "name": package.INSTANCE, "status": "READY", "size_gb": size_gb, "minimum_size_gb": MINIMUM_BOOT_DISK_GB, "architecture": str(value.get("architecture") or "unspecified"), } def _subnet_readback(runner: Runner) -> dict[str, Any]: value = _json( [ "gcloud", "compute", "networks", "subnets", "describe", SUBNET, "--project", package.PROJECT, "--region", REGION, "--format=json(name,region,network,ipCidrRange,privateIpGoogleAccess)", ], runner, ) _require(value.get("name") == SUBNET, "staging subnet name drifted") _require(_suffix(value.get("region")) == REGION, "staging subnet region drifted") _require(_suffix(value.get("network")) == NETWORK, "staging subnet network drifted") try: subnet_cidr = ipaddress.ip_network(str(value.get("ipCidrRange") or ""), strict=True) except ValueError as exc: raise PreflightError("invalid_readback", "staging subnet CIDR is invalid") from exc _require(subnet_cidr == SUBNET_CIDR, "staging subnet CIDR drifted") _require(value.get("privateIpGoogleAccess") is True, "staging subnet lacks Private Google Access") return { "name": SUBNET, "region": REGION, "network": NETWORK, "cidr": str(SUBNET_CIDR), "private_google_access": True, } def _artifact_readback(runner: Runner) -> dict[str, Any]: value = _json( [ "gcloud", "artifacts", "repositories", "describe", ARTIFACT_REPOSITORY, "--project", package.PROJECT, "--location", REGION, "--format=json(name,format,mode,dockerConfig.immutableTags)", ], runner, ) _require(_suffix(value.get("name")) == ARTIFACT_REPOSITORY, "Artifact Registry repository drifted") _require(value.get("format") == "DOCKER", "Artifact Registry repository is not Docker") _require(value.get("mode") == "STANDARD_REPOSITORY", "Artifact Registry repository mode drifted") docker_config = value.get("dockerConfig") _require(isinstance(docker_config, dict), "Artifact Registry Docker policy is missing") _require(docker_config.get("immutableTags") is True, "Artifact Registry tags are not immutable") return {"name": ARTIFACT_REPOSITORY, "region": REGION, "format": "DOCKER", "immutable_tags": True} def _cloudsql_readback(runner: Runner) -> tuple[dict[str, Any], str]: value = _json( [ "gcloud", "sql", "instances", "describe", CLOUDSQL_INSTANCE, "--project", package.PROJECT, "--format=json(name,state,region,databaseVersion,ipAddresses,settings.ipConfiguration)", ], runner, ) _require(value.get("name") == CLOUDSQL_INSTANCE, "Cloud SQL instance name drifted") _require(value.get("state") == "RUNNABLE", "Cloud SQL instance is not runnable") _require(value.get("region") == REGION, "Cloud SQL region drifted") _require(value.get("databaseVersion") == "POSTGRES_16", "Cloud SQL is not exactly PostgreSQL 16") settings = value.get("settings") ip_config = settings.get("ipConfiguration") if isinstance(settings, dict) else None _require(isinstance(ip_config, dict), "Cloud SQL IP configuration is missing") _require(ip_config.get("ipv4Enabled") is False, "Cloud SQL public IPv4 is enabled") _require(_suffix(ip_config.get("privateNetwork")) == NETWORK, "Cloud SQL private network drifted") _require(ip_config.get("sslMode") == "ENCRYPTED_ONLY", "Cloud SQL TLS mode is incompatible with the runtime") addresses = value.get("ipAddresses") _require(isinstance(addresses, list) and len(addresses) == 1, "Cloud SQL address set is not private-only") _require(addresses[0].get("type") == "PRIVATE", "Cloud SQL exposes a non-private address") private_ips = [addresses[0].get("ipAddress")] _require(isinstance(private_ips[0], str), "Cloud SQL private address is not exact") try: private_ip = str(ipaddress.ip_address(private_ips[0])) except ValueError as exc: raise PreflightError("invalid_readback", "Cloud SQL private address is invalid") from exc _require(ipaddress.ip_address(private_ip).is_private, "Cloud SQL address is not private") _require( private_ip == _runtime_string_constant("RUNTIME_CLOUDSQL_HOST"), "Cloud SQL private address drifted from the runtime contract", ) return ( { "name": CLOUDSQL_INSTANCE, "state": "RUNNABLE", "region": REGION, "database_version": str(value.get("databaseVersion")), "public_ipv4_enabled": False, "private_network": NETWORK, "ssl_mode": ip_config.get("sslMode"), "private_ip_sha256": hashlib.sha256(private_ip.encode("ascii")).hexdigest(), }, private_ip, ) def _host_script(private_ip: str) -> str: _require(str(ipaddress.ip_address(private_ip)) == private_ip, "Cloud SQL private address is invalid") metadata = "http://metadata.google.internal/computeMetadata/v1" return f"""set -euo pipefail export LC_ALL=C export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin metadata() {{ curl -q --noproxy '*' --proto '=http' --max-time 5 --fail --silent --show-error --header 'Metadata-Flavor: Google' \"{metadata}/$1\"; }} architecture=\"$(uname -m)\" project=\"$(metadata project/project-id)\" instance=\"$(metadata instance/name)\" service_account=\"$(metadata instance/service-accounts/default/email)\" test -x /usr/bin/docker docker_server_version=\"$(sudo -n /usr/bin/docker version --format '{{{{.Server.Version}}}}')\" docker_server_os=\"$(sudo -n /usr/bin/docker version --format '{{{{.Server.Os}}}}')\" docker_server_arch=\"$(sudo -n /usr/bin/docker version --format '{{{{.Server.Arch}}}}')\" docker_root=\"$(sudo -n /usr/bin/docker info --format '{{{{.DockerRootDir}}}}')\" docker_root_free_bytes=\"$(sudo -n /usr/bin/df -B1 --output=avail \"$docker_root\" | tail -n 1 | tr -d ' ')\" unit_load_state=\"$(systemctl show {shlex.quote(package.SERVICE)} --property=LoadState --value)\" container_collision_count=\"$(sudo -n /usr/bin/docker ps -aq --filter 'name=^/{package.CONTAINER_NAME}$' | wc -l | tr -d ' ')\" if timeout 5 bash -c 'exec 3<>/dev/tcp/{private_ip}/5432; exec 3<&-; exec 3>&-'; then cloudsql_tcp=reachable; else cloudsql_tcp=unreachable; fi printf 'architecture=%s\n' \"$architecture\" printf 'project=%s\n' \"$project\" printf 'instance=%s\n' \"$instance\" printf 'service_account=%s\n' \"$service_account\" printf 'docker_server_version=%s\n' \"$docker_server_version\" printf 'docker_server_os=%s\n' \"$docker_server_os\" printf 'docker_server_arch=%s\n' \"$docker_server_arch\" printf 'docker_root_free_bytes=%s\n' \"$docker_root_free_bytes\" printf 'unit_load_state=%s\n' \"$unit_load_state\" printf 'container_collision_count=%s\n' \"$container_collision_count\" printf 'cloudsql_tcp=%s\n' \"$cloudsql_tcp\" """ def _ssh_argv(private_ip: str) -> list[str]: return [ "gcloud", "compute", "ssh", package.INSTANCE, "--project", package.PROJECT, "--zone", package.ZONE, "--tunnel-through-iap", "--quiet", f"--ssh-key-expire-after={SSH_KEY_LIFETIME}", f"--command={_host_script(private_ip)}", ] def _parse_host_output(output: str) -> dict[str, str]: values: dict[str, str] = {} for line in output.splitlines(): key, separator, value = line.partition("=") if not separator or key in values or not re.fullmatch(r"[a-z_]+", key) or not SAFE_VALUE.fullmatch(value): raise PreflightError("invalid_readback", "host preflight output was not bounded key/value data") values[key] = value expected = { "architecture", "project", "instance", "service_account", "docker_server_version", "docker_server_os", "docker_server_arch", "docker_root_free_bytes", "unit_load_state", "container_collision_count", "cloudsql_tcp", } _require(set(values) == expected, "host preflight fields are not exact") return values def _host_readback(runner: Runner, private_ip: str) -> dict[str, Any]: output = _invoke(_ssh_argv(private_ip), runner) value = _parse_host_output(output) version = VERSION.fullmatch(value["docker_server_version"]) _require(value["architecture"] == "x86_64", "staging host architecture is not x86_64") _require(value["project"] == package.PROJECT, "host metadata project drifted") _require(value["instance"] == package.INSTANCE, "host metadata instance drifted") _require(value["service_account"] == package.SERVICE_ACCOUNT, "host metadata service account drifted") _require( version is not None and int(version.group(1)) >= MINIMUM_DOCKER_MAJOR, "Docker server is below the reviewed minimum", ) _require(value["docker_server_os"] == "linux", "Docker server OS is not Linux") _require(value["docker_server_arch"] == "amd64", "Docker server architecture is not amd64") try: free_bytes = int(value["docker_root_free_bytes"]) collision_count = int(value["container_collision_count"]) except ValueError as exc: raise PreflightError("invalid_readback", "host numeric field was invalid") from exc _require(free_bytes >= MINIMUM_DOCKER_FREE_BYTES, "Docker storage has insufficient free space") _require(value["unit_load_state"] == "not-found", "staging service unit name already exists") _require(collision_count == 0, "staging container name already exists") _require(value["cloudsql_tcp"] == "reachable", "Cloud SQL private TCP/5432 is not reachable from staging") return { "architecture": "x86_64", "metadata_identity": { "project": package.PROJECT, "instance": package.INSTANCE, "service_account": package.SERVICE_ACCOUNT, }, "docker": { "server_version": value["docker_server_version"], "server_os": "linux", "server_architecture": "amd64", "minimum_major": MINIMUM_DOCKER_MAJOR, "root_free_bytes": free_bytes, "minimum_free_bytes": MINIMUM_DOCKER_FREE_BYTES, }, "service_unit_load_state": "not-found", "container_collision_count": 0, "cloudsql_private_tcp_5432": "reachable", } def build_preflight(*, runner: Runner = _default_runner, source_revision: str) -> dict[str, Any]: _require(HEX_40.fullmatch(source_revision) is not None, "source revision must be an exact commit") cloudsql, private_ip = _cloudsql_readback(runner) payload = { "schema": SCHEMA, "status": "pass", "observed_at_utc": datetime.now(UTC).isoformat(), "source_revision": source_revision, "target": { "project": package.PROJECT, "zone": package.ZONE, "instance": package.INSTANCE, "service": package.SERVICE, "container": package.CONTAINER_NAME, "platform": package.PLATFORM, }, "control_plane": { "project": _project_readback(runner), "instance": _instance_readback(runner), "boot_disk": _disk_readback(runner), "subnet": _subnet_readback(runner), "artifact_repository": _artifact_readback(runner), "cloudsql": cloudsql, }, "host": _host_readback(runner, private_ip), "authority": { "mode": "bounded_observation_preflight", "secret_values_read": False, "database_credentials_used": False, "database_queries_run": False, "control_plane_observations_before_ssh_are_describe_only": True, "remote_host_runtime_mutations": False, "ssh_access_may_register_time_bounded_key": True, "ssh_key_maximum_lifetime": SSH_KEY_LIFETIME, }, "claim_ceiling": ( "This receipt proves only the listed control-plane observations and host-network private TCP " "reachability. It does not prove Docker-bridge reachability, Artifact Registry pull authorization, " "secret access, database authentication, scoped PostgreSQL permissions, image publication, deployment, " "restart, rollback, model behavior, parity, traffic, or production readiness. The IAP/SSH access path " "may register a time-bounded SSH key; the remote command does not change host runtime state." ), } return payload def _validated_repo_root(repo_root: Path) -> Path: resolved = repo_root.expanduser().resolve() if resolved != SOURCE_ROOT: raise PreflightError("invalid_source_root", "preflight must run against the repository containing this script") return resolved def _normalized_output_path(path: Path) -> Path: expanded = path.expanduser() if not expanded.is_absolute(): expanded = Path.cwd() / expanded if expanded.name in {"", ".", ".."}: raise PreflightError("invalid_output_path", "preflight receipt output path is invalid") try: parent = expanded.parent.resolve(strict=True) except OSError as exc: raise PreflightError( "invalid_output_path", "preflight receipt parent must already exist", ) from exc return parent / expanded.name def _assert_private_output_parent(parent_descriptor: int, parent: Path) -> os.stat_result: try: observed = os.fstat(parent_descriptor) named = os.stat(parent, follow_symlinks=False) except OSError as exc: raise PreflightError( "invalid_output_path", "preflight receipt parent cannot be identified safely", ) from exc if not ( stat.S_ISDIR(observed.st_mode) and (observed.st_dev, observed.st_ino) == (named.st_dev, named.st_ino) and observed.st_uid == os.geteuid() and observed.st_mode & 0o022 == 0 ): raise PreflightError( "invalid_output_path", "preflight receipt parent must be operator-owned and not group/world writable", ) return observed def _output_entry_stat(parent_descriptor: int, name: str) -> os.stat_result | None: try: return os.stat(name, dir_fd=parent_descriptor, follow_symlinks=False) except FileNotFoundError: return None except OSError as exc: raise PreflightError( "invalid_output_path", "preflight receipt output cannot be identified safely", ) from exc def _open_private_output_parent(output_path: Path) -> int: flags = os.O_RDONLY if hasattr(os, "O_DIRECTORY"): flags |= os.O_DIRECTORY if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW try: descriptor = os.open(output_path.parent, flags) except OSError as exc: raise PreflightError( "invalid_output_path", "preflight receipt parent is missing or unsafe", ) from exc try: _assert_private_output_parent(descriptor, output_path.parent) if _output_entry_stat(descriptor, output_path.name) is not None: raise PreflightError( "invalid_output_path", "preflight receipt output already exists", ) except (OSError, PreflightError): os.close(descriptor) raise return descriptor def _write_all(descriptor: int, payload: bytes) -> None: offset = 0 while offset < len(payload): written = os.write(descriptor, payload[offset:]) if written <= 0: raise OSError("preflight receipt write made no progress") offset += written def _publish_private_receipt( output_path: Path, parent_descriptor: int, payload: dict[str, Any], ) -> None: try: encoded = (json.dumps(payload, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8") except (TypeError, ValueError) as exc: raise PreflightError( "receipt_publication_failed", "preflight receipt could not be encoded", ) from exc descriptor: int | None = None try: _assert_private_output_parent(parent_descriptor, output_path.parent) if _output_entry_stat(parent_descriptor, output_path.name) is not None: raise PreflightError( "receipt_publication_failed", "preflight receipt output was created before publication", ) flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW descriptor = os.open(output_path.name, flags, 0o600, dir_fd=parent_descriptor) os.fchmod(descriptor, 0o600) before = os.fstat(descriptor) _write_all(descriptor, encoded) os.fsync(descriptor) after = os.fstat(descriptor) named = os.stat(output_path.name, dir_fd=parent_descriptor, follow_symlinks=False) through_path = os.stat(output_path, follow_symlinks=False) if not ( stat.S_ISREG(after.st_mode) and (before.st_dev, before.st_ino) == (after.st_dev, after.st_ino) and (after.st_dev, after.st_ino) == (named.st_dev, named.st_ino) and (after.st_dev, after.st_ino) == (through_path.st_dev, through_path.st_ino) and after.st_uid == os.geteuid() and stat.S_IMODE(after.st_mode) == 0o600 and after.st_nlink == 1 and after.st_size == len(encoded) ): raise PreflightError( "receipt_publication_failed", "published preflight receipt identity or posture drifted", ) _assert_private_output_parent(parent_descriptor, output_path.parent) os.fsync(parent_descriptor) except PreflightError: raise except OSError as exc: raise PreflightError( "receipt_publication_failed", "preflight receipt publication failed", ) from exc finally: if descriptor is not None: os.close(descriptor) def _source_revision( repo_root: Path, *, critical_paths: tuple[str, ...] = CRITICAL_TRACKED_PATHS, ) -> str: try: status = subprocess.run( ["git", "-C", str(repo_root), "status", "--porcelain=v1", "--untracked-files=normal"], text=True, capture_output=True, check=False, timeout=10, ) revision = subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "HEAD"], text=True, capture_output=True, check=False, timeout=10, ) tracked = subprocess.run( ["git", "-C", str(repo_root), "ls-files", "--error-unmatch", "--", *critical_paths], text=True, capture_output=True, check=False, timeout=10, ) except (OSError, subprocess.SubprocessError) as exc: raise PreflightError("local_preflight_unavailable", "cannot inspect the source revision") from exc if status.returncode != 0 or revision.returncode != 0 or tracked.returncode != 0: raise PreflightError("local_preflight_unavailable", "cannot inspect the source revision") if status.stdout.strip(): raise PreflightError("dirty_source_tree", "preflight requires a clean tracked source tree") source_revision = revision.stdout.strip() _require(HEX_40.fullmatch(source_revision) is not None, "source revision must be an exact commit") return source_revision def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--output", type=Path) args = parser.parse_args(argv) output_path: Path | None = None output_parent_descriptor: int | None = None try: repo_root = _validated_repo_root(args.repo_root) if args.output: output_path = _normalized_output_path(args.output) try: output_path.relative_to(repo_root) except ValueError: pass else: raise PreflightError("invalid_output_path", "preflight receipt must be written outside the source tree") output_parent_descriptor = _open_private_output_parent(output_path) source_revision = _source_revision(repo_root) payload = build_preflight(source_revision=source_revision) if _source_revision(repo_root) != source_revision: raise PreflightError("source_revision_changed", "source revision changed during preflight") if output_path is not None and output_parent_descriptor is not None: _publish_private_receipt(output_path, output_parent_descriptor, payload) except (PreflightError, subprocess.SubprocessError, OSError) as exc: category = exc.category if isinstance(exc, PreflightError) else "local_preflight_unavailable" payload = {"schema": SCHEMA, "status": "blocked", "category": category} print(json.dumps(payload, separators=(",", ":"), sort_keys=True)) return 69 if category in {"authentication_required", "permission_denied", "connection_unavailable"} else 1 finally: if output_parent_descriptor is not None: os.close(output_parent_descriptor) print(json.dumps(payload, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())