From 544b2f615db8794f57b22827a6f97e18efa0361c Mon Sep 17 00:00:00 2001 From: fwazb Date: Fri, 17 Jul 2026 17:17:50 -0700 Subject: [PATCH 1/4] Add GCP no-send staging preflight --- .github/workflows/ci.yml | 2 + docs/gcp-leoclean-nosend-preflight.md | 57 ++ docs/gcp-leoclean-nosend-service-package.md | 9 +- docs/workstreams/gcp-leoclean/goals.md | 31 +- ops/check_gcp_leoclean_nosend_preflight.py | 627 ++++++++++++++++++++ tests/test_gcp_leoclean_nosend_preflight.py | 459 ++++++++++++++ 6 files changed, 1169 insertions(+), 16 deletions(-) create mode 100644 docs/gcp-leoclean-nosend-preflight.md create mode 100644 ops/check_gcp_leoclean_nosend_preflight.py create mode 100644 tests/test_gcp_leoclean_nosend_preflight.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1d76579..9f7c53c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -46,6 +46,7 @@ jobs: ops/apply_gcp_iam_split.py \ ops/attest_gcp_reasoning_compute.py \ ops/capture_vps_canonical_postgres_snapshot.py \ + ops/check_gcp_leoclean_nosend_preflight.py \ ops/check_gcp_infra_readiness.py \ ops/run_gcp_infra_execute_canary.py \ ops/apply_gcp_runtime_baseline.py \ @@ -102,6 +103,7 @@ jobs: tests/test_capture_vps_canonical_postgres_snapshot.py \ tests/test_gcp_infra_execute_canary.py \ tests/test_gcp_infra_readiness_checker.py \ + tests/test_gcp_leoclean_nosend_preflight.py \ tests/test_gcp_leoclean_runtime_permissions.py \ tests/test_gcp_leoclean_nosend_package.py \ tests/test_gcp_leoclean_nosend_oci_smoke.py \ diff --git a/docs/gcp-leoclean-nosend-preflight.md b/docs/gcp-leoclean-nosend-preflight.md new file mode 100644 index 0000000..b7066c9 --- /dev/null +++ b/docs/gcp-leoclean-nosend-preflight.md @@ -0,0 +1,57 @@ +# GCP leoclean no-send staging preflight + +## Outcome + +This slice checks whether the existing `teleo-staging-1` target is ready for +the separately reviewed no-send service package. It creates one revision-bound +receipt and does not deploy the service. + +The checker validates the exact project, VM, machine type, attached service +account, exact boot-disk attachment and architecture, private-only VM +VPC/subnet attachment, Private Google Access, immutable Docker Artifact +Registry repository, exact private PostgreSQL 16 Cloud SQL endpoint, host +architecture, Docker server, Docker storage floor, unused service/container +names, and private TCP/5432 reachability from the VM. + +## Authority boundary + +The control-plane commands are `describe` operations. The bounded remote probe +runs through IAP/SSH and reads metadata, Docker, disk, systemd, container-name, +and TCP state. It does not read Secret Manager, use a database credential, run +a database query, install or pull software, create a container, write a unit, +or start/restart a service. + +`gcloud compute ssh` may register or refresh an SSH key. The checker therefore +uses a maximum five-minute key lifetime and records that caveat explicitly; it +does not claim that the access path or cloud credential metadata is +mutation-free. The remote command itself does not change host runtime state. + +## Run + +Run from the exact clean, tracked source revision with an already authenticated Google +Cloud account that has the reviewed IAP/OS Login read path: + +```bash +python3 ops/check_gcp_leoclean_nosend_preflight.py \ + --output /tmp/leoclean-nosend-gcp-preflight.json +``` + +The output file is created with mode `0600`. Authentication, permission, and +connection failures emit only a bounded category; command stderr is not copied +into the receipt. The checker rejects an alternate repository root, an output +path inside the source tree, untracked critical inputs, and source drift during +the live observation. + +## Claim ceiling and rollback + +A passing receipt proves only the listed control-plane observations and +host-network private TCP reachability at one exact, clean Git revision. 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. + +This slice changes no repository-independent service state. Its local rollback +boundary is deletion of the branch/worktree and any generated receipt. Any +time-bounded SSH key created by the access path expires after at most five +minutes. diff --git a/docs/gcp-leoclean-nosend-service-package.md b/docs/gcp-leoclean-nosend-service-package.md index 422a500..bec9237 100644 --- a/docs/gcp-leoclean-nosend-service-package.md +++ b/docs/gcp-leoclean-nosend-service-package.md @@ -275,10 +275,11 @@ Cloud SQL permissions, a real Artifact Registry push, the registry's tag policy, or the live Artifact Registry digest contents, or that `teleo-staging-1` currently has a compatible container engine, enough disk, the expected architecture, the expected service account, -or an unused unit name. Those are the next read-only live preflight. If the -engine is absent, installation is separate infrastructure work; the deploy must -stop instead of falling back to the production sync script or a mutable host -virtual environment. +or an unused unit name. Those are checked by the separate +`ops/check_gcp_leoclean_nosend_preflight.py` observation receipt. If the engine +is absent, installation is separate infrastructure work; the deploy must stop +instead of falling back to the production sync script or a mutable host virtual +environment. Publishing the branch and stacked PR is only a review handoff. After both the digest-inspection package and this receipt producer receive exact-head human diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index 1140584..93ef22a 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -60,13 +60,18 @@ cut over safely and retire the VPS after soak. commit. - Current exact-head validation: 206 focused package/OCI tests and the full repository suite at 2,480 passed with 3 expected skips; Ruff, Python compile, - and diff checks pass. Two independent advisory exact-head reviews approved + and diff checks pass. Three independent advisory exact-head reviews approved the receipt deletion/identity boundary. Tests mock Docker and perform no registry push. -- Current gate: publish and review the final #209-based #210 head, allow PR - #209 to correct PR #206's feature branch, rerun/review corrected #206, and - only then merge/retarget the narrow stack. Registry access remains a later, - explicitly authorized live gate. +- Current gate: human exact-head review remains pending on #206 and #210. PR + #209 must first correct #206's feature branch; corrected #206 must be + re-reviewed and merged before #210 is retargeted to `main` for its own human + review and merge. +- In progress offline: a separate GCP staging preflight rebuilt on the final + #210 head. It checks exact control-plane and host prerequisites without + reading secrets, logging into PostgreSQL, deploying, or restarting. Its + bounded IAP/SSH path uses a fixed remote command and a maximum five-minute + key lifetime, but no live run is currently authorized. - Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud SQL effective identity, restart behavior, or model/database parity. @@ -76,13 +81,13 @@ cut over safely and retire the VPS after soak. human review, and merge the corrected PR #206. 2. Update, review, retarget, and merge draft PR #210's structured build/push receipt producer and receipt-bound release-v3 finalizer. -3. Under explicit live authorization, verify the staging repository/tag policy, - build and push one candidate, publish its receipt, and independently finalize - the exact digest. -4. Add the separate fail-closed installer/restart/verification/rollback slice +3. Review and merge the bounded staging preflight; under separate live + authorization, run its control-plane and IAP/SSH observations and verify the + staging repository/tag policy. +4. Under explicit live authorization, build and push one candidate, publish + its receipt, and independently finalize the exact digest. +5. Add the separate fail-closed installer/restart/verification/rollback slice that consumes the exact inspection-bound release bundle. -5. Run the read-only `teleo-staging-1` - host/engine/service-account preflight. 6. Provision the scoped Cloud SQL role/secret and deploy the separate staging service. 7. Prove effective identity, reads, function-only proposal staging, denied @@ -95,6 +100,8 @@ cut over safely and retire the VPS after soak. ## Current live-action boundary No merge without explicit human approval, Artifact Registry push, -GCP/Cloud SQL/Secret Manager mutation, +GCP control-plane or IAP/SSH access, Cloud SQL/Secret Manager mutation, service deployment/restart, Telegram change, production promotion, canonical proposal apply, or VPS access is part of the current offline corrective slice. +Any later authorization for the bounded preflight must explicitly accept that +the SSH access path may register a key with a maximum five-minute lifetime. diff --git a/ops/check_gcp_leoclean_nosend_preflight.py b/ops/check_gcp_leoclean_nosend_preflight.py new file mode 100644 index 0000000..48db655 --- /dev/null +++ b/ops/check_gcp_leoclean_nosend_preflight.py @@ -0,0 +1,627 @@ +#!/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 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-2" +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-subnet" +SUBNET_CIDR = ipaddress.ip_network("10.70.0.0/24") +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.email,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 _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() -> 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() + output_path: Path | None = None + try: + repo_root = _validated_repo_root(args.repo_root) + if args.output: + output_path = args.output.expanduser().resolve() + 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") + 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") + 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 + if output_path: + output_path.parent.mkdir(parents=True, exist_ok=True) + package.write_json(output_path, payload, mode=0o600) + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gcp_leoclean_nosend_preflight.py b/tests/test_gcp_leoclean_nosend_preflight.py new file mode 100644 index 0000000..515dcff --- /dev/null +++ b/tests/test_gcp_leoclean_nosend_preflight.py @@ -0,0 +1,459 @@ +"""Contracts for the read-only leoclean GCP staging preflight.""" + +from __future__ import annotations + +import copy +import hashlib +import json +import subprocess +import sys +from collections.abc import Callable +from pathlib import Path + +import pytest + +from ops import check_gcp_leoclean_nosend_preflight as preflight + +REVISION = "a" * 40 +PRIVATE_IP = "10.61.0.3" + + +def live_fixtures() -> dict[str, object]: + return { + "project": {"lifecycleState": "ACTIVE", "projectId": "teleo-501523"}, + "instance": { + "name": "teleo-staging-1", + "status": "RUNNING", + "zone": "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a", + "machineType": ( + "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a/" + "machineTypes/e2-standard-2" + ), + "cpuPlatform": "Intel Broadwell", + "serviceAccounts": [ + {"email": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com"} + ], + "disks": [ + { + "boot": True, + "source": ( + "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a/" + "disks/teleo-staging-1" + ), + } + ], + "networkInterfaces": [ + { + "network": ( + "https://www.googleapis.com/compute/v1/projects/teleo-501523/global/networks/teleo-staging-net" + ), + "subnetwork": ( + "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6/" + "subnetworks/teleo-staging-subnet" + ), + "networkIP": "10.70.0.2", + "accessConfigs": [], + "ipv6AccessConfigs": [], + "stackType": "IPV4_ONLY", + } + ], + "tags": {"items": ["teleo-staging-ssh"]}, + }, + "disk": { + "name": "teleo-staging-1", + "status": "READY", + "sizeGb": "80", + "zone": "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a", + "architecture": "X86_64", + "users": [ + "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a/" + "instances/teleo-staging-1" + ], + }, + "subnet": { + "name": "teleo-staging-subnet", + "region": "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6", + "network": "https://www.googleapis.com/compute/v1/projects/teleo-501523/global/networks/teleo-staging-net", + "ipCidrRange": "10.70.0.0/24", + "privateIpGoogleAccess": True, + }, + "artifact": { + "name": "projects/teleo-501523/locations/europe-west6/repositories/teleo", + "format": "DOCKER", + "mode": "STANDARD_REPOSITORY", + "dockerConfig": {"immutableTags": True}, + }, + "cloudsql": { + "name": "teleo-pgvector-standby", + "state": "RUNNABLE", + "region": "europe-west6", + "databaseVersion": "POSTGRES_16", + "ipAddresses": [{"type": "PRIVATE", "ipAddress": PRIVATE_IP}], + "settings": { + "ipConfiguration": { + "ipv4Enabled": False, + "privateNetwork": "projects/teleo-501523/global/networks/teleo-staging-net", + "sslMode": "ENCRYPTED_ONLY", + } + }, + }, + "host": { + "architecture": "x86_64", + "project": "teleo-501523", + "instance": "teleo-staging-1", + "service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com", + "docker_server_version": "29.4.0", + "docker_server_os": "linux", + "docker_server_arch": "amd64", + "docker_root_free_bytes": str(20 * 1024**3), + "unit_load_state": "not-found", + "container_collision_count": "0", + "cloudsql_tcp": "reachable", + }, + } + + +def fixture_runner( + values: dict[str, object], + calls: list[list[str]] | None = None, +) -> Callable[[list[str]], subprocess.CompletedProcess[str]]: + def run(argv: list[str]) -> subprocess.CompletedProcess[str]: + if calls is not None: + calls.append(argv) + if argv[:3] == ["gcloud", "projects", "describe"]: + value: object = values["project"] + elif argv[:4] == ["gcloud", "compute", "instances", "describe"]: + value = values["instance"] + elif argv[:4] == ["gcloud", "compute", "disks", "describe"]: + value = values["disk"] + elif argv[:5] == ["gcloud", "compute", "networks", "subnets", "describe"]: + value = values["subnet"] + elif argv[:4] == ["gcloud", "artifacts", "repositories", "describe"]: + value = values["artifact"] + elif argv[:4] == ["gcloud", "sql", "instances", "describe"]: + value = values["cloudsql"] + elif argv[:3] == ["gcloud", "compute", "ssh"]: + host = values["host"] + assert isinstance(host, dict) + output = "".join(f"{key}={item}\n" for key, item in host.items()) + return subprocess.CompletedProcess(argv, 0, stdout=output, stderr="") + else: # pragma: no cover - the assertion shows unexpected command drift + raise AssertionError(argv) + return subprocess.CompletedProcess(argv, 0, stdout=json.dumps(value), stderr="") + + return run + + +def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None: + calls: list[list[str]] = [] + payload = preflight.build_preflight(runner=fixture_runner(live_fixtures(), calls), source_revision=REVISION) + + assert payload["status"] == "pass" + assert payload["source_revision"] == REVISION + assert payload["target"] == { + "project": "teleo-501523", + "zone": "europe-west6-a", + "instance": "teleo-staging-1", + "service": "leoclean-gcp-nosend.service", + "container": "livingip-leoclean-nosend", + "platform": "linux/amd64", + } + assert payload["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": "5m", + } + assert payload["host"]["cloudsql_private_tcp_5432"] == "reachable" + assert payload["control_plane"]["cloudsql"]["private_ip_sha256"] == hashlib.sha256( + PRIVATE_IP.encode("ascii") + ).hexdigest() + assert PRIVATE_IP not in json.dumps(payload) + assert len(calls) == 7 + assert calls[:-1] == [ + [ + "gcloud", + "sql", + "instances", + "describe", + "teleo-pgvector-standby", + "--project", + "teleo-501523", + "--format=json(name,state,region,databaseVersion,ipAddresses,settings.ipConfiguration)", + ], + [ + "gcloud", + "projects", + "describe", + "teleo-501523", + "--format=json(projectId,lifecycleState)", + ], + [ + "gcloud", + "compute", + "instances", + "describe", + "teleo-staging-1", + "--project", + "teleo-501523", + "--zone", + "europe-west6-a", + "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts.email,disks,networkInterfaces,tags.items)", + ], + [ + "gcloud", + "compute", + "disks", + "describe", + "teleo-staging-1", + "--project", + "teleo-501523", + "--zone", + "europe-west6-a", + "--format=json(name,status,sizeGb,zone,architecture,users)", + ], + [ + "gcloud", + "compute", + "networks", + "subnets", + "describe", + "teleo-staging-subnet", + "--project", + "teleo-501523", + "--region", + "europe-west6", + "--format=json(name,region,network,ipCidrRange,privateIpGoogleAccess)", + ], + [ + "gcloud", + "artifacts", + "repositories", + "describe", + "teleo", + "--project", + "teleo-501523", + "--location", + "europe-west6", + "--format=json(name,format,mode,dockerConfig.immutableTags)", + ], + ] + ssh = calls[-1] + assert ssh == preflight._ssh_argv(PRIVATE_IP) + assert "--tunnel-through-iap" in ssh + assert "--ssh-key-expire-after=5m" in ssh + remote = next(item.removeprefix("--command=") for item in ssh if item.startswith("--command=")) + assert remote.startswith("set -euo pipefail\n") + for forbidden in ( + "docker run", + "docker pull", + "docker rm", + "secretmanager", + "secrets versions access", + "systemctl start", + "systemctl restart", + "systemctl enable", + ): + assert forbidden not in remote.casefold() + + +@pytest.mark.parametrize( + ("path", "field", "unsafe", "message"), + [ + (("instance", "serviceAccounts"), None, [{"email": "postgres-admin@example.com"}], "service account"), + (("disk",), "sizeGb", "40", "boot disk"), + (("disk",), "architecture", "ARM64", "architecture"), + (("disk",), "users", [], "attachment"), + (("instance", "networkInterfaces"), None, [], "network interface"), + (("instance", "networkInterfaces", 0), "accessConfigs", [{"natIP": "203.0.113.10"}], "public"), + (("instance", "networkInterfaces", 0), "ipv6AccessConfigs", [{"externalIpv6": "2001:db8::1"}], "IPv6"), + (("instance", "networkInterfaces", 0), "externalIpv6", "2001:db8::1", "IPv6"), + (("instance", "networkInterfaces", 0), "networkIP", "10.99.0.2", "reviewed subnet"), + (("subnet",), "privateIpGoogleAccess", False, "Private Google Access"), + (("subnet",), "ipCidrRange", "10.99.0.0/24", "CIDR drifted"), + (("artifact", "dockerConfig"), "immutableTags", False, "not immutable"), + (("cloudsql", "settings", "ipConfiguration"), "ipv4Enabled", True, "public IPv4"), + (("cloudsql",), "databaseVersion", "POSTGRES_160", "exactly PostgreSQL 16"), + (("host",), "container_collision_count", "1", "container name"), + (("host",), "unit_load_state", "loaded", "unit name"), + (("host",), "docker_root_free_bytes", "1024", "free space"), + (("host",), "cloudsql_tcp", "unreachable", "TCP/5432"), + ], +) +def test_preflight_fails_closed_on_prerequisite_drift( + path: tuple[str | int, ...], + field: str | None, + unsafe: object, + message: str, +) -> None: + values = copy.deepcopy(live_fixtures()) + target: object = values + for part in path: + if isinstance(part, int): + assert isinstance(target, list) + target = target[part] + else: + assert isinstance(target, dict) + target = target[part] + if field is None: + assert isinstance(target, list) + parent: object = values + for part in path[:-1]: + if isinstance(part, int): + assert isinstance(parent, list) + parent = parent[part] + else: + assert isinstance(parent, dict) + parent = parent[part] + assert isinstance(parent, dict) + last = path[-1] + assert isinstance(last, str) + parent[last] = unsafe + else: + assert isinstance(target, dict) + target[field] = unsafe + + with pytest.raises(preflight.PreflightError, match=message): + preflight.build_preflight(runner=fixture_runner(values), source_revision=REVISION) + + +def test_authentication_failure_is_classified_without_echoing_error() -> None: + sentinel = "sensitive-browser-session-sentinel" + + def fail(argv: list[str]) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + argv, + 1, + stdout="", + stderr=f"Reauthentication failed. {sentinel}. Run gcloud auth login.", + ) + + with pytest.raises(preflight.PreflightError) as caught: + preflight.build_preflight(runner=fail, source_revision=REVISION) + + assert caught.value.category == "authentication_required" + assert sentinel not in str(caught.value) + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + ({"ipAddresses": [{"type": "PRIVATE", "ipAddress": "10.99.0.2"}]}, "address drifted"), + ( + { + "ipAddresses": [ + {"type": "PRIVATE", "ipAddress": PRIVATE_IP}, + {"type": "PRIMARY", "ipAddress": "203.0.113.10"}, + ] + }, + "private-only", + ), + ( + {"settings": {"ipConfiguration": {"sslMode": "TRUSTED_CLIENT_CERTIFICATE_REQUIRED"}}}, + "TLS mode", + ), + ], +) +def test_cloudsql_endpoint_and_tls_are_exact(mutation: dict[str, object], message: str) -> None: + values = copy.deepcopy(live_fixtures()) + cloudsql = values["cloudsql"] + assert isinstance(cloudsql, dict) + if "settings" in mutation: + settings = cloudsql["settings"] + assert isinstance(settings, dict) + ip_config = settings["ipConfiguration"] + assert isinstance(ip_config, dict) + changed = mutation["settings"] + assert isinstance(changed, dict) + changed_ip_config = changed["ipConfiguration"] + assert isinstance(changed_ip_config, dict) + ip_config.update(changed_ip_config) + else: + cloudsql.update(mutation) + with pytest.raises(preflight.PreflightError, match=message): + preflight.build_preflight(runner=fixture_runner(values), source_revision=REVISION) + + +def test_host_output_rejects_extra_duplicate_or_unbounded_fields() -> None: + valid = live_fixtures()["host"] + assert isinstance(valid, dict) + encoded = "".join(f"{key}={item}\n" for key, item in valid.items()) + + for unsafe in ( + encoded + "unexpected=value\n", + encoded + "project=duplicate\n", + encoded.replace("docker_server_os=linux", "docker_server_os=linux value with spaces"), + ): + with pytest.raises(preflight.PreflightError, match="host preflight"): + preflight._parse_host_output(unsafe) + + +def test_command_gate_rejects_mutating_or_unreviewed_gcloud_calls() -> None: + for command in ( + ["gcloud", "compute", "instances", "start", "teleo-staging-1"], + ["gcloud", "secrets", "versions", "access", "latest"], + ["gcloud", "compute", "scp", "file", "teleo-staging-1:/tmp/file"], + ["gcloud", "compute", "ssh", "teleo-staging-1", "--command=sudo systemctl restart x"], + ["gcloud", "compute", "ssh", "teleo-staging-1", "--command=touch /tmp/preflight-was-here"], + ["bash", "-c", "true"], + ): + with pytest.raises(preflight.PreflightError): + preflight._assert_read_only_command(command) + + +def test_source_revision_requires_clean_tracked_commit(tmp_path: Path) -> None: + subprocess.run(["git", "init", "--quiet", str(tmp_path)], check=True) + tracked = tmp_path / "tracked.txt" + tracked.write_text("committed\n", encoding="utf-8") + subprocess.run(["git", "-C", str(tmp_path), "add", "tracked.txt"], check=True) + subprocess.run( + [ + "git", + "-C", + str(tmp_path), + "-c", + "user.name=Preflight Test", + "-c", + "user.email=preflight@example.invalid", + "commit", + "--quiet", + "-m", + "fixture", + ], + check=True, + ) + revision = subprocess.run( + ["git", "-C", str(tmp_path), "rev-parse", "HEAD"], + text=True, + capture_output=True, + check=True, + ).stdout.strip() + assert preflight._source_revision(tmp_path, critical_paths=("tracked.txt",)) == revision + + (tmp_path / "untracked.txt").write_text("drift\n", encoding="utf-8") + with pytest.raises(preflight.PreflightError, match="clean tracked"): + preflight._source_revision(tmp_path, critical_paths=("tracked.txt",)) + + +def test_preflight_rejects_an_unrelated_clean_repository(tmp_path: Path) -> None: + with pytest.raises(preflight.PreflightError, match="containing this script"): + preflight._validated_repo_root(tmp_path) + + +def test_direct_cli_imports_then_rejects_an_unrelated_repository(tmp_path: Path) -> None: + completed = subprocess.run( + [sys.executable, str(preflight.SOURCE_ROOT / "ops" / "check_gcp_leoclean_nosend_preflight.py"), "--repo-root", str(tmp_path)], + text=True, + capture_output=True, + check=False, + ) + assert completed.returncode == 1 + assert json.loads(completed.stdout) == { + "category": "invalid_source_root", + "schema": "livingip.leocleanNoSendGcpPreflight.v1", + "status": "blocked", + } From 7bafcb4314112cc84c6da557c9e1d209af3e779e Mon Sep 17 00:00:00 2001 From: fwazb Date: Fri, 17 Jul 2026 17:48:23 -0700 Subject: [PATCH 2/4] Reconcile staging preflight inventory --- docs/gcp-leoclean-nosend-preflight.md | 5 +++-- docs/workstreams/gcp-leoclean/goals.md | 8 ++++++++ ops/check_gcp_leoclean_nosend_preflight.py | 8 ++++---- tests/test_gcp_leoclean_nosend_preflight.py | 14 +++++++------- 4 files changed, 22 insertions(+), 13 deletions(-) diff --git a/docs/gcp-leoclean-nosend-preflight.md b/docs/gcp-leoclean-nosend-preflight.md index b7066c9..8444616 100644 --- a/docs/gcp-leoclean-nosend-preflight.md +++ b/docs/gcp-leoclean-nosend-preflight.md @@ -7,8 +7,9 @@ the separately reviewed no-send service package. It creates one revision-bound receipt and does not deploy the service. The checker validates the exact project, VM, machine type, attached service -account, exact boot-disk attachment and architecture, private-only VM -VPC/subnet attachment, Private Google Access, immutable Docker Artifact +account, exact `e2-standard-4` machine type, boot-disk attachment and +architecture, private-only VM attachment to `teleo-staging-europe-west6` +(`10.60.0.0/20`), Private Google Access, immutable Docker Artifact Registry repository, exact private PostgreSQL 16 Cloud SQL endpoint, host architecture, Docker server, Docker storage floor, unused service/container names, and private TCP/5432 reachability from the VM. diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index 93ef22a..d4ce310 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -72,6 +72,14 @@ cut over safely and retire the VPS after soak. reading secrets, logging into PostgreSQL, deploying, or restarting. Its bounded IAP/SSH path uses a fixed remote command and a maximum five-minute key lifetime, but no live run is currently authorized. +- Historical only: the superseded preflight branch records an earlier live + observation of `e2-standard-4` on `teleo-staging-europe-west6` + (`10.60.0.0/20`) with an external IPv4 attachment. That observation is not + current-revision proof and has not been refreshed in this slice. The checker + intentionally fails while any public attachment remains. +- The older provisioning baseline names `e2-standard-2` and + `teleo-staging-subnet`; reconciling that declarative baseline is a separate + infrastructure change, not part of this preflight implementation. - Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud SQL effective identity, restart behavior, or model/database parity. diff --git a/ops/check_gcp_leoclean_nosend_preflight.py b/ops/check_gcp_leoclean_nosend_preflight.py index 48db655..4c52abb 100644 --- a/ops/check_gcp_leoclean_nosend_preflight.py +++ b/ops/check_gcp_leoclean_nosend_preflight.py @@ -32,14 +32,14 @@ CRITICAL_TRACKED_PATHS = ( "hermes-agent/leoclean-bin/cloudsql_memory_tool.py", ) REGION = package.ZONE.rsplit("-", 1)[0] -MACHINE_TYPE = "e2-standard-2" +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-subnet" -SUBNET_CIDR = ipaddress.ip_network("10.70.0.0/24") +SUBNET = "teleo-staging-europe-west6" +SUBNET_CIDR = ipaddress.ip_network("10.60.0.0/20") CLOUDSQL_INSTANCE = "teleo-pgvector-standby" ARTIFACT_REPOSITORY = "teleo" @@ -190,7 +190,7 @@ def _instance_readback(runner: Runner) -> dict[str, Any]: package.PROJECT, "--zone", package.ZONE, - "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts.email,disks,networkInterfaces,tags.items)", + "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts,disks,networkInterfaces,tags.items)", ], runner, ) diff --git a/tests/test_gcp_leoclean_nosend_preflight.py b/tests/test_gcp_leoclean_nosend_preflight.py index 515dcff..59aa584 100644 --- a/tests/test_gcp_leoclean_nosend_preflight.py +++ b/tests/test_gcp_leoclean_nosend_preflight.py @@ -27,7 +27,7 @@ def live_fixtures() -> dict[str, object]: "zone": "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a", "machineType": ( "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a/" - "machineTypes/e2-standard-2" + "machineTypes/e2-standard-4" ), "cpuPlatform": "Intel Broadwell", "serviceAccounts": [ @@ -49,9 +49,9 @@ def live_fixtures() -> dict[str, object]: ), "subnetwork": ( "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6/" - "subnetworks/teleo-staging-subnet" + "subnetworks/teleo-staging-europe-west6" ), - "networkIP": "10.70.0.2", + "networkIP": "10.60.0.2", "accessConfigs": [], "ipv6AccessConfigs": [], "stackType": "IPV4_ONLY", @@ -71,10 +71,10 @@ def live_fixtures() -> dict[str, object]: ], }, "subnet": { - "name": "teleo-staging-subnet", + "name": "teleo-staging-europe-west6", "region": "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6", "network": "https://www.googleapis.com/compute/v1/projects/teleo-501523/global/networks/teleo-staging-net", - "ipCidrRange": "10.70.0.0/24", + "ipCidrRange": "10.60.0.0/20", "privateIpGoogleAccess": True, }, "artifact": { @@ -202,7 +202,7 @@ def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None: "teleo-501523", "--zone", "europe-west6-a", - "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts.email,disks,networkInterfaces,tags.items)", + "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts,disks,networkInterfaces,tags.items)", ], [ "gcloud", @@ -222,7 +222,7 @@ def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None: "networks", "subnets", "describe", - "teleo-staging-subnet", + "teleo-staging-europe-west6", "--project", "teleo-501523", "--region", From 16b32193f7951272f16c62eb41418ff0c6602b66 Mon Sep 17 00:00:00 2001 From: fwazb Date: Mon, 20 Jul 2026 23:30:14 -0700 Subject: [PATCH 3/4] Harden staging preflight receipt publication --- docs/gcp-leoclean-nosend-preflight.md | 31 ++-- docs/workstreams/gcp-leoclean/goals.md | 17 +- ops/check_gcp_leoclean_nosend_preflight.py | 167 ++++++++++++++++++- tests/test_gcp_leoclean_nosend_preflight.py | 171 +++++++++++++++++++- 4 files changed, 361 insertions(+), 25 deletions(-) diff --git a/docs/gcp-leoclean-nosend-preflight.md b/docs/gcp-leoclean-nosend-preflight.md index 8444616..36ac724 100644 --- a/docs/gcp-leoclean-nosend-preflight.md +++ b/docs/gcp-leoclean-nosend-preflight.md @@ -33,15 +33,25 @@ Run from the exact clean, tracked source revision with an already authenticated Cloud account that has the reviewed IAP/OS Login read path: ```bash +receipt_dir="$(mktemp -d "${TMPDIR:-/tmp}/livingip-leoclean-preflight.XXXXXX")" +chmod 0700 "$receipt_dir" python3 ops/check_gcp_leoclean_nosend_preflight.py \ - --output /tmp/leoclean-nosend-gcp-preflight.json + --output "$receipt_dir/preflight.json" ``` -The output file is created with mode `0600`. Authentication, permission, and -connection failures emit only a bounded category; command stderr is not copied -into the receipt. The checker rejects an alternate repository root, an output -path inside the source tree, untracked critical inputs, and source drift during -the live observation. +The output parent must already exist, be owned by the operator, and have no +group/other write permission. The output name must not exist. The checker +creates it once with no-clobber/no-symlink semantics, keeps the same file and +directory identity open through publication, writes mode `0600`, and fsyncs +the file and directory. It never removes or overwrites the output pathname; +an ambiguous publication failure can therefore leave a private residue for +operator inspection. + +Authentication, permission, connection, and receipt-publication failures emit +only a bounded category; command stderr and filesystem exception details are +not copied into the receipt or stdout. The checker rejects an alternate +repository root, an output path inside the source tree, untracked critical +inputs, and source drift during the live observation. ## Claim ceiling and rollback @@ -52,7 +62,8 @@ secret access, database authentication, scoped PostgreSQL permissions, image publication, deployment, restart, rollback, model behavior, parity, traffic, or production readiness. -This slice changes no repository-independent service state. Its local rollback -boundary is deletion of the branch/worktree and any generated receipt. Any -time-bounded SSH key created by the access path expires after at most five -minutes. +This slice changes no repository-independent service state beyond the stated +possibility of a time-bounded SSH key. Its local rollback boundary is deletion +of the branch/worktree; generated receipts and ambiguous private residues are +operator-owned evidence and are not deleted automatically. Any time-bounded +SSH key created by the access path expires after at most five minutes. diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index d4ce310..b5afa94 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -71,7 +71,22 @@ cut over safely and retire the VPS after soak. #210 head. It checks exact control-plane and host prerequisites without reading secrets, logging into PostgreSQL, deploying, or restarting. Its bounded IAP/SSH path uses a fixed remote command and a maximum five-minute - key lifetime, but no live run is currently authorized. + key lifetime. Receipt publication now requires a pre-existing private + operator-owned directory, creates one mode-`0600` no-clobber/no-symlink + file through a held directory descriptor, verifies stable identity, fsyncs + both file and directory, and preserves ambiguous residues. No live run is + currently authorized. +- Artifact Registry cleanup policy is not an additional hard gate for the + package: immutable tags prevent deletion of the retained tagged candidate. + The post-push/finalization path must still re-prove the exact retained + candidate tag-to-digest binding immediately before deployment. +- Current offline evidence for the preflight extraction: 243 focused + preflight/package/OCI tests pass; Ruff, Python compilation, formatting, and + diff checks pass. The repository suite produced 2,497 passes and 5 expected + skips inside the desktop sandbox; all 18 sandbox-denied Docker, loopback, and + Swift-cache cases passed in a separately authorized 59-test local rerun, + reconciling the suite to 2,515 passes and 5 skips. No cloud or registry + endpoint was contacted. - Historical only: the superseded preflight branch records an earlier live observation of `e2-standard-4` on `teleo-staging-europe-west6` (`10.60.0.0/20`) with an external IPv4 attachment. That observation is not diff --git a/ops/check_gcp_leoclean_nosend_preflight.py b/ops/check_gcp_leoclean_nosend_preflight.py index 4c52abb..1aeb62f 100644 --- a/ops/check_gcp_leoclean_nosend_preflight.py +++ b/ops/check_gcp_leoclean_nosend_preflight.py @@ -11,6 +11,7 @@ import json import os import re import shlex +import stat import subprocess import sys from collections.abc import Callable @@ -467,7 +468,10 @@ def _host_readback(runner: Runner, private_ip: str) -> dict[str, Any]: _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( + 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: @@ -553,6 +557,151 @@ def _validated_repo_root(repo_root: Path) -> Path: 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, *, @@ -591,34 +740,38 @@ def _source_revision( return source_revision -def main() -> int: +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() + 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 = args.output.expanduser().resolve() + 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 - if output_path: - output_path.parent.mkdir(parents=True, exist_ok=True) - package.write_json(output_path, payload, mode=0o600) + finally: + if output_parent_descriptor is not None: + os.close(output_parent_descriptor) print(json.dumps(payload, indent=2, sort_keys=True)) return 0 diff --git a/tests/test_gcp_leoclean_nosend_preflight.py b/tests/test_gcp_leoclean_nosend_preflight.py index 59aa584..e9fdad6 100644 --- a/tests/test_gcp_leoclean_nosend_preflight.py +++ b/tests/test_gcp_leoclean_nosend_preflight.py @@ -5,6 +5,8 @@ from __future__ import annotations import copy import hashlib import json +import os +import stat import subprocess import sys from collections.abc import Callable @@ -30,9 +32,7 @@ def live_fixtures() -> dict[str, object]: "machineTypes/e2-standard-4" ), "cpuPlatform": "Intel Broadwell", - "serviceAccounts": [ - {"email": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com"} - ], + "serviceAccounts": [{"email": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com"}], "disks": [ { "boot": True, @@ -169,9 +169,10 @@ def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None: "ssh_key_maximum_lifetime": "5m", } assert payload["host"]["cloudsql_private_tcp_5432"] == "reachable" - assert payload["control_plane"]["cloudsql"]["private_ip_sha256"] == hashlib.sha256( - PRIVATE_IP.encode("ascii") - ).hexdigest() + assert ( + payload["control_plane"]["cloudsql"]["private_ip_sha256"] + == hashlib.sha256(PRIVATE_IP.encode("ascii")).hexdigest() + ) assert PRIVATE_IP not in json.dumps(payload) assert len(calls) == 7 assert calls[:-1] == [ @@ -446,7 +447,12 @@ def test_preflight_rejects_an_unrelated_clean_repository(tmp_path: Path) -> None def test_direct_cli_imports_then_rejects_an_unrelated_repository(tmp_path: Path) -> None: completed = subprocess.run( - [sys.executable, str(preflight.SOURCE_ROOT / "ops" / "check_gcp_leoclean_nosend_preflight.py"), "--repo-root", str(tmp_path)], + [ + sys.executable, + str(preflight.SOURCE_ROOT / "ops" / "check_gcp_leoclean_nosend_preflight.py"), + "--repo-root", + str(tmp_path), + ], text=True, capture_output=True, check=False, @@ -457,3 +463,154 @@ def test_direct_cli_imports_then_rejects_an_unrelated_repository(tmp_path: Path) "schema": "livingip.leocleanNoSendGcpPreflight.v1", "status": "blocked", } + + +def _private_output_directory(tmp_path: Path) -> Path: + directory = tmp_path / "private-receipts" + directory.mkdir(mode=0o700) + directory.chmod(0o700) + return directory + + +def _stub_successful_cli(monkeypatch: pytest.MonkeyPatch) -> dict[str, object]: + payload: dict[str, object] = { + "schema": preflight.SCHEMA, + "status": "pass", + "source_revision": REVISION, + } + monkeypatch.setattr(preflight, "_source_revision", lambda *_args, **_kwargs: REVISION) + monkeypatch.setattr(preflight, "build_preflight", lambda **_kwargs: payload) + return payload + + +def test_cli_publishes_one_private_no_clobber_receipt( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + payload = _stub_successful_cli(monkeypatch) + output = _private_output_directory(tmp_path) / "preflight.json" + + assert preflight.main(["--output", str(output)]) == 0 + + assert json.loads(capsys.readouterr().out) == payload + assert json.loads(output.read_text(encoding="utf-8")) == payload + observed = output.stat(follow_symlinks=False) + assert stat.S_ISREG(observed.st_mode) + assert stat.S_IMODE(observed.st_mode) == 0o600 + assert observed.st_uid == os.geteuid() + assert observed.st_nlink == 1 + + +@pytest.mark.parametrize("entry_kind", ["file", "directory", "symlink"]) +def test_cli_rejects_every_preexisting_output_before_live_observation( + entry_kind: str, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = _private_output_directory(tmp_path) / "preflight.json" + if entry_kind == "file": + output.write_text("existing\n", encoding="utf-8") + elif entry_kind == "directory": + output.mkdir() + else: + target = output.with_name("target.json") + target.write_text("target\n", encoding="utf-8") + output.symlink_to(target) + called = False + + def unexpected_live_observation(**_kwargs: object) -> dict[str, object]: + nonlocal called + called = True + raise AssertionError("live observation must not run") + + monkeypatch.setattr(preflight, "build_preflight", unexpected_live_observation) + + assert preflight.main(["--output", str(output)]) == 1 + + assert called is False + assert json.loads(capsys.readouterr().out) == { + "category": "invalid_output_path", + "schema": preflight.SCHEMA, + "status": "blocked", + } + + +@pytest.mark.parametrize("unsafe_mode", [0o720, 0o702, 0o777]) +def test_cli_rejects_group_or_world_writable_output_parent( + unsafe_mode: int, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + parent = _private_output_directory(tmp_path) + parent.chmod(unsafe_mode) + called = False + + def unexpected_live_observation(**_kwargs: object) -> dict[str, object]: + nonlocal called + called = True + raise AssertionError("live observation must not run") + + monkeypatch.setattr(preflight, "build_preflight", unexpected_live_observation) + + assert preflight.main(["--output", str(parent / "preflight.json")]) == 1 + + assert called is False + assert json.loads(capsys.readouterr().out)["category"] == "invalid_output_path" + + +def test_publication_race_fails_without_overwriting_the_created_entry( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = _private_output_directory(tmp_path) / "preflight.json" + monkeypatch.setattr(preflight, "_source_revision", lambda *_args, **_kwargs: REVISION) + + def race(**_kwargs: object) -> dict[str, object]: + output.write_text("replacement-sentinel\n", encoding="utf-8") + return {"schema": preflight.SCHEMA, "status": "pass", "source_revision": REVISION} + + monkeypatch.setattr(preflight, "build_preflight", race) + + assert preflight.main(["--output", str(output)]) == 1 + + assert output.read_text(encoding="utf-8") == "replacement-sentinel\n" + assert json.loads(capsys.readouterr().out) == { + "category": "receipt_publication_failed", + "schema": preflight.SCHEMA, + "status": "blocked", + } + + +def test_publication_failure_is_bounded_and_preserves_created_residue( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + _stub_successful_cli(monkeypatch) + output = _private_output_directory(tmp_path) / "preflight.json" + real_fsync = preflight.os.fsync + sentinel = "private-publication-fault-sentinel" + + def fail_file_fsync(descriptor: int) -> None: + if stat.S_ISREG(os.fstat(descriptor).st_mode): + raise OSError(sentinel) + real_fsync(descriptor) + + monkeypatch.setattr(preflight.os, "fsync", fail_file_fsync) + + assert preflight.main(["--output", str(output)]) == 1 + + stdout = capsys.readouterr().out + assert json.loads(stdout) == { + "category": "receipt_publication_failed", + "schema": preflight.SCHEMA, + "status": "blocked", + } + assert sentinel not in stdout + assert output.is_file() + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + assert json.loads(output.read_text(encoding="utf-8"))["status"] == "pass" From 2f6cfeaaa41beb14c1057dac23c67e2b61dee7f5 Mon Sep 17 00:00:00 2001 From: fwazb Date: Mon, 20 Jul 2026 23:43:10 -0700 Subject: [PATCH 4/4] Close staging preflight review gaps --- docs/gcp-leoclean-nosend-preflight.md | 17 ++- docs/workstreams/gcp-leoclean/goals.md | 15 ++- ops/check_gcp_leoclean_nosend_preflight.py | 126 +++++++++++++++++++- tests/test_gcp_leoclean_nosend_preflight.py | 121 ++++++++++++++++++- 4 files changed, 257 insertions(+), 22 deletions(-) diff --git a/docs/gcp-leoclean-nosend-preflight.md b/docs/gcp-leoclean-nosend-preflight.md index 36ac724..a82622b 100644 --- a/docs/gcp-leoclean-nosend-preflight.md +++ b/docs/gcp-leoclean-nosend-preflight.md @@ -12,7 +12,11 @@ architecture, private-only VM attachment to `teleo-staging-europe-west6` (`10.60.0.0/20`), Private Google Access, immutable Docker Artifact Registry repository, exact private PostgreSQL 16 Cloud SQL endpoint, host architecture, Docker server, Docker storage floor, unused service/container -names, and private TCP/5432 reachability from the VM. +names, and private TCP/5432 reachability from the VM. Before the first Docker +client command, it requires `docker.service` to be loaded, active, and running; +after the final client command it requires unchanged service state, `MainPID`, +and restart count. This prevents a read probe from activating Docker through +`docker.socket` or silently accepting a daemon restart. ## Authority boundary @@ -42,10 +46,13 @@ python3 ops/check_gcp_leoclean_nosend_preflight.py \ The output parent must already exist, be owned by the operator, and have no group/other write permission. The output name must not exist. The checker creates it once with no-clobber/no-symlink semantics, keeps the same file and -directory identity open through publication, writes mode `0600`, and fsyncs -the file and directory. It never removes or overwrites the output pathname; -an ambiguous publication failure can therefore leave a private residue for -operator inspection. +directory identity open through publication, writes and fsyncs a mode-`0600` +private staged inode, and only then hard-links that complete inode to the +canonical output name without replacement. The retained staged and canonical +names are two links to the same inode; they are never removed automatically. +A pre-publication failure therefore leaves only the private staged residue, +not a valid-looking canonical pass receipt. A later ambiguous publication +failure preserves the exact retained inode for operator inspection. Authentication, permission, connection, and receipt-publication failures emit only a bounded category; command stderr and filesystem exception details are diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index b5afa94..5d5876d 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -73,19 +73,22 @@ cut over safely and retire the VPS after soak. bounded IAP/SSH path uses a fixed remote command and a maximum five-minute key lifetime. Receipt publication now requires a pre-existing private operator-owned directory, creates one mode-`0600` no-clobber/no-symlink - file through a held directory descriptor, verifies stable identity, fsyncs - both file and directory, and preserves ambiguous residues. No live run is - currently authorized. + staged inode through a held directory descriptor, fsyncs it before a + no-clobber canonical hard link, and retains both links without automatic + cleanup. The host probe proves Docker loaded/active/running before its first + client call and unchanged service state, `MainPID`, and restart count after + the last, preventing socket activation from changing runtime state. No live + run is currently authorized. - Artifact Registry cleanup policy is not an additional hard gate for the package: immutable tags prevent deletion of the retained tagged candidate. The post-push/finalization path must still re-prove the exact retained candidate tag-to-digest binding immediately before deployment. -- Current offline evidence for the preflight extraction: 243 focused +- Current offline evidence for the preflight extraction: 250 focused preflight/package/OCI tests pass; Ruff, Python compilation, formatting, and - diff checks pass. The repository suite produced 2,497 passes and 5 expected + diff checks pass. The repository suite produced 2,504 passes and 5 expected skips inside the desktop sandbox; all 18 sandbox-denied Docker, loopback, and Swift-cache cases passed in a separately authorized 59-test local rerun, - reconciling the suite to 2,515 passes and 5 skips. No cloud or registry + reconciling the suite to 2,522 passes and 5 skips. No cloud or registry endpoint was contacted. - Historical only: the superseded preflight branch records an earlier live observation of `e2-standard-4` on `teleo-staging-europe-west6` diff --git a/ops/check_gcp_leoclean_nosend_preflight.py b/ops/check_gcp_leoclean_nosend_preflight.py index 1aeb62f..8b7a48f 100644 --- a/ops/check_gcp_leoclean_nosend_preflight.py +++ b/ops/check_gcp_leoclean_nosend_preflight.py @@ -14,6 +14,7 @@ import shlex import stat import subprocess import sys +import uuid from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path @@ -397,6 +398,16 @@ project=\"$(metadata project/project-id)\" instance=\"$(metadata instance/name)\" service_account=\"$(metadata instance/service-accounts/default/email)\" test -x /usr/bin/docker +docker_service_load_state_before=\"$(systemctl show docker.service --property=LoadState --value)\" +docker_service_active_state_before=\"$(systemctl show docker.service --property=ActiveState --value)\" +docker_service_sub_state_before=\"$(systemctl show docker.service --property=SubState --value)\" +docker_service_main_pid_before=\"$(systemctl show docker.service --property=MainPID --value)\" +docker_service_n_restarts_before=\"$(systemctl show docker.service --property=NRestarts --value)\" +test \"$docker_service_load_state_before\" = loaded +test \"$docker_service_active_state_before\" = active +test \"$docker_service_sub_state_before\" = running +test \"$docker_service_main_pid_before\" -gt 1 +test \"$docker_service_n_restarts_before\" -ge 0 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}}}}')\" @@ -404,11 +415,26 @@ 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 ' ')\" +docker_service_load_state_after=\"$(systemctl show docker.service --property=LoadState --value)\" +docker_service_active_state_after=\"$(systemctl show docker.service --property=ActiveState --value)\" +docker_service_sub_state_after=\"$(systemctl show docker.service --property=SubState --value)\" +docker_service_main_pid_after=\"$(systemctl show docker.service --property=MainPID --value)\" +docker_service_n_restarts_after=\"$(systemctl show docker.service --property=NRestarts --value)\" +test \"$docker_service_load_state_after\" = \"$docker_service_load_state_before\" +test \"$docker_service_active_state_after\" = \"$docker_service_active_state_before\" +test \"$docker_service_sub_state_after\" = \"$docker_service_sub_state_before\" +test \"$docker_service_main_pid_after\" = \"$docker_service_main_pid_before\" +test \"$docker_service_n_restarts_after\" = \"$docker_service_n_restarts_before\" 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_service_load_state=%s\n' \"$docker_service_load_state_after\" +printf 'docker_service_active_state=%s\n' \"$docker_service_active_state_after\" +printf 'docker_service_sub_state=%s\n' \"$docker_service_sub_state_after\" +printf 'docker_service_main_pid=%s\n' \"$docker_service_main_pid_after\" +printf 'docker_service_n_restarts=%s\n' \"$docker_service_n_restarts_after\" 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\" @@ -448,6 +474,11 @@ def _parse_host_output(output: str) -> dict[str, str]: "project", "instance", "service_account", + "docker_service_load_state", + "docker_service_active_state", + "docker_service_sub_state", + "docker_service_main_pid", + "docker_service_n_restarts", "docker_server_version", "docker_server_os", "docker_server_arch", @@ -477,8 +508,15 @@ def _host_readback(runner: Runner, private_ip: str) -> dict[str, Any]: try: free_bytes = int(value["docker_root_free_bytes"]) collision_count = int(value["container_collision_count"]) + docker_main_pid = int(value["docker_service_main_pid"]) + docker_n_restarts = int(value["docker_service_n_restarts"]) except ValueError as exc: raise PreflightError("invalid_readback", "host numeric field was invalid") from exc + _require(value["docker_service_load_state"] == "loaded", "Docker service is not loaded") + _require(value["docker_service_active_state"] == "active", "Docker service is not active") + _require(value["docker_service_sub_state"] == "running", "Docker service is not running") + _require(docker_main_pid > 1, "Docker service MainPID is invalid") + _require(docker_n_restarts >= 0, "Docker service restart count is invalid") _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") @@ -491,6 +529,14 @@ def _host_readback(runner: Runner, private_ip: str) -> dict[str, Any]: "service_account": package.SERVICE_ACCOUNT, }, "docker": { + "service": { + "load_state": "loaded", + "active_state": "active", + "sub_state": "running", + "main_pid": docker_main_pid, + "n_restarts": docker_n_restarts, + "stable_across_client_observation": True, + }, "server_version": value["docker_server_version"], "server_os": "linux", "server_architecture": "amd64", @@ -535,6 +581,8 @@ def build_preflight(*, runner: Runner = _default_runner, source_revision: str) - "database_credentials_used": False, "database_queries_run": False, "control_plane_observations_before_ssh_are_describe_only": True, + "docker_service_proven_active_before_client_calls": True, + "docker_service_identity_stable_after_client_calls": True, "remote_host_runtime_mutations": False, "ssh_access_may_register_time_bounded_key": True, "ssh_key_maximum_lifetime": SSH_KEY_LIFETIME, @@ -642,6 +690,37 @@ def _write_all(descriptor: int, payload: bytes) -> None: offset += written +def _verify_receipt_entry( + parent_descriptor: int, + name: str, + expected_identity: tuple[int, int], + *, + expected_size: int, + expected_links: int, + label: str, +) -> os.stat_result: + try: + observed = os.stat(name, dir_fd=parent_descriptor, follow_symlinks=False) + except OSError as exc: + raise PreflightError( + "receipt_publication_failed", + f"{label} cannot be identified safely", + ) from exc + if not ( + stat.S_ISREG(observed.st_mode) + and (observed.st_dev, observed.st_ino) == expected_identity + and observed.st_uid == os.geteuid() + and stat.S_IMODE(observed.st_mode) == 0o600 + and observed.st_nlink == expected_links + and observed.st_size == expected_size + ): + raise PreflightError( + "receipt_publication_failed", + f"{label} identity or posture drifted", + ) + return observed + + def _publish_private_receipt( output_path: Path, parent_descriptor: int, @@ -656,6 +735,7 @@ def _publish_private_receipt( ) from exc descriptor: int | None = None + staged_name = f".{output_path.name}.{uuid.uuid4().hex}.tmp" try: _assert_private_output_parent(parent_descriptor, output_path.parent) if _output_entry_stat(parent_descriptor, output_path.name) is not None: @@ -666,19 +746,16 @@ def _publish_private_receipt( 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) + descriptor = os.open(staged_name, flags, 0o600, dir_fd=parent_descriptor) os.fchmod(descriptor, 0o600) before = os.fstat(descriptor) + identity = (before.st_dev, before.st_ino) _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 @@ -686,7 +763,44 @@ def _publish_private_receipt( ): raise PreflightError( "receipt_publication_failed", - "published preflight receipt identity or posture drifted", + "staged preflight receipt identity or posture drifted", + ) + _verify_receipt_entry( + parent_descriptor, + staged_name, + identity, + expected_size=len(encoded), + expected_links=1, + label="staged preflight receipt", + ) + os.link( + staged_name, + output_path.name, + src_dir_fd=parent_descriptor, + dst_dir_fd=parent_descriptor, + follow_symlinks=False, + ) + _verify_receipt_entry( + parent_descriptor, + staged_name, + identity, + expected_size=len(encoded), + expected_links=2, + label="retained staged preflight receipt", + ) + _verify_receipt_entry( + parent_descriptor, + output_path.name, + identity, + expected_size=len(encoded), + expected_links=2, + label="published preflight receipt", + ) + through_path = os.stat(output_path, follow_symlinks=False) + if (through_path.st_dev, through_path.st_ino) != identity: + raise PreflightError( + "receipt_publication_failed", + "published preflight receipt path identity drifted", ) _assert_private_output_parent(parent_descriptor, output_path.parent) os.fsync(parent_descriptor) diff --git a/tests/test_gcp_leoclean_nosend_preflight.py b/tests/test_gcp_leoclean_nosend_preflight.py index e9fdad6..75e7278 100644 --- a/tests/test_gcp_leoclean_nosend_preflight.py +++ b/tests/test_gcp_leoclean_nosend_preflight.py @@ -102,6 +102,11 @@ def live_fixtures() -> dict[str, object]: "project": "teleo-501523", "instance": "teleo-staging-1", "service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com", + "docker_service_load_state": "loaded", + "docker_service_active_state": "active", + "docker_service_sub_state": "running", + "docker_service_main_pid": "1701", + "docker_service_n_restarts": "0", "docker_server_version": "29.4.0", "docker_server_os": "linux", "docker_server_arch": "amd64", @@ -164,11 +169,21 @@ def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None: "database_credentials_used": False, "database_queries_run": False, "control_plane_observations_before_ssh_are_describe_only": True, + "docker_service_proven_active_before_client_calls": True, + "docker_service_identity_stable_after_client_calls": True, "remote_host_runtime_mutations": False, "ssh_access_may_register_time_bounded_key": True, "ssh_key_maximum_lifetime": "5m", } assert payload["host"]["cloudsql_private_tcp_5432"] == "reachable" + assert payload["host"]["docker"]["service"] == { + "load_state": "loaded", + "active_state": "active", + "sub_state": "running", + "main_pid": 1701, + "n_restarts": 0, + "stable_across_client_observation": True, + } assert ( payload["control_plane"]["cloudsql"]["private_ip_sha256"] == hashlib.sha256(PRIVATE_IP.encode("ascii")).hexdigest() @@ -248,7 +263,30 @@ def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None: assert "--tunnel-through-iap" in ssh assert "--ssh-key-expire-after=5m" in ssh remote = next(item.removeprefix("--command=") for item in ssh if item.startswith("--command=")) + syntax = subprocess.run(["/bin/bash", "-n"], input=remote, text=True, capture_output=True, check=False) + assert syntax.returncode == 0, syntax.stderr assert remote.startswith("set -euo pipefail\n") + first_docker_client = remote.index("sudo -n /usr/bin/docker version") + assert remote.index("docker_service_active_state_before=") < first_docker_client + assert remote.index('test "$docker_service_active_state_before" = active') < first_docker_client + assert remote.index('test "$docker_service_sub_state_before" = running') < first_docker_client + assert remote.index('test "$docker_service_main_pid_before" -gt 1') < first_docker_client + last_docker_client = remote.index("sudo -n /usr/bin/docker ps") + assert remote.index("docker_service_active_state_after=", last_docker_client) > last_docker_client + assert ( + remote.index( + 'test "$docker_service_main_pid_after" = "$docker_service_main_pid_before"', + last_docker_client, + ) + > last_docker_client + ) + assert ( + remote.index( + 'test "$docker_service_n_restarts_after" = "$docker_service_n_restarts_before"', + last_docker_client, + ) + > last_docker_client + ) for forbidden in ( "docker run", "docker pull", @@ -279,6 +317,11 @@ def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None: (("artifact", "dockerConfig"), "immutableTags", False, "not immutable"), (("cloudsql", "settings", "ipConfiguration"), "ipv4Enabled", True, "public IPv4"), (("cloudsql",), "databaseVersion", "POSTGRES_160", "exactly PostgreSQL 16"), + (("host",), "docker_service_load_state", "not-found", "not loaded"), + (("host",), "docker_service_active_state", "inactive", "not active"), + (("host",), "docker_service_sub_state", "dead", "not running"), + (("host",), "docker_service_main_pid", "0", "MainPID"), + (("host",), "docker_service_n_restarts", "-1", "restart count"), (("host",), "container_collision_count", "1", "container name"), (("host",), "unit_load_state", "loaded", "unit name"), (("host",), "docker_root_free_bytes", "1024", "free space"), @@ -499,7 +542,10 @@ def test_cli_publishes_one_private_no_clobber_receipt( assert stat.S_ISREG(observed.st_mode) assert stat.S_IMODE(observed.st_mode) == 0o600 assert observed.st_uid == os.geteuid() - assert observed.st_nlink == 1 + assert observed.st_nlink == 2 + retained = list(output.parent.glob(".preflight.json.*.tmp")) + assert len(retained) == 1 + assert retained[0].stat().st_ino == observed.st_ino @pytest.mark.parametrize("entry_kind", ["file", "directory", "symlink"]) @@ -578,6 +624,7 @@ def test_publication_race_fails_without_overwriting_the_created_entry( assert preflight.main(["--output", str(output)]) == 1 assert output.read_text(encoding="utf-8") == "replacement-sentinel\n" + assert list(output.parent.glob(".preflight.json.*.tmp")) == [] assert json.loads(capsys.readouterr().out) == { "category": "receipt_publication_failed", "schema": preflight.SCHEMA, @@ -585,7 +632,35 @@ def test_publication_race_fails_without_overwriting_the_created_entry( } -def test_publication_failure_is_bounded_and_preserves_created_residue( +def test_link_race_preserves_the_complete_staged_inode_without_overwrite( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = _private_output_directory(tmp_path) / "preflight.json" + parent_descriptor = preflight._open_private_output_parent(output) + payload = {"schema": preflight.SCHEMA, "status": "pass", "source_revision": REVISION} + real_link = preflight.os.link + + def race_link(source: str, destination: str, **kwargs: object) -> None: + output.write_text("replacement-sentinel\n", encoding="utf-8") + real_link(source, destination, **kwargs) + + monkeypatch.setattr(preflight.os, "link", race_link) + try: + with pytest.raises(preflight.PreflightError) as caught: + preflight._publish_private_receipt(output, parent_descriptor, payload) + finally: + os.close(parent_descriptor) + + assert caught.value.category == "receipt_publication_failed" + assert output.read_text(encoding="utf-8") == "replacement-sentinel\n" + retained = list(output.parent.glob(".preflight.json.*.tmp")) + assert len(retained) == 1 + assert json.loads(retained[0].read_text(encoding="utf-8")) == payload + assert retained[0].stat().st_nlink == 1 + + +def test_prepublication_fsync_failure_is_bounded_and_preserves_only_staged_residue( tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, @@ -611,6 +686,42 @@ def test_publication_failure_is_bounded_and_preserves_created_residue( "status": "blocked", } assert sentinel not in stdout - assert output.is_file() - assert stat.S_IMODE(output.stat().st_mode) == 0o600 - assert json.loads(output.read_text(encoding="utf-8"))["status"] == "pass" + assert not output.exists() + retained = list(output.parent.glob(".preflight.json.*.tmp")) + assert len(retained) == 1 + assert stat.S_IMODE(retained[0].stat().st_mode) == 0o600 + assert retained[0].stat().st_nlink == 1 + assert json.loads(retained[0].read_text(encoding="utf-8"))["status"] == "pass" + + +def test_prepublication_write_failure_is_bounded_and_never_creates_canonical_output( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + _stub_successful_cli(monkeypatch) + output = _private_output_directory(tmp_path) / "preflight.json" + sentinel = "private-write-fault-sentinel" + + def fail_after_partial_write(descriptor: int, payload: bytes) -> None: + os.write(descriptor, payload[:16]) + raise OSError(sentinel) + + monkeypatch.setattr(preflight, "_write_all", fail_after_partial_write) + + assert preflight.main(["--output", str(output)]) == 1 + + stdout = capsys.readouterr().out + assert json.loads(stdout) == { + "category": "receipt_publication_failed", + "schema": preflight.SCHEMA, + "status": "blocked", + } + assert sentinel not in stdout + assert not output.exists() + retained = list(output.parent.glob(".preflight.json.*.tmp")) + assert len(retained) == 1 + assert stat.S_IMODE(retained[0].stat().st_mode) == 0o600 + assert retained[0].stat().st_nlink == 1 + with pytest.raises(json.JSONDecodeError): + json.loads(retained[0].read_text(encoding="utf-8"))