"""Contracts for the read-only leoclean GCP staging preflight.""" from __future__ import annotations import copy import hashlib import json import os import stat 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-4" ), "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-europe-west6" ), "networkIP": "10.60.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-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.60.0.0/20", "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_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", "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, "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() ) 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,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-europe-west6", "--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=")) 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", "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",), "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"), (("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", } 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 == 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"]) 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 list(output.parent.glob(".preflight.json.*.tmp")) == [] assert json.loads(capsys.readouterr().out) == { "category": "receipt_publication_failed", "schema": preflight.SCHEMA, "status": "blocked", } 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, ) -> 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 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"))