diff --git a/ops/detect_vps_gcp_source_drift.py b/ops/detect_vps_gcp_source_drift.py new file mode 100644 index 0000000..efceef7 --- /dev/null +++ b/ops/detect_vps_gcp_source_drift.py @@ -0,0 +1,1040 @@ +#!/usr/bin/env python3 +"""Fail-closed, read-only VPS/Cloud SQL source fingerprint comparison.""" + +from __future__ import annotations + +import argparse +import hashlib +import ipaddress +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" +REMOTE_ROUTE_SCHEMA = "livingip.remoteSourceRoute.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") +ROWSET_MD5_RE = re.compile(r"[0-9a-f]{32}\Z") +ROUTE_PREFIX = "LEO_SOURCE_ROUTE\t" +EXIT_CODE_BY_STATUS = {"identical": 0, "drift": 1, "incomplete": 2, "invalid": 3} +SSH_TRANSPORT_EXIT_CODE = 255 +PSQL_CONNECTION_EXIT_CODE = 2 +INVALID_PROBE_EXIT_CODE = 3 + + +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 same_network_address(left: Any, right: Any) -> bool: + if left is None or right is None: + return left is right + try: + return ipaddress.ip_address(str(left)) == ipaddress.ip_address(str(right)) + except ValueError: + return str(left) == str(right) + + +def classify_nonzero_probe_result( + returncode: int, + route: dict[str, Any] | None, +) -> tuple[str, str]: + if returncode == SSH_TRANSPORT_EXIT_CODE: + return "unreachable", "ssh_transport_failed" + if route is None: + return "invalid", "remote_precondition_failed_before_route" + if returncode == PSQL_CONNECTION_EXIT_CODE: + return "unreachable", "database_connection_unavailable" + if returncode == INVALID_PROBE_EXIT_CODE: + return "invalid", "manifest_or_probe_script_failed" + return "invalid", "remote_probe_failed_after_route" + + +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") + manifest = {"singleton": singleton, "tables": tables, "performance": performance} + validate_manifest_contract(manifest, label) + return manifest + + +def validate_manifest_contract(manifest: dict[str, Any], label: str) -> None: + singleton = manifest.get("singleton") + tables = manifest.get("tables") + performance = manifest.get("performance") + if not all(isinstance(item, dict) for item in (singleton, tables, performance)): + raise DriftDetectorError(f"{label} manifest sections are invalid") + if set(singleton) != SINGLETON_KINDS: + raise DriftDetectorError(f"{label} manifest singleton set is invalid") + identity = singleton["identity"] + if not isinstance(identity, dict) or identity.get("kind") != "identity": + raise DriftDetectorError(f"{label} manifest identity row is invalid") + for field in ("database", "current_user"): + if not isinstance(identity.get(field), str) or not identity[field]: + raise DriftDetectorError(f"{label} manifest identity {field} is invalid") + server_version = identity.get("server_version_num") + if not isinstance(server_version, int) or isinstance(server_version, bool) or server_version <= 0: + raise DriftDetectorError(f"{label} manifest server version is invalid") + if identity.get("transaction_read_only") not in {"on", "off"}: + raise DriftDetectorError(f"{label} manifest transaction mode is invalid") + address = identity.get("server_address") + port = identity.get("server_port") + if address is None: + if port is not None: + raise DriftDetectorError(f"{label} manifest database socket identity is invalid") + elif ( + not isinstance(address, str) + or not address + or not isinstance(port, int) + or isinstance(port, bool) + or not 1 <= port <= 65535 + ): + raise DriftDetectorError(f"{label} manifest database socket identity is invalid") + if not isinstance(identity.get("ssl"), bool): + raise DriftDetectorError(f"{label} manifest TLS identity is invalid") + for kind in SINGLETON_KINDS - {"identity"}: + row = singleton[kind] + if not isinstance(row, dict) or row.get("kind") != kind or not isinstance(row.get("items"), list): + raise DriftDetectorError(f"{label} manifest {kind} row is invalid") + if not tables: + raise DriftDetectorError(f"{label} manifest has no table rows") + for key, row in tables.items(): + if not isinstance(row, dict) or row.get("kind") != "table": + raise DriftDetectorError(f"{label} manifest table row is invalid") + schema = row.get("schema") + table = row.get("table") + row_count = row.get("row_count") + if not isinstance(schema, str) or not schema or not isinstance(table, str) or not table: + raise DriftDetectorError(f"{label} manifest table identity is invalid") + if key != f"{schema}.{table}": + raise DriftDetectorError(f"{label} manifest table key is detached from its row") + if not isinstance(row_count, int) or isinstance(row_count, bool) or row_count < 0: + raise DriftDetectorError(f"{label} manifest table row count is invalid") + if not isinstance(row.get("rowset_md5"), str) or not ROWSET_MD5_RE.fullmatch(row["rowset_md5"]): + raise DriftDetectorError(f"{label} manifest table row hash is invalid") + for key, row in performance.items(): + if not isinstance(row, dict) or row.get("kind") != "performance" or row.get("query") != key: + raise DriftDetectorError(f"{label} manifest performance row is invalid") + elapsed_ms = row.get("elapsed_ms") + observed_rows = row.get("observed_rows") + if ( + not isinstance(elapsed_ms, int | float) + or isinstance(elapsed_ms, bool) + or elapsed_ms < 0 + or not isinstance(observed_rows, int) + or isinstance(observed_rows, bool) + or observed_rows < 0 + ): + raise DriftDetectorError(f"{label} manifest performance metrics are invalid") + + +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), + } + + +def validate_route_binding( + label: str, + route: dict[str, Any], + expected_target: dict[str, Any], +) -> None: + if route.get("schema") != REMOTE_ROUTE_SCHEMA: + raise DriftDetectorError(f"{label} route schema is invalid") + bindings = { + "route_kind": "route_kind", + "transport": "transport", + "service": "runtime_service", + "expected_database": "database", + "database_endpoint": "database_endpoint", + "database_user": "database_user", + } + for route_key, target_key in bindings.items(): + expected = expected_target.get(target_key) + if expected is None or route.get(route_key) != expected: + raise DriftDetectorError(f"{label} observed route {route_key} does not match the explicit target") + endpoint_identity = route.get("database_endpoint_identity") + if expected_target.get("route_kind") == "vps_docker": + if not isinstance(endpoint_identity, dict) or endpoint_identity.get("kind") != "docker_container": + raise DriftDetectorError(f"{label} observed Docker endpoint identity is missing") + container_id = endpoint_identity.get("container_id") + container_name = endpoint_identity.get("container_name") + selector = str(expected_target["database_endpoint"]).removeprefix("docker:") + if ( + not isinstance(container_id, str) + or not re.fullmatch(r"[0-9a-f]{64}", container_id) + or not isinstance(container_name, str) + or selector not in {container_id, container_id[:12], container_name} + or endpoint_identity.get("running") is not True + ): + raise DriftDetectorError(f"{label} observed Docker endpoint identity is invalid") + elif endpoint_identity is not None: + raise DriftDetectorError(f"{label} route contains an unexpected endpoint identity") + hostname = route.get("hostname") + if not isinstance(hostname, str) or not SAFE_NAME_RE.fullmatch(hostname): + raise DriftDetectorError(f"{label} observed route hostname is invalid") + process_cwd = route.get("process_cwd") + if not isinstance(process_cwd, str) or not Path(process_cwd).is_absolute(): + raise DriftDetectorError(f"{label} observed route process cwd is invalid") + service_state = route.get("service_state") + if not isinstance(service_state, dict): + raise DriftDetectorError(f"{label} observed route service state is missing") + try: + main_pid = int(service_state.get("MainPID") or 0) + except (TypeError, ValueError) as exc: + raise DriftDetectorError(f"{label} observed route MainPID is invalid") from exc + if service_state.get("ActiveState") != "active" or service_state.get("SubState") != "running" or main_pid <= 0: + raise DriftDetectorError(f"{label} observed route service is not active/running") + runtime_commit = route.get("runtime_commit") + if not isinstance(runtime_commit, str) or not COMMIT_RE.fullmatch(runtime_commit): + raise DriftDetectorError(f"{label} runtime commit identity is missing or ambiguous") + candidates = route.get("commit_candidates") + if route.get("commit_identity_ambiguous") is not False or not isinstance(candidates, dict): + raise DriftDetectorError(f"{label} runtime commit identity is missing or ambiguous") + candidate_values = set(candidates.values()) + if candidate_values != {runtime_commit}: + raise DriftDetectorError(f"{label} runtime commit candidates do not bind the observed revision") + + +def validate_endpoint_binding( + label: str, + route: dict[str, Any], + identity: dict[str, Any], + expected_target: dict[str, Any], +) -> dict[str, Any]: + validate_route_binding(label, route, expected_target) + if identity.get("transaction_read_only") != "on": + raise DriftDetectorError(f"{label} database transaction was not read-only") + if identity.get("database") != expected_target.get("database"): + raise DriftDetectorError(f"{label} connected database does not match the explicit target") + if identity.get("current_user") != expected_target.get("database_user"): + raise DriftDetectorError(f"{label} connected database user does not match the explicit target") + + route_kind = expected_target["route_kind"] + observed_address = identity.get("server_address") + observed_port = identity.get("server_port") + expected_address = expected_target.get("database_address") + expected_port = expected_target.get("database_port") + if route_kind == "vps_docker": + if observed_address is not None or observed_port is not None: + raise DriftDetectorError(f"{label} Docker probe did not use the expected local socket endpoint") + elif route_kind == "gcp_private_cloudsql": + if not same_network_address(observed_address, expected_address) or observed_port != expected_port: + raise DriftDetectorError(f"{label} observed database address does not match the explicit endpoint") + if identity.get("ssl") is not True: + raise DriftDetectorError(f"{label} Cloud SQL probe did not use TLS") + else: + raise DriftDetectorError(f"{label} route kind is unsupported") + + return { + "route_kind": route_kind, + "transport": route["transport"], + "hostname": route["hostname"], + "runtime_service": route["service"], + "database_endpoint": route["database_endpoint"], + "database": identity["database"], + "database_user": identity["current_user"], + "server_address": observed_address, + "server_port": observed_port, + "container_id": (route["database_endpoint_identity"]["container_id"] if route_kind == "vps_docker" else None), + } + + +REMOTE_ROUTE_INVENTORY = r"""import json +import os +import re +import subprocess +import sys +from pathlib import Path + +service, route_kind, transport, expected_db, database_endpoint, database_user, endpoint_runtime_identity = sys.argv[1:8] +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())) +database_endpoint_identity = None +if route_kind == "vps_docker": + if not database_endpoint.startswith("docker:"): + raise SystemExit("VPS database endpoint is not a Docker selector") + container_selector = database_endpoint.removeprefix("docker:") + records = json.loads(subprocess.check_output( + ["docker", "inspect", container_selector], + text=True, + stderr=subprocess.DEVNULL, + )) + if not isinstance(records, list) or len(records) != 1: + raise SystemExit("VPS database endpoint did not resolve to exactly one container") + record = records[0] + container_id = record.get("Id") + container_name = str(record.get("Name") or "").removeprefix("/") + container_running = (record.get("State") or {}).get("Running") is True + if not isinstance(container_id, str) or not re.fullmatch(r"[0-9a-f]{64}", container_id): + raise SystemExit("VPS database container identity is invalid") + if ( + container_selector not in {container_id, container_id[:12], container_name} + or endpoint_runtime_identity != container_id + or not container_running + ): + raise SystemExit("VPS database container identity does not match the requested running endpoint") + database_endpoint_identity = { + "kind": "docker_container", + "container_id": container_id, + "container_name": container_name, + "running": container_running, + } +elif endpoint_runtime_identity != "not_applicable": + raise SystemExit("unexpected database endpoint runtime identity") +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, + "database_endpoint": database_endpoint, + "database_endpoint_identity": database_endpoint_identity, + "database_user": database_user, + "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: + database_endpoint = f"docker:{container}" if route_kind == "vps_docker" else f"{gcp_host}:5432" + inventory_parts = [ + "python3", + "-c", + shlex.quote(REMOTE_ROUTE_INVENTORY), + shlex.quote(service), + shlex.quote(route_kind), + shlex.quote(transport), + shlex.quote(database), + shlex.quote(database_endpoint), + shlex.quote(database_user), + ] + if route_kind == "vps_docker": + assert container is not None + container_id = f"container_id=\"$(docker inspect --format '{{{{.Id}}}}' {shlex.quote(container)})\"" + validate_container_id = ( + '[[ "$container_id" =~ ^[0-9a-f]{64}$ ]] ' + "|| { printf 'Docker container identity is invalid\\n' >&2; exit 3; }" + ) + inventory = " ".join([*inventory_parts, '"$container_id"']) + query = " ".join( + [ + "exec docker exec", + "-e", + shlex.quote("PGOPTIONS=-c default_transaction_read_only=on"), + "-i", + '"$container_id"', + "psql -X -Atq -v ON_ERROR_STOP=1", + "-U", + shlex.quote(database_user), + "-d", + shlex.quote(database), + ] + ) + return f"set -euo pipefail\n{container_id}\n{validate_container_id}\n{inventory}\n{query}\n" + assert gcp_host is not None and gcp_project is not None and gcp_secret is not None + inventory = " ".join([*inventory_parts, shlex.quote("not_applicable")]) + 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 3; }", + '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) + if target: + line = line.replace(target, "[TARGET]") + return line[: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() + + def invalid_observation( + error: str, + route: dict[str, Any] | None, + *, + failure_class: str = "invalid_probe_receipt", + exit_code: int | None = None, + ) -> dict[str, Any]: + observation = { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "invalid", + "observed_at_utc": started, + "failure_class": failure_class, + "error": error, + "expected_target": expected_target, + "manifest": None, + "route": route, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + if exit_code is not None: + observation["exit_code"] = exit_code + return observation + + def unreachable_observation( + error: str, + route: dict[str, Any] | None, + *, + failure_class: str, + exit_code: int | None = None, + ) -> dict[str, Any]: + observation = { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "unreachable", + "observed_at_utc": started, + "failure_class": failure_class, + "error": error, + "expected_target": expected_target, + "manifest": None, + "route": route, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + if exit_code is not None: + observation["exit_code"] = exit_code + return observation + + try: + completed = subprocess.run( + command, + input=manifest_sql, + capture_output=True, + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired: + return unreachable_observation( + f"{label} probe exceeded {timeout}s", + None, + failure_class="probe_timeout", + ) + stdout = completed.stdout.decode(errors="replace") + stderr = completed.stderr.decode(errors="replace") + lines = stdout.splitlines() + route_lines = [line for line in lines if line.startswith(ROUTE_PREFIX)] + route: dict[str, Any] | None = None + route_error: str | None = None + if len(route_lines) != 1: + route_error = f"{label} probe emitted {len(route_lines)} route observations" + else: + try: + candidate = json.loads(route_lines[0].removeprefix(ROUTE_PREFIX)) + except json.JSONDecodeError: + route_error = f"{label} route observation is not valid JSON" + else: + if isinstance(candidate, dict): + route = candidate + else: + route_error = f"{label} route observation is not an object" + if route is not None: + try: + validate_route_binding(label, route, expected_target) + except (DriftDetectorError, KeyError, TypeError, ValueError) as exc: + route_error = str(exc) + if completed.returncode == 0 and route_error is not None: + return invalid_observation(route_error, route) + if completed.returncode != 0: + if route_error is not None and route_lines: + return invalid_observation( + route_error, + route, + failure_class="malformed_probe_output", + exit_code=completed.returncode, + ) + try: + target = command[command.index("--") - 1] + except (ValueError, IndexError): + target = "" + error = sanitize_error(stderr, target=target) + status, failure_class = classify_nonzero_probe_result(completed.returncode, route) + if status == "unreachable": + return unreachable_observation( + error, + route, + failure_class=failure_class, + exit_code=completed.returncode, + ) + return invalid_observation( + error, + route, + failure_class=failure_class, + exit_code=completed.returncode, + ) + if route is None: + return invalid_observation(route_error or f"{label} route observation is missing", None) + try: + manifest = parse_manifest_lines([line for line in lines if line not in route_lines], label) + identity = manifest["singleton"]["identity"] + fingerprint = manifest_fingerprint(manifest) + endpoint_identity = validate_endpoint_binding(label, route, identity, expected_target) + except (DriftDetectorError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + return invalid_observation(str(exc), route) + return { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "observed_current", + "observed_at_utc": utc_now(), + "expected_target": expected_target, + "route": route, + "database_identity": identity, + "endpoint_identity": endpoint_identity, + "endpoint_identity_sha256": canonical_hash(endpoint_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 invalid_public_observation(label: str, observation: Any, error: str) -> dict[str, Any]: + value = observation if isinstance(observation, dict) else {} + return { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "invalid", + "observed_at_utc": value.get("observed_at_utc"), + "failure_class": "invalid_observation_payload", + "error": error, + "expected_target": value.get("expected_target") if isinstance(value.get("expected_target"), dict) else None, + "route": value.get("route") if isinstance(value.get("route"), dict) else None, + } + + +def observation_validation_error( + label: str, + observation: Any, + requested_target: dict[str, Any] | None = None, +) -> str | None: + if not isinstance(observation, dict): + return f"{label} observation is not an object" + if observation.get("schema") != OBSERVATION_SCHEMA: + return f"{label} observation schema is invalid" + if observation.get("label") != label: + return f"{label} observation label does not match its collector slot" + status = observation.get("status") + if status not in {"observed_current", "unreachable", "invalid"}: + return f"{label} observation status is invalid" + expected_target = observation.get("expected_target") + if not isinstance(expected_target, dict): + return f"{label} observation is missing its explicit target binding" + if requested_target is not None and expected_target != requested_target: + return f"{label} observation target does not match the requested collector target" + if status == "invalid": + return None + if status == "unreachable": + if observation.get("manifest") is not None: + return f"{label} unreachable observation contains a manifest" + if not isinstance(observation.get("failure_class"), str) or not isinstance(observation.get("error"), str): + return f"{label} unreachable observation is missing failure details" + route = observation.get("route") + if route is not None: + if not isinstance(route, dict): + return f"{label} unreachable observation route is not an object" + try: + validate_route_binding(label, route, expected_target) + except (DriftDetectorError, KeyError, TypeError, ValueError) as exc: + return str(exc) + return None + + try: + route = observation["route"] + manifest = observation["manifest"] + identity = observation["database_identity"] + fingerprint = observation["fingerprint"] + endpoint_identity = observation["endpoint_identity"] + if not all(isinstance(item, dict) for item in (route, manifest, identity, fingerprint, endpoint_identity)): + raise DriftDetectorError(f"{label} current observation has a malformed object field") + manifest_identity = manifest["singleton"]["identity"] + if identity != manifest_identity: + raise DriftDetectorError(f"{label} database identity is detached from its manifest") + validate_manifest_contract(manifest, label) + computed_fingerprint = manifest_fingerprint(manifest) + if fingerprint != computed_fingerprint: + raise DriftDetectorError(f"{label} manifest fingerprint does not match its payload") + computed_endpoint_identity = validate_endpoint_binding(label, route, identity, expected_target) + if endpoint_identity != computed_endpoint_identity: + raise DriftDetectorError(f"{label} endpoint identity does not match its route and database") + if observation.get("endpoint_identity_sha256") != canonical_hash(endpoint_identity): + raise DriftDetectorError(f"{label} endpoint identity hash does not match its payload") + except (DriftDetectorError, KeyError, TypeError, ValueError, OverflowError) as exc: + return str(exc) + return None + + +def endpoint_collision_reason(vps: dict[str, Any], gcp: dict[str, Any]) -> str | None: + vps_identity = vps["endpoint_identity"] + gcp_identity = gcp["endpoint_identity"] + if vps["endpoint_identity_sha256"] == gcp["endpoint_identity_sha256"]: + return "VPS and GCP observations resolve to the same endpoint identity" + if vps_identity["hostname"] == gcp_identity["hostname"]: + return "VPS and GCP observations resolve to the same route hostname" + if ( + vps_identity["server_address"] is not None + and same_network_address(vps_identity["server_address"], gcp_identity["server_address"]) + and vps_identity["server_port"] == gcp_identity["server_port"] + and vps_identity["database"] == gcp_identity["database"] + ): + return "VPS and GCP observations resolve to the same database socket" + return None + + +def evaluate_observations( + vps: Any, + gcp: Any, + *, + generated_at_utc: str | None = None, + requested_targets: dict[str, dict[str, Any]] | None = None, +) -> dict[str, Any]: + observations = {"vps": vps, "gcp": gcp} + public_observations: dict[str, Any] = {} + unavailable: list[str] = [] + invalid: list[str] = [] + validation_errors: dict[str, str] = {} + for label, observation in observations.items(): + requested_target = requested_targets.get(label) if requested_targets is not None else None + error = observation_validation_error(label, observation, requested_target) + if error is not None: + invalid.append(label) + validation_errors[label] = error + public_observations[label] = invalid_public_observation(label, observation, error) + else: + public_observations[label] = public_observation(observation) + if observation["status"] == "invalid": + invalid.append(label) + elif observation["status"] == "unreachable": + unavailable.append(label) + initial_status = "invalid" if invalid else "incomplete" if unavailable else "pending_comparison" + receipt: dict[str, Any] = { + "schema": RECEIPT_SCHEMA, + "generated_at_utc": generated_at_utc or utc_now(), + "mode": "live_readonly_no_fallback", + "status": initial_status, + "comparison": None, + "observations": public_observations, + "unavailable_targets": unavailable, + "invalid_targets": invalid, + "validation_errors": validation_errors, + "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 invalid: + receipt["decision"] = "No source-of-truth decision: at least one endpoint observation is invalid." + return receipt + if unavailable: + receipt["decision"] = "No source-of-truth decision: every target must be observed live." + return receipt + collision = endpoint_collision_reason(vps, gcp) + if collision is not None: + receipt["status"] = "invalid" + receipt["invalid_targets"] = ["vps", "gcp"] + receipt["validation_errors"] = {"endpoint_pair": collision} + receipt["decision"] = "No source-of-truth decision: endpoint independence was not proven." + return receipt + try: + problems, details = compare_manifests( + vps["manifest"], + gcp["manifest"], + max_target_query_ms=10_000.0, + max_target_source_ratio=1_000.0, + ) + except (KeyError, TypeError, ValueError, OverflowError) as exc: + receipt["status"] = "invalid" + receipt["invalid_targets"] = ["vps", "gcp"] + receipt["validation_errors"] = {"manifest_comparison": str(exc)} + receipt["decision"] = "No source-of-truth decision: manifest comparison input is invalid." + return receipt + content_identical = vps["fingerprint"]["table_fingerprint_sha256"] == gcp["fingerprint"]["table_fingerprint_sha256"] + structure_identical = ( + vps["fingerprint"]["structure_fingerprint_sha256"] == gcp["fingerprint"]["structure_fingerprint_sha256"] + ) + runtime_revision_identical = vps["route"]["runtime_commit"] == gcp["route"]["runtime_commit"] + if not runtime_revision_identical: + problems.append("runtime_commit_mismatch") + state = ( + "identical" + if not problems and content_identical and structure_identical and runtime_revision_identical + else "drift" + ) + receipt["status"] = state + receipt["comparison"] = { + "state": state, + "content_identical": content_identical, + "structure_identical": structure_identical, + "runtime_revision_identical": runtime_revision_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 = { + "route_kind": "vps_docker", + "transport": vps_target, + "runtime_service": args.vps_service, + "database_endpoint": f"docker:{args.vps_container}", + "database_address": None, + "database_port": None, + "database": args.vps_database, + "database_user": args.vps_database_user, + "identity_source": "explicit_cli_target", + } + gcp_expected_target = { + "route_kind": "gcp_private_cloudsql", + "transport": gcp_target, + "runtime_service": args.gcp_service, + "database_endpoint": f"{args.gcp_host}:5432", + "database_address": args.gcp_host, + "database_port": 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, + requested_targets={"vps": vps_expected_target, "gcp": gcp_expected_target}, + ) + receipt["manifest_sql_sha256"] = manifest_sha256 + return receipt, EXIT_CODE_BY_STATUS[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 invalidate_current_receipt(path: Path) -> None: + """Remove the previous canonical receipt before any new probe can start.""" + try: + path.unlink() + except FileNotFoundError: + pass + + +def main() -> int: + args = parse_args() + invalidate_current_receipt(args.output) + 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_name": args.output.name, + "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..6bfac45 --- /dev/null +++ b/tests/test_detect_vps_gcp_source_drift.py @@ -0,0 +1,800 @@ +import argparse +import copy +import hashlib +import json +from pathlib import Path + +import pytest + +from ops import detect_vps_gcp_source_drift as detector +from ops import private_receipt_io + + +def endpoint_target(label: str) -> dict: + if label == "vps": + return { + "route_kind": "vps_docker", + "transport": "root@192.0.2.10", + "runtime_service": "leoclean-gateway.service", + "database_endpoint": "docker:teleo-pg", + "database_address": None, + "database_port": None, + "database": "teleo", + "database_user": "postgres", + "identity_source": "explicit_cli_target", + } + return { + "route_kind": "gcp_private_cloudsql", + "transport": "teleo-gcp-staging", + "runtime_service": "leoclean-gcp-prod-parallel.service", + "database_endpoint": "10.61.0.3:5432", + "database_address": "10.61.0.3", + "database_port": 5432, + "database": "teleo", + "database_user": "postgres", + "project": "teleo-501523", + "credential_source": "secret-manager:gcp-teleo-pgvector-standby-postgres-password", + "identity_source": "explicit_cli_target", + } + + +def manifest(*, label: str, row_count: int = 3, rowset_md5: str = "1" * 32) -> dict: + singleton = {kind: {"kind": kind, "items": []} for kind in detector.SINGLETON_KINDS if kind != "identity"} + is_gcp = label == "gcp" + singleton["identity"] = { + "kind": "identity", + "database": "teleo", + "current_user": "postgres", + "transaction_read_only": "on", + "server_version_num": 160000, + "server_address": "10.61.0.3" if is_gcp else None, + "server_port": 5432 if is_gcp else None, + "ssl": is_gcp, + "ssl_version": "TLSv1.3" if is_gcp else None, + } + return { + "singleton": singleton, + "tables": { + "public.claims": { + "kind": "table", + "schema": "public", + "table": "claims", + "row_count": row_count, + "rowset_md5": rowset_md5, + } + }, + "performance": {}, + } + + +def route_observation(label: str, *, runtime_commit: str = "a" * 40) -> dict: + target = endpoint_target(label) + process_cwd = f"/srv/{label}" + database_endpoint_identity = None + if label == "vps": + database_endpoint_identity = { + "kind": "docker_container", + "container_id": "c" * 64, + "container_name": "teleo-pg", + "running": True, + } + return { + "schema": detector.REMOTE_ROUTE_SCHEMA, + "route_kind": target["route_kind"], + "transport": target["transport"], + "hostname": f"{label}-host", + "service": target["runtime_service"], + "service_state": { + "ActiveState": "active", + "SubState": "running", + "MainPID": "123", + }, + "process_cwd": process_cwd, + "expected_database": target["database"], + "database_endpoint": target["database_endpoint"], + "database_endpoint_identity": database_endpoint_identity, + "database_user": target["database_user"], + "runtime_commit": runtime_commit, + "commit_candidates": {f"git:{process_cwd}": runtime_commit}, + "commit_identity_ambiguous": False, + "secret_values_emitted": False, + } + + +def observation( + label: str, + value: dict | None = None, + *, + runtime_commit: str = "a" * 40, +) -> dict: + selected = copy.deepcopy(value or manifest(label=label)) + target = endpoint_target(label) + route = route_observation(label, runtime_commit=runtime_commit) + endpoint_identity = detector.validate_endpoint_binding( + label, + route, + selected["singleton"]["identity"], + target, + ) + return { + "schema": detector.OBSERVATION_SCHEMA, + "label": label, + "status": "observed_current", + "observed_at_utc": "2026-07-15T20:00:00Z", + "expected_target": target, + "route": route, + "database_identity": selected["singleton"]["identity"], + "endpoint_identity": endpoint_identity, + "endpoint_identity_sha256": detector.canonical_hash(endpoint_identity), + "fingerprint": detector.manifest_fingerprint(selected), + "manifest": selected, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + + +def live_args(tmp_path: Path, manifest_bytes: bytes) -> argparse.Namespace: + manifest_path = tmp_path / "manifest.sql" + manifest_path.write_bytes(manifest_bytes) + key = tmp_path / "key" + key.write_text("not-a-real-key") + return argparse.Namespace( + output=tmp_path / "output.json", + manifest=manifest_path, + connect_timeout=3, + probe_timeout=30, + vps_ssh_target="root@192.0.2.10", + 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", + gcp_database_user="postgres", + gcp_project="teleo-501523", + gcp_secret="gcp-teleo-pgvector-standby-postgres-password", + gcp_service="leoclean-gcp-prod-parallel.service", + ) + + +def collect_completed_probe( + monkeypatch: pytest.MonkeyPatch, + *, + label: str, + returncode: int, + route: dict | None, + stderr: str, +) -> dict: + expected = endpoint_target(label) + stdout = b"" if route is None else (detector.ROUTE_PREFIX + json.dumps(route) + "\n").encode() + monkeypatch.setattr( + detector.subprocess, + "run", + lambda *args, **kwargs: detector.subprocess.CompletedProcess( + args[0], + returncode, + stdout, + stderr.encode(), + ), + ) + return detector.collect_endpoint( + label=label, + command=["ssh", expected["transport"], "--", "true"], + manifest_sql=b"select 1", + timeout=30, + expected_target=expected, + ) + + +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_remote_route_inventory_source_compiles() -> None: + compile(detector.REMOTE_ROUTE_INVENTORY, "", "exec") + + +def test_vps_remote_script_executes_manifest_against_inspected_container_id() -> None: + script = detector.remote_script( + route_kind="vps_docker", + transport="root@192.0.2.10", + service="leoclean-gateway.service", + database="teleo", + container="teleo-pg", + database_user="postgres", + gcp_host=None, + gcp_project=None, + gcp_secret=None, + ) + + syntax = detector.subprocess.run(["bash", "-n"], input=script, text=True, capture_output=True, check=False) + + assert syntax.returncode == 0, syntax.stderr + assert "docker inspect --format" in script + assert ' -i "$container_id" psql ' in script + assert "exec docker exec -e PGOPTIONS=-c default_transaction_read_only=on -i teleo-pg" not in script + + +def test_gcp_remote_script_declares_empty_credentials_invalid() -> None: + script = detector.remote_script( + route_kind="gcp_private_cloudsql", + transport="teleo-gcp-staging", + service="leoclean-gcp-prod-parallel.service", + database="teleo", + container=None, + database_user="postgres", + gcp_host="10.61.0.3", + gcp_project="teleo-501523", + gcp_secret="gcp-teleo-pgvector-standby-postgres-password", + ) + + syntax = detector.subprocess.run(["bash", "-n"], input=script, text=True, capture_output=True, check=False) + + assert syntax.returncode == 0, syntax.stderr + credential_guard = next(line for line in script.splitlines() if "credential resolved empty" in line) + assert "exit 3" in credential_guard + + +def test_content_drift_fails_and_names_exact_table_mismatch() -> None: + gcp_manifest = manifest(label="gcp", 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", + "expected_target": endpoint_target("gcp"), + "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_declared_invalid_observation_takes_precedence_over_unreachable() -> None: + vps = { + "schema": detector.OBSERVATION_SCHEMA, + "label": "vps", + "status": "invalid", + "failure_class": "invalid_probe_receipt", + "error": "manifest row is malformed", + "expected_target": endpoint_target("vps"), + "manifest": None, + "route": None, + } + gcp = { + "schema": detector.OBSERVATION_SCHEMA, + "label": "gcp", + "status": "unreachable", + "failure_class": "probe_timeout", + "error": "target unavailable", + "expected_target": endpoint_target("gcp"), + "manifest": None, + "route": None, + } + + receipt = detector.evaluate_observations(vps, gcp) + + assert receipt["status"] == "invalid" + assert receipt["invalid_targets"] == ["vps"] + assert receipt["unavailable_targets"] == ["gcp"] + assert receipt["comparison"] is None + + +@pytest.mark.parametrize( + ("field", "mutated_value"), + [ + ("schema", "livingip.sourceFingerprintObservation.v0"), + ("label", "vps"), + ("status", "unknown"), + ("endpoint_identity_sha256", "0" * 64), + ], +) +def test_mutated_observation_envelope_is_invalid(field: str, mutated_value: str) -> None: + gcp = observation("gcp") + gcp[field] = mutated_value + + receipt = detector.evaluate_observations(observation("vps"), gcp) + + assert receipt["status"] == "invalid" + assert receipt["invalid_targets"] == ["gcp"] + assert "gcp" in receipt["validation_errors"] + + +@pytest.mark.parametrize( + ("route_field", "mutated_value"), + [ + ("transport", "other-gcp-host"), + ("service", "other.service"), + ("database_endpoint", "10.61.0.4:5432"), + ], +) +def test_mutated_route_binding_is_invalid(route_field: str, mutated_value: str) -> None: + gcp = observation("gcp") + gcp["route"][route_field] = mutated_value + + receipt = detector.evaluate_observations(observation("vps"), gcp) + + assert receipt["status"] == "invalid" + assert receipt["invalid_targets"] == ["gcp"] + + +def test_swapped_vps_container_identity_is_invalid() -> None: + vps = observation("vps") + vps["route"]["database_endpoint_identity"]["container_name"] = "other-postgres" + + receipt = detector.evaluate_observations(vps, observation("gcp")) + + assert receipt["status"] == "invalid" + assert "Docker endpoint identity is invalid" in receipt["validation_errors"]["vps"] + + +def test_mutated_database_address_is_invalid_even_when_manifest_and_hash_are_recomputed() -> None: + gcp = observation("gcp") + gcp["database_identity"]["server_address"] = "10.61.0.4" + gcp["fingerprint"] = detector.manifest_fingerprint(gcp["manifest"]) + gcp["endpoint_identity"]["server_address"] = "10.61.0.4" + gcp["endpoint_identity_sha256"] = detector.canonical_hash(gcp["endpoint_identity"]) + + receipt = detector.evaluate_observations(observation("vps"), gcp) + + assert receipt["status"] == "invalid" + assert "address does not match" in receipt["validation_errors"]["gcp"] + + +def test_malformed_current_manifest_is_invalid_instead_of_incomplete() -> None: + gcp = observation("gcp") + gcp["manifest"]["tables"]["public.claims"].pop("row_count") + + receipt = detector.evaluate_observations(observation("vps"), gcp) + + assert receipt["status"] == "invalid" + assert receipt["invalid_targets"] == ["gcp"] + assert receipt["unavailable_targets"] == [] + + +def test_identically_malformed_manifests_cannot_compare_as_identical() -> None: + vps = observation("vps") + gcp = observation("gcp") + for item in (vps, gcp): + item["manifest"]["singleton"]["schemas"]["items"] = {} + item["fingerprint"] = detector.manifest_fingerprint(item["manifest"]) + + receipt = detector.evaluate_observations(vps, gcp) + + assert receipt["status"] == "invalid" + assert receipt["invalid_targets"] == ["vps", "gcp"] + + +def test_coherent_same_host_endpoint_aliases_are_invalid() -> None: + vps = observation("vps") + gcp = observation("gcp") + gcp["route"]["hostname"] = vps["route"]["hostname"] + gcp["endpoint_identity"]["hostname"] = vps["endpoint_identity"]["hostname"] + gcp["endpoint_identity_sha256"] = detector.canonical_hash(gcp["endpoint_identity"]) + + receipt = detector.evaluate_observations(vps, gcp) + + assert receipt["status"] == "invalid" + assert receipt["invalid_targets"] == ["vps", "gcp"] + assert "same route hostname" in receipt["validation_errors"]["endpoint_pair"] + + +def test_runtime_revision_mismatch_is_drift() -> None: + receipt = detector.evaluate_observations( + observation("vps", runtime_commit="a" * 40), + observation("gcp", runtime_commit="b" * 40), + ) + + assert receipt["status"] == "drift" + assert receipt["comparison"]["runtime_revision_identical"] is False + assert "runtime_commit_mismatch" in receipt["comparison"]["problems"] + + +def test_collect_ssh_255_without_route_is_unreachable(monkeypatch: pytest.MonkeyPatch) -> None: + result = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=255, + route=None, + stderr="ssh: connect to host teleo-gcp-staging port 22: timed out password=hunter2", + ) + + assert result["status"] == "unreachable" + assert result["failure_class"] == "ssh_transport_failed" + assert result["exit_code"] == 255 + assert result["route"] is None + assert "teleo-gcp-staging" not in result["error"] + assert "hunter2" not in result["error"] + + +@pytest.mark.parametrize("returncode", [1, 3]) +def test_collect_remote_invalid_container_without_route_is_invalid( + monkeypatch: pytest.MonkeyPatch, + returncode: int, +) -> None: + result = collect_completed_probe( + monkeypatch, + label="vps", + returncode=returncode, + route=None, + stderr="docker: Error: No such container: teleo-pg password=hunter2 root@192.0.2.10", + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "remote_precondition_failed_before_route" + assert result["exit_code"] == returncode + assert result["route"] is None + assert "root@192.0.2.10" not in result["error"] + assert "hunter2" not in result["error"] + + +def test_collect_valid_route_with_psql_connection_failure_is_unreachable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + route = route_observation("gcp") + result = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=2, + route=route, + stderr="psql: connection unavailable token=private-token teleo-gcp-staging", + ) + + assert result["status"] == "unreachable" + assert result["failure_class"] == "database_connection_unavailable" + assert result["exit_code"] == 2 + assert result["route"] == route + assert "private-token" not in result["error"] + assert "teleo-gcp-staging" not in result["error"] + + +def test_collect_valid_route_with_manifest_script_exit_3_is_invalid( + monkeypatch: pytest.MonkeyPatch, +) -> None: + route = route_observation("gcp") + result = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=3, + route=route, + stderr="psql: manifest script failed secret=private-secret teleo-gcp-staging", + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "manifest_or_probe_script_failed" + assert result["exit_code"] == 3 + assert result["route"] == route + assert "private-secret" not in result["error"] + assert "teleo-gcp-staging" not in result["error"] + + +def test_collect_malformed_route_output_is_invalid_even_on_ssh_exit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + expected = endpoint_target("gcp") + stdout = (detector.ROUTE_PREFIX + "{not-json}\n").encode() + monkeypatch.setattr( + detector.subprocess, + "run", + lambda *args, **kwargs: detector.subprocess.CompletedProcess(args[0], 255, stdout, b"transport closed"), + ) + + result = detector.collect_endpoint( + label="gcp", + command=["ssh", expected["transport"], "--", "true"], + manifest_sql=b"select 1", + timeout=30, + expected_target=expected, + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "malformed_probe_output" + assert result["exit_code"] == 255 + + +def test_probe_timeout_preserves_exact_expected_target(monkeypatch: pytest.MonkeyPatch) -> None: + expected = endpoint_target("gcp") + expected["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(label="gcp") + lines = [json.dumps(row) for row in selected["singleton"].values()] + lines.extend(json.dumps(row) for row in selected["tables"].values()) + expected = endpoint_target("gcp") + expected["database"] = "different" + route = route_observation("gcp") + route["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=expected, + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "invalid_probe_receipt" + assert "does not match the explicit target" in result["error"] + + +def test_collector_rejects_route_that_does_not_match_requested_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + selected = manifest(label="gcp") + lines = [json.dumps(row) for row in selected["singleton"].values()] + lines.extend(json.dumps(row) for row in selected["tables"].values()) + route = route_observation("gcp") + route["transport"] = "attacker-controlled-alias" + 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", "teleo-gcp-staging", "--", "true"], + manifest_sql=b"select 1", + timeout=30, + expected_target=endpoint_target("gcp"), + ) + + assert result["status"] == "invalid" + assert "route transport does not match" in result["error"] + + +def test_live_runs_both_independent_collectors_and_returns_incomplete(tmp_path: Path) -> None: + manifest_bytes = b"reviewed manifest" + 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", + "expected_target": endpoint_target("gcp"), + "manifest": None, + "route": None, + } + + monkeypatch = pytest.MonkeyPatch() + monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest()) + args = live_args(tmp_path, manifest_bytes) + 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 + + +@pytest.mark.parametrize( + ("returncode", "include_route", "failure_class", "expected_status", "expected_exit"), + [ + (255, False, "ssh_transport_failed", "incomplete", 2), + (1, False, "remote_precondition_failed_before_route", "invalid", 3), + (3, False, "remote_precondition_failed_before_route", "invalid", 3), + (2, True, "database_connection_unavailable", "incomplete", 2), + (3, True, "manifest_or_probe_script_failed", "invalid", 3), + ], +) +def test_run_live_maps_collected_nonzero_partition( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + returncode: int, + include_route: bool, + failure_class: str, + expected_status: str, + expected_exit: int, +) -> None: + gcp = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=returncode, + route=route_observation("gcp") if include_route else None, + stderr="bounded probe failure password=hunter2 teleo-gcp-staging", + ) + manifest_bytes = b"reviewed manifest" + args = live_args(tmp_path, manifest_bytes) + monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest()) + + def fake_collector(**kwargs): + return observation("vps") if kwargs["label"] == "vps" else gcp + + receipt, exit_code = detector.run_live(args, collector=fake_collector) + + assert receipt["status"] == expected_status + assert receipt["observations"]["gcp"]["failure_class"] == failure_class + assert exit_code == expected_exit + + +def test_live_invalid_collector_payload_returns_exit_3( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manifest_bytes = b"reviewed manifest" + args = live_args(tmp_path, manifest_bytes) + monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest()) + + def fake_collector(**kwargs): + result = observation(kwargs["label"]) + if kwargs["label"] == "gcp": + result.pop("endpoint_identity_sha256") + return result + + receipt, exit_code = detector.run_live(args, collector=fake_collector) + + assert receipt["status"] == "invalid" + assert receipt["invalid_targets"] == ["gcp"] + assert exit_code == 3 + + +def test_live_rejects_collector_that_coherently_retargets_an_endpoint( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manifest_bytes = b"reviewed manifest" + args = live_args(tmp_path, manifest_bytes) + monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest()) + + def fake_collector(**kwargs): + result = observation(kwargs["label"]) + if kwargs["label"] == "gcp": + result["expected_target"]["transport"] = "other-gcp-host" + result["route"]["transport"] = "other-gcp-host" + result["endpoint_identity"]["transport"] = "other-gcp-host" + result["endpoint_identity_sha256"] = detector.canonical_hash(result["endpoint_identity"]) + return result + + receipt, exit_code = detector.run_live(args, collector=fake_collector) + + assert receipt["status"] == "invalid" + assert exit_code == 3 + assert "requested collector target" in receipt["validation_errors"]["gcp"] + + +def test_interruption_removes_stale_receipt_before_probe( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = tmp_path / "current.json" + output.write_text('{"status": "identical", "run": "stale"}\n') + args = argparse.Namespace(output=output) + monkeypatch.setattr(detector, "parse_args", lambda: args) + + def interrupt(_args): + assert not output.exists() + raise KeyboardInterrupt + + monkeypatch.setattr(detector, "run_live", interrupt) + + with pytest.raises(KeyboardInterrupt): + detector.main() + + assert not output.exists() + + +def test_interruption_during_atomic_publication_cannot_restore_stale_receipt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = tmp_path / "current.json" + output.write_text('{"status": "identical", "run": "stale"}\n') + args = argparse.Namespace(output=output) + receipt = {"schema": detector.RECEIPT_SCHEMA, "status": "identical"} + monkeypatch.setattr(detector, "parse_args", lambda: args) + monkeypatch.setattr(detector, "run_live", lambda _args: (receipt, 0)) + monkeypatch.setattr( + private_receipt_io.os, + "replace", + lambda _source, _destination: (_ for _ in ()).throw(KeyboardInterrupt()), + ) + + with pytest.raises(KeyboardInterrupt): + detector.main() + + assert not output.exists() + assert list(tmp_path.glob(".*.tmp")) == [] + + +def test_successful_atomic_publication_replaces_stale_receipt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + output = tmp_path / "private" / "current.json" + output.parent.mkdir() + output.write_text('{"status": "identical", "run": "stale"}\n') + args = argparse.Namespace(output=output) + receipt = {"schema": detector.RECEIPT_SCHEMA, "status": "invalid", "run": "current"} + monkeypatch.setattr(detector, "parse_args", lambda: args) + monkeypatch.setattr(detector, "run_live", lambda _args: (receipt, 3)) + + assert detector.main() == 3 + + assert json.loads(output.read_text()) == receipt + assert output.stat().st_mode & 0o777 == 0o600 + assert str(output.parent) not in capsys.readouterr().out + + +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