#!/usr/bin/env python3 """Restore and verify a canonical PostgreSQL dump in an isolated Docker canary.""" from __future__ import annotations import argparse import hashlib import json import os import re import signal import subprocess import tempfile import time import uuid from datetime import UTC, datetime from pathlib import Path from typing import Any try: from .verify_postgres_parity_manifest import compare_manifests, load_manifest except ImportError: # pragma: no cover - direct script execution from verify_postgres_parity_manifest import compare_manifests, load_manifest DEFAULT_IMAGE = "postgres:16-alpine" DEFAULT_MANIFEST_SQL = Path(__file__).with_name("postgres_parity_manifest.sql") DATA_DIR = "/var/lib/postgresql/data" DUMP_DESTINATION = "/tmp/teleo-canonical.dump" MANIFEST_DESTINATION = "/tmp/postgres-parity-manifest.sql" CANARY_LABEL = "com.livingip.teleo.canonical-rebuild-canary" DATABASE_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,62}\Z") CONTAINER_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,39}\Z") ALLOWED_APPLICATION_ROLES = frozenset({"kb_apply", "kb_review"}) ROLE_BOOLEAN_FIELDS = ( "can_login", "superuser", "create_db", "create_role", "inherit", "replication", "bypass_rls", ) KEY_TABLES = ( "public.claims", "public.sources", "public.claim_evidence", "public.claim_edges", "public.reasoning_tools", "kb_stage.kb_proposals", ) class TerminationRequested(RuntimeError): """Raised after SIGINT/SIGTERM so the normal cleanup path still runs.""" def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def validate_custom_dump(path: Path) -> None: if not path.is_file(): raise ValueError(f"custom dump does not exist or is not a regular file: {path}") if path.stat().st_size <= 5: raise ValueError("custom dump is empty or truncated") with path.open("rb") as handle: signature = handle.read(5) if signature != b"PGDMP": raise ValueError("dump is not PostgreSQL custom format (missing PGDMP signature)") def new_container_name(prefix: str) -> str: timestamp = datetime.now(UTC).strftime("%Y%m%d%H%M%S") return f"{prefix}-{timestamp}-{uuid.uuid4().hex[:12]}" def _run( command: list[str], *, timeout: float, ) -> subprocess.CompletedProcess[str]: return subprocess.run( command, capture_output=True, check=False, text=True, timeout=timeout, ) def _bounded_error(completed: subprocess.CompletedProcess[str]) -> str: message = (completed.stderr or completed.stdout or "no command output").strip() return message[-2000:] def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> str: if completed.returncode != 0: raise RuntimeError(f"{action} failed (exit {completed.returncode}): {_bounded_error(completed)}") return completed.stdout def _absence_result(completed: subprocess.CompletedProcess[str]) -> tuple[bool | None, str | None]: if completed.returncode == 0: return False, None message = (completed.stderr or completed.stdout or "").strip() lowered = message.lower() if "no such object" in lowered or "no such container" in lowered: return True, None return None, message[-2000:] or f"docker inspect exited {completed.returncode}" def inspect_container_absence( docker_bin: str, container: str, *, timeout: float, ) -> tuple[bool | None, str | None]: completed = _run([docker_bin, "container", "inspect", container], timeout=timeout) return _absence_result(completed) def inspect_container_isolation( docker_bin: str, container: str, *, expected_image: str, expected_label: str, timeout: float, ) -> dict[str, Any]: completed = _run([docker_bin, "container", "inspect", container], timeout=timeout) raw = _require_success(completed, "container isolation inspection") try: rows = json.loads(raw) observed = rows[0] host_config = observed["HostConfig"] config = observed["Config"] except (IndexError, KeyError, TypeError, json.JSONDecodeError) as exc: raise RuntimeError("container isolation inspection returned invalid JSON") from exc network_mode = host_config.get("NetworkMode") tmpfs = host_config.get("Tmpfs") or {} labels = config.get("Labels") or {} image = config.get("Image") running = bool((observed.get("State") or {}).get("Running")) problems = [] if network_mode != "none": problems.append(f"network_mode={network_mode!r}") if DATA_DIR not in tmpfs: problems.append("postgres_data_tmpfs_missing") if labels.get(CANARY_LABEL) != expected_label: problems.append("canary_label_mismatch") if image != expected_image: problems.append(f"image={image!r}") if not running: problems.append("container_not_running") if problems: raise RuntimeError("container isolation check failed: " + ", ".join(problems)) return { "network_mode": network_mode, "postgres_data_tmpfs": tmpfs[DATA_DIR], "image": image, "label": labels[CANARY_LABEL], "running": running, } def _psql_command(container: str, database: str, *arguments: str, docker_bin: str) -> list[str]: return [ docker_bin, "exec", container, "psql", "-X", "-U", "postgres", "-d", database, "-Atq", "-v", "ON_ERROR_STOP=1", *arguments, ] def wait_for_named_database( docker_bin: str, container: str, database: str, *, startup_timeout: float, command_timeout: float, poll_interval: float, ) -> int: deadline = time.monotonic() + startup_timeout attempts = 0 last_error = "no psql attempt completed" while True: attempts += 1 remaining = max(0.1, deadline - time.monotonic()) completed = _run( _psql_command( container, database, "-c", "select current_database();", docker_bin=docker_bin, ), timeout=min(command_timeout, remaining), ) if completed.returncode == 0 and completed.stdout.strip() == database: return attempts last_error = _bounded_error(completed) if time.monotonic() >= deadline: raise RuntimeError( f"named database did not accept a psql connection within {startup_timeout:g}s: {last_error}" ) time.sleep(min(poll_interval, max(0.0, deadline - time.monotonic()))) def _role_keyword(value: bool, enabled: str, disabled: str) -> str: return enabled if value else disabled def application_role_sql(source_manifest: dict[str, Any]) -> str: items = source_manifest["singleton"]["application_roles"].get("items", []) if not isinstance(items, list): raise ValueError("source manifest application_roles.items must be a list") statements = [] seen: set[str] = set() for item in items: if not isinstance(item, dict): raise ValueError("source manifest application role must be an object") name = item.get("name") if name not in ALLOWED_APPLICATION_ROLES: raise ValueError(f"source manifest contains non-allowlisted application role: {name!r}") if name in seen: raise ValueError(f"source manifest contains duplicate application role: {name}") seen.add(name) for field in ROLE_BOOLEAN_FIELDS: if type(item.get(field)) is not bool: raise ValueError(f"source manifest application role {name} has invalid {field}") attributes = [ _role_keyword(item["can_login"], "LOGIN", "NOLOGIN"), _role_keyword(item["superuser"], "SUPERUSER", "NOSUPERUSER"), _role_keyword(item["create_db"], "CREATEDB", "NOCREATEDB"), _role_keyword(item["create_role"], "CREATEROLE", "NOCREATEROLE"), _role_keyword(item["inherit"], "INHERIT", "NOINHERIT"), _role_keyword(item["replication"], "REPLICATION", "NOREPLICATION"), _role_keyword(item["bypass_rls"], "BYPASSRLS", "NOBYPASSRLS"), ] statements.append(f'CREATE ROLE "{name}" {" ".join(attributes)};') return "\n".join(statements) def execute_application_role_bootstrap( docker_bin: str, container: str, database: str, source_manifest: dict[str, Any], *, timeout: float, ) -> list[str]: sql = application_role_sql(source_manifest) if not sql: return [] completed = _run( _psql_command(container, database, "-c", sql, docker_bin=docker_bin), timeout=timeout, ) _require_success(completed, "application role bootstrap") return sorted(item["name"] for item in source_manifest["singleton"]["application_roles"].get("items", [])) def query_scalar( docker_bin: str, container: str, database: str, sql: str, *, timeout: float, ) -> str: completed = _run( _psql_command(container, database, "-c", sql, docker_bin=docker_bin), timeout=timeout, ) return _require_success(completed, "PostgreSQL readback").strip() def collect_key_counts( docker_bin: str, container: str, database: str, *, timeout: float, ) -> tuple[dict[str, int], int]: counts: dict[str, int] = {} for table in KEY_TABLES: exists = query_scalar( docker_bin, container, database, f"select to_regclass('{table}') is not null;", timeout=timeout, ) if exists != "t": raise RuntimeError(f"required canonical table is missing after restore: {table}") schema, name = table.split(".", 1) raw_count = query_scalar( docker_bin, container, database, f'select count(*) from "{schema}"."{name}";', timeout=timeout, ) try: counts[table] = int(raw_count) except ValueError as exc: raise RuntimeError(f"invalid row count for {table}: {raw_count!r}") from exc raw_table_count = query_scalar( docker_bin, container, database, "select count(*) from pg_tables where schemaname in ('public', 'kb_stage');", timeout=timeout, ) try: table_count = int(raw_table_count) except ValueError as exc: raise RuntimeError(f"invalid canonical schema table count: {raw_table_count!r}") from exc return counts, table_count def run_parity_check( docker_bin: str, container: str, database: str, *, source_manifest_path: Path, source_manifest: dict[str, Any], manifest_sql: Path, command_timeout: float, max_target_query_ms: float, max_target_source_ratio: float, ) -> dict[str, Any]: copy_result = _run( [docker_bin, "cp", str(manifest_sql), f"{container}:{MANIFEST_DESTINATION}"], timeout=command_timeout, ) _require_success(copy_result, "parity manifest copy") completed = _run( _psql_command( container, database, "-f", MANIFEST_DESTINATION, docker_bin=docker_bin, ), timeout=command_timeout, ) target_payload = _require_success(completed, "target parity manifest capture") with tempfile.TemporaryDirectory(prefix="teleo-canonical-parity-") as temp_dir: target_path = Path(temp_dir) / "target-manifest.jsonl" target_path.write_text(target_payload, encoding="utf-8") target_manifest = load_manifest(target_path) problems, details = compare_manifests( source_manifest, target_manifest, max_target_query_ms=max_target_query_ms, max_target_source_ratio=max_target_source_ratio, ) target_receipt = { "bytes": target_path.stat().st_size, "sha256": sha256_file(target_path), "line_count": sum(1 for line in target_payload.splitlines() if line.strip()), } return { "requested": True, "status": "pass" if not problems else "fail", "verifier": "ops.verify_postgres_parity_manifest.compare_manifests", "source_manifest": { "path": str(source_manifest_path), "bytes": source_manifest_path.stat().st_size, "sha256": sha256_file(source_manifest_path), }, "target_manifest": target_receipt, "problems": problems, "details": details, } def cleanup_container( docker_bin: str, container: str, *, command_timeout: float, poll_interval: float, ) -> dict[str, Any]: cleanup: dict[str, Any] = { "attempted": True, "container": container, "force_remove_attempted": True, "container_absent": None, "status": "unproven", } try: removed = _run([docker_bin, "rm", "--force", container], timeout=command_timeout) cleanup["force_remove_exit_code"] = removed.returncode if removed.returncode != 0: cleanup["force_remove_message"] = _bounded_error(removed) except (OSError, subprocess.TimeoutExpired) as exc: cleanup["force_remove_error"] = str(exc) except (KeyboardInterrupt, TerminationRequested) as exc: cleanup["interrupted"] = str(exc) cleanup["force_remove_error"] = str(exc) last_error = None for attempt in range(1, 6): cleanup["absence_inspection_attempts"] = attempt try: absent, error = inspect_container_absence( docker_bin, container, timeout=command_timeout, ) except (OSError, subprocess.TimeoutExpired) as exc: absent, error = None, str(exc) except (KeyboardInterrupt, TerminationRequested) as exc: cleanup["interrupted"] = str(exc) absent, error = None, str(exc) if absent is True: cleanup["container_absent"] = True cleanup["status"] = "pass" cleanup.pop("absence_inspection_error", None) return cleanup cleanup["container_absent"] = absent last_error = error if error and "interrupted" not in cleanup: cleanup["absence_inspection_error"] = error break if attempt < 5: try: time.sleep(poll_interval) except (KeyboardInterrupt, TerminationRequested) as exc: cleanup["interrupted"] = str(exc) if last_error: cleanup["absence_inspection_error"] = last_error cleanup["status"] = "fail" if cleanup["container_absent"] is False else "unproven" return cleanup def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]: return { "artifact": "local_canonical_postgres_rebuild_canary", "generated_at_utc": datetime.now(UTC).isoformat(), "status": "running", "database": args.database, "dump": { "path": str(args.dump.resolve()), "bytes": None, "sha256": None, "custom_format_signature_valid": False, }, "container": { "name": container, "image": args.image, "network_mode_required": "none", "postgres_data_tmpfs_required": f"rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m", "preexisting_absence_proven": False, "started": False, "isolation": None, }, "application_roles_bootstrapped": [], "key_counts": {}, "canonical_schema_table_count": None, "parity": { "requested": args.source_manifest is not None, "status": "pending" if args.source_manifest is not None else "not_requested", "reason": None if args.source_manifest is not None else "--source-manifest was not supplied", }, "timings_seconds": {}, "cleanup": { "attempted": False, "container": container, "container_absent": None, "status": "pending", }, "safety": { "production_database_touched": False, "network_access_available_to_container": False, "persistent_postgres_storage_used": False, }, "error": None, } def run_canary(args: argparse.Namespace) -> dict[str, Any]: started_at = time.monotonic() container = new_container_name(args.container_prefix) receipt = _initial_receipt(args, container) phase = "validation" source_manifest: dict[str, Any] | None = None primary_error: dict[str, str] | None = None try: hash_started = time.monotonic() validate_custom_dump(args.dump) receipt["dump"].update( { "bytes": args.dump.stat().st_size, "sha256": sha256_file(args.dump), "custom_format_signature_valid": True, } ) receipt["timings_seconds"]["dump_validation_and_hash"] = round(time.monotonic() - hash_started, 6) if args.source_manifest is not None: phase = "source_manifest_validation" source_manifest = load_manifest(args.source_manifest) phase = "container_name_preflight" absent, absence_error = inspect_container_absence( args.docker_bin, container, timeout=args.command_timeout, ) if absent is not True: detail = absence_error or "generated container name unexpectedly already exists" raise RuntimeError(f"could not prove generated container name is unused: {detail}") receipt["container"]["preexisting_absence_proven"] = True phase = "container_start" startup_started = time.monotonic() tmpfs_options = f"{DATA_DIR}:rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m" run_result = _run( [ args.docker_bin, "run", "--detach", "--rm", "--network", "none", "--name", container, "--label", f"{CANARY_LABEL}={container}", "--tmpfs", tmpfs_options, "--env", f"POSTGRES_DB={args.database}", "--env", "POSTGRES_HOST_AUTH_METHOD=trust", args.image, ], timeout=args.command_timeout, ) container_id = _require_success(run_result, "networkless PostgreSQL container start").strip() if not container_id: raise RuntimeError("networkless PostgreSQL container start returned no container id") receipt["container"]["started"] = True receipt["container"]["id_prefix"] = container_id[:12] receipt["container"]["isolation"] = inspect_container_isolation( args.docker_bin, container, expected_image=args.image, expected_label=container, timeout=args.command_timeout, ) phase = "named_database_readiness" readiness_attempts = wait_for_named_database( args.docker_bin, container, args.database, startup_timeout=args.startup_timeout, command_timeout=args.command_timeout, poll_interval=args.poll_interval, ) receipt["container"]["named_database_psql_ready"] = True receipt["container"]["named_database_psql_attempts"] = readiness_attempts receipt["timings_seconds"]["container_start_and_named_database_ready"] = round( time.monotonic() - startup_started, 6, ) if source_manifest is not None: phase = "application_role_bootstrap" receipt["application_roles_bootstrapped"] = execute_application_role_bootstrap( args.docker_bin, container, args.database, source_manifest, timeout=args.command_timeout, ) phase = "dump_copy" copy_result = _run( [args.docker_bin, "cp", str(args.dump.resolve()), f"{container}:{DUMP_DESTINATION}"], timeout=args.command_timeout, ) _require_success(copy_result, "custom dump copy") phase = "restore" restore_started = time.monotonic() restore_result = _run( [ args.docker_bin, "exec", container, "pg_restore", "-U", "postgres", "-d", args.database, "--no-owner", "--no-privileges", "--exit-on-error", DUMP_DESTINATION, ], timeout=args.restore_timeout, ) _require_success(restore_result, "canonical PostgreSQL restore") receipt["timings_seconds"]["pg_restore"] = round(time.monotonic() - restore_started, 6) phase = "key_count_readback" count_started = time.monotonic() key_counts, table_count = collect_key_counts( args.docker_bin, container, args.database, timeout=args.command_timeout, ) receipt["key_counts"] = key_counts receipt["canonical_schema_table_count"] = table_count receipt["timings_seconds"]["key_count_readback"] = round(time.monotonic() - count_started, 6) if source_manifest is not None: phase = "parity_manifest_and_compare" parity_started = time.monotonic() receipt["parity"] = run_parity_check( args.docker_bin, container, args.database, source_manifest_path=args.source_manifest, source_manifest=source_manifest, manifest_sql=args.manifest_sql, command_timeout=args.manifest_timeout, max_target_query_ms=args.max_target_query_ms, max_target_source_ratio=args.max_target_source_ratio, ) receipt["timings_seconds"]["manifest_capture_and_parity_compare"] = round( time.monotonic() - parity_started, 6, ) if receipt["parity"]["status"] != "pass": raise RuntimeError("canonical parity verification failed") receipt["status"] = "pass" except ( OSError, ValueError, RuntimeError, subprocess.TimeoutExpired, json.JSONDecodeError, KeyboardInterrupt, TerminationRequested, ) as exc: primary_error = { "phase": phase, "type": type(exc).__name__, "message": str(exc)[-2000:], } receipt["status"] = "fail" receipt["error"] = primary_error finally: cleanup_started = time.monotonic() cleanup = cleanup_container( args.docker_bin, container, command_timeout=args.command_timeout, poll_interval=args.poll_interval, ) receipt["cleanup"] = cleanup receipt["timings_seconds"]["cleanup"] = round(time.monotonic() - cleanup_started, 6) receipt["timings_seconds"]["total"] = round(time.monotonic() - started_at, 6) if cleanup.get("container_absent") is not True: receipt["status"] = "fail" cleanup_error = { "phase": "cleanup", "type": "CleanupProofError", "message": "container absence was not independently proven", } if primary_error is None: receipt["error"] = cleanup_error else: receipt["cleanup_error"] = cleanup_error if cleanup.get("interrupted"): receipt["status"] = "fail" interruption_error = { "phase": "cleanup", "type": "TerminationRequested", "message": cleanup["interrupted"], } if primary_error is None: receipt["error"] = interruption_error else: receipt["cleanup_interruption"] = interruption_error return receipt def write_receipt(path: Path, payload: dict[str, Any]) -> None: path = path.expanduser() if not path.is_absolute(): path = Path.cwd() / path path.parent.mkdir(parents=True, exist_ok=True) path = path.parent.resolve() / path.name encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode() temporary = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp" descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(descriptor, "wb") as handle: handle.write(encoded) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) os.chmod(path, 0o600) finally: try: temporary.unlink() except FileNotFoundError: pass def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--dump", required=True, type=Path, help="PostgreSQL custom-format pg_dump") parser.add_argument("--database", default="teleo") parser.add_argument("--image", default=DEFAULT_IMAGE) parser.add_argument("--container-prefix", default="teleo-canonical-rebuild") parser.add_argument("--tmpfs-mb", default=1024, type=int) parser.add_argument("--source-manifest", type=Path) parser.add_argument("--manifest-sql", default=DEFAULT_MANIFEST_SQL, type=Path) parser.add_argument("--max-target-query-ms", default=2000.0, type=float) parser.add_argument("--max-target-source-ratio", default=20.0, type=float) parser.add_argument("--startup-timeout", default=60.0, type=float) parser.add_argument("--restore-timeout", default=300.0, type=float) parser.add_argument("--manifest-timeout", default=180.0, type=float) parser.add_argument("--command-timeout", default=30.0, type=float) parser.add_argument("--poll-interval", default=0.25, type=float) parser.add_argument("--docker-bin", default="docker") parser.add_argument("--output", required=True, type=Path) args = parser.parse_args(argv) if not DATABASE_NAME_RE.fullmatch(args.database): parser.error("--database must be a safe PostgreSQL identifier up to 63 characters") if not CONTAINER_PREFIX_RE.fullmatch(args.container_prefix): parser.error("--container-prefix must be 1-40 Docker-safe characters") if not args.image or any(character.isspace() for character in args.image): parser.error("--image must be a non-empty Docker image reference without whitespace") if not args.docker_bin or any(character.isspace() for character in args.docker_bin): parser.error("--docker-bin must be one executable name or path without whitespace") if not 64 <= args.tmpfs_mb <= 65536: parser.error("--tmpfs-mb must be between 64 and 65536") for value, flag in ( (args.startup_timeout, "--startup-timeout"), (args.restore_timeout, "--restore-timeout"), (args.manifest_timeout, "--manifest-timeout"), (args.command_timeout, "--command-timeout"), (args.poll_interval, "--poll-interval"), (args.max_target_query_ms, "--max-target-query-ms"), (args.max_target_source_ratio, "--max-target-source-ratio"), ): if value <= 0: parser.error(f"{flag} must be positive") if not args.dump.is_file(): parser.error("--dump must be an existing regular file") if args.source_manifest is not None: if not args.source_manifest.is_file(): parser.error("--source-manifest must be an existing regular file") if not args.manifest_sql.is_file(): parser.error("--manifest-sql must be an existing regular file when parity is requested") protected_inputs = {args.dump.resolve(), args.manifest_sql.resolve()} if args.source_manifest is not None: protected_inputs.add(args.source_manifest.resolve()) if args.output.resolve() in protected_inputs: parser.error("--output must not overwrite a dump or manifest input") return args def _termination_handler(signum: int, _frame: Any) -> None: raise TerminationRequested(f"received {signal.Signals(signum).name}") def main(argv: list[str] | None = None) -> int: args = parse_args(argv) previous_handlers = {} for watched_signal in (signal.SIGINT, signal.SIGTERM): previous_handlers[watched_signal] = signal.getsignal(watched_signal) signal.signal(watched_signal, _termination_handler) try: payload = run_canary(args) finally: for watched_signal, previous in previous_handlers.items(): signal.signal(watched_signal, previous) try: write_receipt(args.output, payload) except OSError as exc: payload["status"] = "fail" payload["receipt_write_error"] = str(exc) print(json.dumps(payload, indent=2, sort_keys=True)) return 0 if payload["status"] == "pass" else 1 if __name__ == "__main__": raise SystemExit(main())