From 5646bd5972f9164e2cd42545846eaeca46ef9d41 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Wed, 15 Jul 2026 22:54:50 +0200 Subject: [PATCH] feat: add fail-closed VPS GCP source drift detector --- ops/detect_vps_gcp_source_drift.py | 582 ++++++++++++++++++++++ tests/test_detect_vps_gcp_source_drift.py | 229 +++++++++ 2 files changed, 811 insertions(+) create mode 100644 ops/detect_vps_gcp_source_drift.py create mode 100644 tests/test_detect_vps_gcp_source_drift.py diff --git a/ops/detect_vps_gcp_source_drift.py b/ops/detect_vps_gcp_source_drift.py new file mode 100644 index 0000000..332fd84 --- /dev/null +++ b/ops/detect_vps_gcp_source_drift.py @@ -0,0 +1,582 @@ +#!/usr/bin/env python3 +"""Fail-closed, read-only VPS/Cloud SQL source fingerprint comparison.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import shlex +import subprocess +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +try: + from .private_receipt_io import print_private_receipt_summary, write_private_json + from .verify_postgres_parity_manifest import ( + REVIEWED_MANIFEST_SQL_SHA256, + SINGLETON_KINDS, + canonical_hash, + compare_manifests, + ) +except ImportError: # pragma: no cover - direct script execution + from private_receipt_io import print_private_receipt_summary, write_private_json + from verify_postgres_parity_manifest import ( + REVIEWED_MANIFEST_SQL_SHA256, + SINGLETON_KINDS, + canonical_hash, + compare_manifests, + ) + + +RECEIPT_SCHEMA = "livingip.vpsGcpSourceDrift.v1" +OBSERVATION_SCHEMA = "livingip.sourceFingerprintObservation.v1" +SAFE_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:@/+\-]{0,255}\Z") +COMMIT_RE = re.compile(r"[0-9a-f]{40}\Z") +ROUTE_PREFIX = "LEO_SOURCE_ROUTE\t" + + +class DriftDetectorError(RuntimeError): + """Raised for an invalid detector configuration or observation.""" + + +def utc_now() -> str: + return datetime.now(UTC).isoformat().replace("+00:00", "Z") + + +def sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def safe_value(value: str, flag: str) -> str: + if not SAFE_NAME_RE.fullmatch(value): + raise DriftDetectorError(f"{flag} contains an unsafe value") + return value + + +def parse_manifest_lines(lines: list[str], label: str) -> dict[str, Any]: + singleton: dict[str, dict[str, Any]] = {} + tables: dict[str, dict[str, Any]] = {} + performance: dict[str, dict[str, Any]] = {} + for line_number, raw_line in enumerate(lines, 1): + line = raw_line.strip() + if not line or line in {"BEGIN", "SET", "ROLLBACK"}: + continue + try: + row = json.loads(line) + except json.JSONDecodeError as exc: + raise DriftDetectorError(f"{label} manifest line {line_number} is not JSON") from exc + if not isinstance(row, dict): + raise DriftDetectorError(f"{label} manifest line {line_number} is not an object") + kind = row.get("kind") + if kind in SINGLETON_KINDS: + if kind in singleton: + raise DriftDetectorError(f"{label} manifest repeats {kind}") + singleton[str(kind)] = row + elif kind == "table": + key = f"{row.get('schema')}.{row.get('table')}" + if key in tables: + raise DriftDetectorError(f"{label} manifest repeats table {key}") + tables[key] = row + elif kind == "performance": + key = str(row.get("query")) + if key in performance: + raise DriftDetectorError(f"{label} manifest repeats performance row {key}") + performance[key] = row + else: + raise DriftDetectorError(f"{label} manifest has unknown kind {kind!r}") + missing = SINGLETON_KINDS - set(singleton) + if missing: + raise DriftDetectorError(f"{label} manifest is missing {sorted(missing)}") + if not tables: + raise DriftDetectorError(f"{label} manifest has no table rows") + return {"singleton": singleton, "tables": tables, "performance": performance} + + +def manifest_fingerprint(manifest: dict[str, Any]) -> dict[str, Any]: + tables = manifest["tables"] + table_rows = [tables[key] for key in sorted(tables)] + structure = {kind: manifest["singleton"][kind].get("items", []) for kind in sorted(SINGLETON_KINDS - {"identity"})} + return { + "table_count": len(tables), + "total_rows": sum(int(row["row_count"]) for row in table_rows), + "table_fingerprint_sha256": canonical_hash(table_rows), + "structure_fingerprint_sha256": canonical_hash(structure), + } + + +REMOTE_ROUTE_INVENTORY = r"""import json +import os +import re +import subprocess +import sys +from pathlib import Path + +service, route_kind, transport, expected_db = sys.argv[1:5] +raw = subprocess.check_output([ + "systemctl", "show", service, + "-p", "ActiveState", "-p", "SubState", "-p", "MainPID", "-p", "NRestarts", + "-p", "FragmentPath", "-p", "WorkingDirectory", "--no-pager", +], text=True, stderr=subprocess.DEVNULL).strip() +state = dict(line.split("=", 1) for line in raw.splitlines() if "=" in line) +pid = int(state.get("MainPID") or 0) +if state.get("ActiveState") != "active" or state.get("SubState") != "running" or pid <= 0: + raise SystemExit("runtime service is not active/running with a positive MainPID") +cwd = Path(f"/proc/{pid}/cwd").resolve() +commit_candidates = {} +try: + value = subprocess.check_output( + ["git", "-C", str(cwd), "rev-parse", "HEAD"], + text=True, + stderr=subprocess.DEVNULL, + ).strip() + if re.fullmatch(r"[0-9a-f]{40}", value): + commit_candidates[f"git:{cwd}"] = value +except (FileNotFoundError, subprocess.CalledProcessError): + pass +for candidate in ( + cwd / ".last-deploy-sha", + Path("/opt/teleo-eval/.last-deploy-sha"), + Path("/opt/teleo-codex/.last-deploy-sha"), +): + if candidate.is_file() and not candidate.is_symlink(): + value = candidate.read_text(encoding="utf-8", errors="replace").strip() + if re.fullmatch(r"[0-9a-f]{40}", value): + commit_candidates[str(candidate)] = value +commits = sorted(set(commit_candidates.values())) +payload = { + "schema": "livingip.remoteSourceRoute.v1", + "route_kind": route_kind, + "transport": transport, + "hostname": os.uname().nodename, + "service": service, + "service_state": state, + "process_cwd": str(cwd), + "expected_database": expected_db, + "runtime_commit": commits[0] if len(commits) == 1 else None, + "commit_candidates": commit_candidates, + "commit_identity_ambiguous": len(commits) != 1, + "secret_values_emitted": False, +} +print("LEO_SOURCE_ROUTE\t" + json.dumps(payload, sort_keys=True)) +""" + + +def ssh_base(target: str, *, identity_file: Path | None, timeout: int) -> list[str]: + command = [ + "ssh", + "-T", + "-o", + "BatchMode=yes", + "-o", + f"ConnectTimeout={timeout}", + ] + if identity_file is not None: + command.extend(["-i", str(identity_file)]) + command.extend([target, "--"]) + return command + + +def remote_script( + *, + route_kind: str, + transport: str, + service: str, + database: str, + container: str | None, + database_user: str, + gcp_host: str | None, + gcp_project: str | None, + gcp_secret: str | None, +) -> str: + inventory = " ".join( + [ + "python3", + "-c", + shlex.quote(REMOTE_ROUTE_INVENTORY), + shlex.quote(service), + shlex.quote(route_kind), + shlex.quote(transport), + shlex.quote(database), + ] + ) + if route_kind == "vps_docker": + assert container is not None + query = " ".join( + [ + "exec docker exec", + "-e", + shlex.quote("PGOPTIONS=-c default_transaction_read_only=on"), + "-i", + shlex.quote(container), + "psql -X -Atq -v ON_ERROR_STOP=1", + "-U", + shlex.quote(database_user), + "-d", + shlex.quote(database), + ] + ) + return f"set -euo pipefail\n{inventory}\n{query}\n" + assert gcp_host is not None and gcp_project is not None and gcp_secret is not None + connection = f"host={gcp_host} port=5432 dbname={database} user={database_user} sslmode=require connect_timeout=8" + return "\n".join( + [ + "set -euo pipefail", + inventory, + f'password="$(gcloud secrets versions access latest --secret={shlex.quote(gcp_secret)} ' + f'--project={shlex.quote(gcp_project)} --quiet)"', + "[[ -n \"$password\" ]] || { printf 'Cloud SQL credential resolved empty\\n' >&2; exit 2; }", + 'export PGPASSWORD="$password"', + "export PGOPTIONS='-c default_transaction_read_only=on'", + f"exec psql {shlex.quote(connection)} -X -Atq -v ON_ERROR_STOP=1", + "", + ] + ) + + +def sanitize_error(stderr: str, *, target: str) -> str: + line = next((line.strip() for line in reversed(stderr.splitlines()) if line.strip()), "probe failed") + line = re.sub(r"(?i)(password|token|secret|credential)=\S+", r"\1=[REDACTED]", line) + line = re.sub(r"(?i)postgres(?:ql)?://\S+", "postgresql://[REDACTED]", line) + return line.replace(target, "[TARGET]")[:800] + + +def collect_endpoint( + *, + label: str, + command: list[str], + manifest_sql: bytes, + timeout: int, + expected_target: dict[str, Any], +) -> dict[str, Any]: + started = utc_now() + try: + completed = subprocess.run( + command, + input=manifest_sql, + capture_output=True, + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired: + return { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "unreachable", + "observed_at_utc": started, + "failure_class": "probe_timeout", + "error": f"{label} probe exceeded {timeout}s", + "expected_target": expected_target, + "manifest": None, + "route": None, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + stdout = completed.stdout.decode(errors="replace") + stderr = completed.stderr.decode(errors="replace") + lines = stdout.splitlines() + route_line = next((line for line in lines if line.startswith(ROUTE_PREFIX)), None) + route = None + if route_line is not None: + try: + route = json.loads(route_line.removeprefix(ROUTE_PREFIX)) + except json.JSONDecodeError: + route = None + if not isinstance(route, dict): + route = None + if completed.returncode != 0 or route is None: + return { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "unreachable", + "observed_at_utc": started, + "failure_class": ( + "database_probe_failed_after_route_observed" + if route is not None + else "transport_or_remote_probe_failed" + ), + "exit_code": completed.returncode, + "error": sanitize_error(stderr, target=command[command.index("--") - 1]), + "expected_target": expected_target, + "manifest": None, + "route": route, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + try: + manifest = parse_manifest_lines([line for line in lines if line != route_line], label) + identity = manifest["singleton"]["identity"] + fingerprint = manifest_fingerprint(manifest) + if route.get("runtime_commit") is None or not COMMIT_RE.fullmatch(str(route["runtime_commit"])): + raise DriftDetectorError(f"{label} runtime commit identity is missing or ambiguous") + if identity.get("transaction_read_only") != "on": + raise DriftDetectorError(f"{label} database transaction was not read-only") + if identity.get("database") != route.get("expected_database"): + raise DriftDetectorError( + f"{label} connected database {identity.get('database')!r} does not match explicit target " + f"{route.get('expected_database')!r}" + ) + except (DriftDetectorError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + return { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "invalid", + "observed_at_utc": started, + "failure_class": "invalid_probe_receipt", + "error": str(exc), + "expected_target": expected_target, + "manifest": None, + "route": route, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + return { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "observed_current", + "observed_at_utc": utc_now(), + "expected_target": expected_target, + "route": route, + "database_identity": identity, + "fingerprint": fingerprint, + "manifest": manifest, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + "secret_values_emitted": False, + } + + +def public_observation(observation: dict[str, Any]) -> dict[str, Any]: + return {key: value for key, value in observation.items() if key != "manifest"} + + +def evaluate_observations( + vps: dict[str, Any], + gcp: dict[str, Any], + *, + generated_at_utc: str | None = None, +) -> dict[str, Any]: + unavailable = [item["label"] for item in (vps, gcp) if item.get("status") != "observed_current"] + receipt: dict[str, Any] = { + "schema": RECEIPT_SCHEMA, + "generated_at_utc": generated_at_utc or utc_now(), + "mode": "live_readonly_no_fallback", + "status": "incomplete" if unavailable else "pending_comparison", + "comparison": None, + "observations": {"vps": public_observation(vps), "gcp": public_observation(gcp)}, + "unavailable_targets": unavailable, + "fallback_used": False, + "historical_receipt_substituted": False, + "database_mutated": False, + "routing_mutated": False, + "secrets_logged": False, + "cleanup": { + "remote_files_created": False, + "local_temp_files_created": False, + "orphan_processes_expected": False, + }, + } + if unavailable: + receipt["decision"] = "No source-of-truth decision: every target must be observed live." + return receipt + problems, details = compare_manifests( + vps["manifest"], + gcp["manifest"], + max_target_query_ms=10_000.0, + max_target_source_ratio=1_000.0, + ) + content_identical = vps["fingerprint"]["table_fingerprint_sha256"] == gcp["fingerprint"]["table_fingerprint_sha256"] + structure_identical = ( + vps["fingerprint"]["structure_fingerprint_sha256"] == gcp["fingerprint"]["structure_fingerprint_sha256"] + ) + state = "identical" if not problems and content_identical and structure_identical else "drift" + receipt["status"] = state + receipt["comparison"] = { + "state": state, + "content_identical": content_identical, + "structure_identical": structure_identical, + "problems": problems, + "details": details, + } + receipt["decision"] = ( + "VPS and GCP manifests are identical under the reviewed parity contract." + if state == "identical" + else "Drift detected; do not promote or reroute until every listed mismatch is reconciled." + ) + return receipt + + +def run_live( + args: argparse.Namespace, + *, + collector: Callable[..., dict[str, Any]] = collect_endpoint, +) -> tuple[dict[str, Any], int]: + manifest_sql = args.manifest.read_bytes() + manifest_sha256 = sha256_bytes(manifest_sql) + if manifest_sha256 != REVIEWED_MANIFEST_SQL_SHA256: + raise DriftDetectorError( + "manifest SQL hash does not match the reviewed parity contract: " + f"expected={REVIEWED_MANIFEST_SQL_SHA256} actual={manifest_sha256}" + ) + vps_target = safe_value(args.vps_ssh_target, "--vps-ssh-target") + gcp_target = safe_value(args.gcp_ssh_target, "--gcp-ssh-target") + for value, flag in ( + (args.vps_container, "--vps-container"), + (args.vps_database, "--vps-database"), + (args.vps_service, "--vps-service"), + (args.vps_database_user, "--vps-database-user"), + (args.gcp_database, "--gcp-database"), + (args.gcp_service, "--gcp-service"), + (args.gcp_database_user, "--gcp-database-user"), + (args.gcp_host, "--gcp-host"), + (args.gcp_project, "--gcp-project"), + (args.gcp_secret, "--gcp-secret"), + ): + safe_value(value, flag) + if not args.vps_ssh_key.is_file(): + raise DriftDetectorError("--vps-ssh-key must be an existing file") + + vps_script = remote_script( + route_kind="vps_docker", + transport=vps_target, + service=args.vps_service, + database=args.vps_database, + container=args.vps_container, + database_user=args.vps_database_user, + gcp_host=None, + gcp_project=None, + gcp_secret=None, + ) + gcp_script = remote_script( + route_kind="gcp_private_cloudsql", + transport=gcp_target, + service=args.gcp_service, + database=args.gcp_database, + container=None, + database_user=args.gcp_database_user, + gcp_host=args.gcp_host, + gcp_project=args.gcp_project, + gcp_secret=args.gcp_secret, + ) + vps_remote_command = shlex.join(["bash", "-c", vps_script]) + gcp_remote_command = shlex.join(["sudo", "-n", "bash", "-c", gcp_script]) + vps_command = [ + *ssh_base(vps_target, identity_file=args.vps_ssh_key.resolve(), timeout=args.connect_timeout), + vps_remote_command, + ] + gcp_command = [ + *ssh_base(gcp_target, identity_file=None, timeout=args.connect_timeout), + gcp_remote_command, + ] + vps_expected_target = { + "transport": vps_target, + "runtime_service": args.vps_service, + "database_endpoint": f"docker:{args.vps_container}", + "database": args.vps_database, + "database_user": args.vps_database_user, + "identity_source": "explicit_cli_target", + } + gcp_expected_target = { + "transport": gcp_target, + "runtime_service": args.gcp_service, + "database_endpoint": f"{args.gcp_host}:5432", + "database": args.gcp_database, + "database_user": args.gcp_database_user, + "project": args.gcp_project, + "credential_source": f"secret-manager:{args.gcp_secret}", + "identity_source": "explicit_cli_target", + } + with ThreadPoolExecutor(max_workers=2, thread_name_prefix="source-drift") as pool: + vps_future = pool.submit( + collector, + label="vps", + command=vps_command, + manifest_sql=manifest_sql, + timeout=args.probe_timeout, + expected_target=vps_expected_target, + ) + gcp_future = pool.submit( + collector, + label="gcp", + command=gcp_command, + manifest_sql=manifest_sql, + timeout=args.probe_timeout, + expected_target=gcp_expected_target, + ) + vps = vps_future.result() + gcp = gcp_future.result() + receipt = evaluate_observations(vps, gcp) + receipt["manifest_sql_sha256"] = manifest_sha256 + return receipt, {"identical": 0, "drift": 1, "incomplete": 2}[receipt["status"]] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--manifest", default=Path("ops/postgres_parity_manifest.sql"), type=Path) + parser.add_argument("--connect-timeout", type=int, default=12) + parser.add_argument("--probe-timeout", type=int, default=180) + parser.add_argument("--vps-ssh-target", default="root@77.42.65.182") + parser.add_argument( + "--vps-ssh-key", + default=Path.home() / ".ssh/livingip_hetzner_20260604_ed25519", + type=Path, + ) + parser.add_argument("--vps-container", default="teleo-pg") + parser.add_argument("--vps-database", default="teleo") + parser.add_argument("--vps-database-user", default="postgres") + parser.add_argument("--vps-service", default="leoclean-gateway.service") + parser.add_argument("--gcp-ssh-target", default="teleo-gcp-staging") + parser.add_argument("--gcp-host", default="10.61.0.3") + parser.add_argument("--gcp-database", required=True) + parser.add_argument("--gcp-database-user", default="postgres") + parser.add_argument("--gcp-project", default="teleo-501523") + parser.add_argument("--gcp-secret", default="gcp-teleo-pgvector-standby-postgres-password") + parser.add_argument("--gcp-service", default="leoclean-gcp-prod-parallel.service") + args = parser.parse_args() + if not (1 <= args.connect_timeout <= 60): + parser.error("--connect-timeout must be between 1 and 60 seconds") + if not (10 <= args.probe_timeout <= 900): + parser.error("--probe-timeout must be between 10 and 900 seconds") + if not args.manifest.is_file(): + parser.error("--manifest must be an existing file") + return args + + +def main() -> int: + args = parse_args() + try: + receipt, exit_code = run_live(args) + except DriftDetectorError as exc: + receipt = { + "schema": RECEIPT_SCHEMA, + "generated_at_utc": utc_now(), + "mode": "live_readonly_no_fallback", + "status": "invalid", + "error": str(exc), + "fallback_used": False, + "historical_receipt_substituted": False, + "database_mutated": False, + "routing_mutated": False, + "secrets_logged": False, + } + exit_code = 3 + write_private_json(args.output, receipt) + print_private_receipt_summary(args.output, receipt) + print( + json.dumps( + { + "status": receipt["status"], + "output": str(args.output), + "fallback_used": receipt.get("fallback_used", False), + "database_mutated": receipt.get("database_mutated", False), + "routing_mutated": receipt.get("routing_mutated", False), + }, + sort_keys=True, + ) + ) + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_detect_vps_gcp_source_drift.py b/tests/test_detect_vps_gcp_source_drift.py new file mode 100644 index 0000000..5fe4fcd --- /dev/null +++ b/tests/test_detect_vps_gcp_source_drift.py @@ -0,0 +1,229 @@ +import argparse +import copy +import hashlib +import json +from pathlib import Path + +import pytest + +from ops import detect_vps_gcp_source_drift as detector + + +def manifest(*, row_count: int = 3, rowset_md5: str = "1" * 32) -> dict: + singleton = {kind: {"kind": kind, "items": []} for kind in detector.SINGLETON_KINDS if kind != "identity"} + singleton["identity"] = { + "kind": "identity", + "database": "teleo", + "transaction_read_only": "on", + "server_version_num": 160000, + "server_address": "10.0.0.1", + "server_port": 5432, + "ssl": True, + "ssl_version": "TLSv1.3", + } + return { + "singleton": singleton, + "tables": { + "public.claims": { + "kind": "table", + "schema": "public", + "table": "claims", + "row_count": row_count, + "rowset_md5": rowset_md5, + } + }, + "performance": {}, + } + + +def observation(label: str, value: dict | None = None) -> dict: + selected = copy.deepcopy(value or manifest()) + return { + "schema": detector.OBSERVATION_SCHEMA, + "label": label, + "status": "observed_current", + "observed_at_utc": "2026-07-15T20:00:00Z", + "route": { + "hostname": f"{label}-host", + "service": f"{label}.service", + "runtime_commit": "a" * 40, + "expected_database": "teleo", + }, + "database_identity": selected["singleton"]["identity"], + "fingerprint": detector.manifest_fingerprint(selected), + "manifest": selected, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + + +def test_identical_current_observations_pass() -> None: + receipt = detector.evaluate_observations( + observation("vps"), + observation("gcp"), + generated_at_utc="2026-07-15T20:01:00Z", + ) + + assert receipt["status"] == "identical" + assert receipt["comparison"]["state"] == "identical" + assert receipt["fallback_used"] is False + assert receipt["historical_receipt_substituted"] is False + assert receipt["database_mutated"] is False + assert "manifest" not in receipt["observations"]["vps"] + + +def test_content_drift_fails_and_names_exact_table_mismatch() -> None: + gcp_manifest = manifest(row_count=4, rowset_md5="2" * 32) + receipt = detector.evaluate_observations(observation("vps"), observation("gcp", gcp_manifest)) + + assert receipt["status"] == "drift" + assert receipt["comparison"]["content_identical"] is False + assert "table_content_mismatches=1" in receipt["comparison"]["problems"] + assert receipt["comparison"]["details"]["table_mismatches"] == [ + { + "table": "public.claims", + "mismatch": { + "row_count": {"source": 3, "target": 4}, + "rowset_md5": {"source": "1" * 32, "target": "2" * 32}, + }, + } + ] + + +def test_unreachable_gcp_is_incomplete_without_vps_substitution() -> None: + gcp = { + "schema": detector.OBSERVATION_SCHEMA, + "label": "gcp", + "status": "unreachable", + "failure_class": "probe_timeout", + "error": "gcp probe exceeded 30s", + "manifest": None, + "route": None, + } + + receipt = detector.evaluate_observations(observation("vps"), gcp) + + assert receipt["status"] == "incomplete" + assert receipt["unavailable_targets"] == ["gcp"] + assert receipt["comparison"] is None + assert receipt["observations"]["gcp"]["route"] is None + assert receipt["observations"]["gcp"] != receipt["observations"]["vps"] + assert receipt["fallback_used"] is False + assert receipt["historical_receipt_substituted"] is False + + +def test_probe_timeout_preserves_exact_expected_target(monkeypatch: pytest.MonkeyPatch) -> None: + expected = { + "transport": "teleo-gcp-staging", + "database_endpoint": "10.61.0.3:5432", + "database": "teleo_canonical", + } + + def timeout(*args, **kwargs): + raise detector.subprocess.TimeoutExpired(args[0], kwargs["timeout"]) + + monkeypatch.setattr(detector.subprocess, "run", timeout) + result = detector.collect_endpoint( + label="gcp", + command=["ssh", "teleo-gcp-staging", "--", "true"], + manifest_sql=b"select 1", + timeout=30, + expected_target=expected, + ) + + assert result["status"] == "unreachable" + assert result["failure_class"] == "probe_timeout" + assert result["expected_target"] == expected + assert result["route"] is None + + +def test_parser_rejects_non_readonly_or_wrong_database(monkeypatch: pytest.MonkeyPatch) -> None: + selected = manifest() + lines = [json.dumps(row) for row in selected["singleton"].values()] + lines.extend(json.dumps(row) for row in selected["tables"].values()) + route = { + "runtime_commit": "a" * 40, + "expected_database": "different", + } + stdout = (detector.ROUTE_PREFIX + json.dumps(route) + "\n" + "\n".join(lines)).encode() + + monkeypatch.setattr( + detector.subprocess, + "run", + lambda *args, **kwargs: detector.subprocess.CompletedProcess(args[0], 0, stdout, b""), + ) + result = detector.collect_endpoint( + label="gcp", + command=["ssh", "target", "--", "true"], + manifest_sql=b"select 1", + timeout=30, + expected_target={"database_endpoint": "10.61.0.3:5432", "database": "different"}, + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "invalid_probe_receipt" + assert "does not match explicit target" in result["error"] + + +def test_live_runs_both_independent_collectors_and_returns_incomplete(tmp_path: Path) -> None: + manifest_path = tmp_path / "manifest.sql" + manifest_bytes = b"reviewed manifest" + manifest_path.write_bytes(manifest_bytes) + key = tmp_path / "key" + key.write_text("not-a-real-key") + called = [] + + def fake_collector(**kwargs): + called.append(kwargs["label"]) + if kwargs["label"] == "vps": + return observation("vps") + return { + "schema": detector.OBSERVATION_SCHEMA, + "label": "gcp", + "status": "unreachable", + "failure_class": "transport_or_remote_probe_failed", + "error": "target unavailable", + "manifest": None, + "route": None, + } + + monkeypatch = pytest.MonkeyPatch() + monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest()) + args = argparse.Namespace( + output=tmp_path / "output.json", + manifest=manifest_path, + connect_timeout=3, + probe_timeout=30, + vps_ssh_target="root@127.0.0.1", + vps_ssh_key=key, + vps_container="teleo-pg", + vps_database="teleo", + vps_database_user="postgres", + vps_service="leoclean-gateway.service", + gcp_ssh_target="teleo-gcp-staging", + gcp_host="10.61.0.3", + gcp_database="teleo_restore", + gcp_database_user="postgres", + gcp_project="teleo-501523", + gcp_secret="gcp-teleo-pgvector-standby-postgres-password", + gcp_service="leoclean-gcp-prod-parallel.service", + ) + try: + receipt, exit_code = detector.run_live(args, collector=fake_collector) + finally: + monkeypatch.undo() + + assert sorted(called) == ["gcp", "vps"] + assert receipt["status"] == "incomplete" + assert receipt["manifest_sql_sha256"] == hashlib.sha256(manifest_bytes).hexdigest() + assert exit_code == 2 + + +def test_sanitizer_redacts_connection_secrets() -> None: + error = detector.sanitize_error( + "fatal password=hunter2 postgresql://user:pw@db.internal/teleo", + target="db.internal", + ) + + assert "hunter2" not in error + assert "user:pw" not in error + assert "[REDACTED]" in error