From 544b2f615db8794f57b22827a6f97e18efa0361c Mon Sep 17 00:00:00 2001 From: fwazb Date: Fri, 17 Jul 2026 17:17:50 -0700 Subject: [PATCH] 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", + }