#!/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())