#!/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") SETTING_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_.]{0,127}\Z") ALLOWED_APPLICATION_ROLES = frozenset({"kb_gate_owner", "kb_apply", "kb_review", "kb_observatory_read"}) ALLOWED_DATABASE_PRIVILEGES = frozenset({"CONNECT", "CREATE", "TEMPORARY"}) 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 _quote_identifier(value: str) -> str: return '"' + value.replace('"', '""') + '"' def _quote_literal(value: str) -> str: return "'" + value.replace("'", "''") + "'" def _manifest_items(source_manifest: dict[str, Any], kind: str) -> list[Any]: try: items = source_manifest["singleton"][kind].get("items", []) except (AttributeError, KeyError, TypeError) as exc: raise ValueError(f"source manifest is missing {kind}.items") from exc if not isinstance(items, list): raise ValueError(f"source manifest {kind}.items must be a list") return items def _validated_role_name(value: Any, context: str) -> str: if not isinstance(value, str) or not DATABASE_NAME_RE.fullmatch(value): raise ValueError(f"source manifest {context} is not a safe PostgreSQL role name") if value == "PUBLIC": raise ValueError(f"source manifest {context} cannot use PUBLIC as a role name") return value def _validated_role_settings(item: dict[str, Any], field: str, role: str) -> list[tuple[str, str]]: settings = item.get(field) if not isinstance(settings, list): raise ValueError(f"source manifest application role {role} {field} must be a list") parsed: list[tuple[str, str]] = [] seen: set[str] = set() for setting in settings: if not isinstance(setting, dict): raise ValueError(f"source manifest application role {role} {field} item must be an object") name = setting.get("name") value = setting.get("value") if not isinstance(name, str) or not SETTING_NAME_RE.fullmatch(name): raise ValueError(f"source manifest application role {role} has invalid setting name") if not isinstance(value, str) or "\x00" in value: raise ValueError(f"source manifest application role {role} setting {name} has invalid value") if name in seen: raise ValueError(f"source manifest application role {role} has duplicate {field} setting {name}") seen.add(name) parsed.append((name, value)) return sorted(parsed) def _application_role_plan(source_manifest: dict[str, Any], database: str) -> dict[str, Any]: if not DATABASE_NAME_RE.fullmatch(database): raise ValueError("target database name is invalid") items = _manifest_items(source_manifest, "application_roles") parsed_roles: list[dict[str, Any]] = [] 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}") connection_limit = item.get("connection_limit") if type(connection_limit) is not int or connection_limit < -1: raise ValueError(f"source manifest application role {name} has invalid connection_limit") default_transaction_read_only = item.get("default_transaction_read_only") if default_transaction_read_only not in {None, "on", "off"}: raise ValueError(f"source manifest application role {name} has invalid default_transaction_read_only") settings = _validated_role_settings(item, "settings", str(name)) database_settings = _validated_role_settings(item, "database_settings", str(name)) captured_read_only = [value for setting, value in settings if setting == "default_transaction_read_only"] expected_read_only = [] if default_transaction_read_only is None else [default_transaction_read_only] if captured_read_only != expected_read_only: raise ValueError(f"source manifest application role {name} has inconsistent default_transaction_read_only") parsed_roles.append( { "item": item, "name": str(name), "settings": settings, "database_settings": database_settings, } ) statements: list[str] = [] role_setting_statement_count = 0 database_role_setting_statement_count = 0 for parsed in sorted(parsed_roles, key=lambda role: role["name"]): item = parsed["item"] name = parsed["name"] 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"), f"CONNECTION LIMIT {item['connection_limit']}", ] statements.append(f'CREATE ROLE "{name}" {" ".join(attributes)};') for setting, value in parsed["settings"]: statements.append( f"ALTER ROLE {_quote_identifier(name)} SET {_quote_identifier(setting)} TO {_quote_literal(value)};" ) role_setting_statement_count += 1 for setting, value in parsed["database_settings"]: statements.append( f"ALTER ROLE {_quote_identifier(name)} IN DATABASE {_quote_identifier(database)} " f"SET {_quote_identifier(setting)} TO {_quote_literal(value)};" ) database_role_setting_statement_count += 1 return { "sql": "\n".join(statements), "roles": sorted(parsed["name"] for parsed in parsed_roles), "role_setting_statement_count": role_setting_statement_count, "database_role_setting_statement_count": database_role_setting_statement_count, } def application_role_sql(source_manifest: dict[str, Any], database: str) -> str: return str(_application_role_plan(source_manifest, database)["sql"]) def execute_application_role_bootstrap( docker_bin: str, container: str, database: str, source_manifest: dict[str, Any], *, timeout: float, ) -> list[str]: plan = _application_role_plan(source_manifest, database) sql = str(plan["sql"]) 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 list(plan["roles"]) def _membership_plan(source_manifest: dict[str, Any]) -> dict[str, Any]: application_roles = _manifest_items(source_manifest, "application_roles") application_role_names: set[str] = set() initial_grantors = {"postgres"} for item in application_roles: 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}") application_role_names.add(str(name)) if item.get("superuser") is True or item.get("create_role") is True: initial_grantors.add(str(name)) parsed: list[dict[str, Any]] = [] seen: set[tuple[str, str, str]] = set() for item in _manifest_items(source_manifest, "role_memberships"): if not isinstance(item, dict): raise ValueError("source manifest role membership must be an object") role = _validated_role_name(item.get("role"), "role membership role") member = _validated_role_name(item.get("member"), "role membership member") grantor = _validated_role_name(item.get("grantor"), "role membership grantor") if role not in application_role_names and member not in application_role_names: raise ValueError("source manifest role membership is not bound to an application role") for field in ("admin_option", "inherit_option", "set_option"): if type(item.get(field)) is not bool: raise ValueError(f"source manifest role membership has invalid {field}") key = (role, member, grantor) if key in seen: raise ValueError("source manifest contains a duplicate role membership grant") seen.add(key) parsed.append( { "role": role, "member": member, "grantor": grantor, "admin_option": item["admin_option"], "inherit_option": item["inherit_option"], "set_option": item["set_option"], } ) remaining = sorted(parsed, key=lambda row: (row["role"], row["member"], row["grantor"])) ordered: list[dict[str, Any]] = [] authorized: dict[str, set[str]] = {row["role"]: set(initial_grantors) for row in remaining} while remaining: progress = False for row in list(remaining): if row["grantor"] not in authorized[row["role"]]: continue ordered.append(row) remaining.remove(row) if row["admin_option"]: authorized[row["role"]].add(row["member"]) progress = True if not progress: blocked = ", ".join(f"{row['grantor']}->{row['role']}->{row['member']}" for row in remaining) raise ValueError(f"source manifest role membership grantor dependencies cannot be replayed: {blocked}") statements: list[str] = [] for row in ordered: statements.extend( [ f"SET ROLE {_quote_identifier(row['grantor'])};", ( f"GRANT {_quote_identifier(row['role'])} TO {_quote_identifier(row['member'])} WITH " f"ADMIN {str(row['admin_option']).upper()}, " f"INHERIT {str(row['inherit_option']).upper()}, " f"SET {str(row['set_option']).upper()} " f"GRANTED BY {_quote_identifier(row['grantor'])};" ), "RESET ROLE;", ] ) return {"sql": "\n".join(statements), "statement_count": len(ordered)} def role_membership_sql(source_manifest: dict[str, Any]) -> str: return str(_membership_plan(source_manifest)["sql"]) def execute_role_membership_replay( docker_bin: str, container: str, database: str, source_manifest: dict[str, Any], *, timeout: float, ) -> int: plan = _membership_plan(source_manifest) sql = str(plan["sql"]) if not sql: return 0 completed = _run( _psql_command(container, database, "-c", sql, docker_bin=docker_bin), timeout=timeout, ) _require_success(completed, "role membership replay") return int(plan["statement_count"]) def _database_owner(source_manifest: dict[str, Any]) -> str: rows = [ item for item in _manifest_items(source_manifest, "object_ownership") if isinstance(item, dict) and item.get("object_type") == "database" ] if len(rows) != 1: raise ValueError("source manifest must contain exactly one database ownership row") row = rows[0] if row.get("name") != "canonical_database" or row.get("schema") is not None: raise ValueError("source manifest database owner row is not bound to canonical_database") owner = _validated_role_name(row.get("owner"), "database owner") if owner != "postgres": raise ValueError("local parity canary requires postgres as the captured database owner") return owner def _database_acl_plan(source_manifest: dict[str, Any], database: str) -> dict[str, Any]: if not DATABASE_NAME_RE.fullmatch(database): raise ValueError("target database name is invalid") owner = _database_owner(source_manifest) application_roles = _manifest_items(source_manifest, "application_roles") initial_grantors = {owner, "postgres"} for role in application_roles: if not isinstance(role, dict) or role.get("name") not in ALLOWED_APPLICATION_ROLES: raise ValueError("source manifest database ACL references require exact allowlisted roles") if role.get("superuser") is True: initial_grantors.add(str(role["name"])) parsed: list[dict[str, Any]] = [] seen: set[tuple[str, str, str, bool]] = set() for item in _manifest_items(source_manifest, "object_acl"): if not isinstance(item, dict) or item.get("object_type") != "database": continue if item.get("name") != "canonical_database" or item.get("schema") is not None: raise ValueError("source manifest database ACL row is not bound to canonical_database") grantee_value = item.get("grantee") grantee = "PUBLIC" if grantee_value == "PUBLIC" else _validated_role_name(grantee_value, "database ACL grantee") grantor = _validated_role_name(item.get("grantor"), "database ACL grantor") privilege = item.get("privilege_type") if privilege not in ALLOWED_DATABASE_PRIVILEGES: raise ValueError(f"source manifest contains unsupported database privilege: {privilege!r}") is_grantable = item.get("is_grantable") if type(is_grantable) is not bool: raise ValueError("source manifest database ACL row has invalid is_grantable") key = (grantor, grantee, str(privilege), is_grantable) if key in seen: raise ValueError("source manifest contains a duplicate database ACL grant") seen.add(key) parsed.append( { "grantor": grantor, "grantee": grantee, "privilege": str(privilege), "is_grantable": is_grantable, } ) remaining = sorted( parsed, key=lambda row: (row["privilege"], row["grantor"], row["grantee"], row["is_grantable"]), ) ordered: list[dict[str, Any]] = [] authorized: dict[str, set[str]] = {privilege: set(initial_grantors) for privilege in ALLOWED_DATABASE_PRIVILEGES} while remaining: progress = False for row in list(remaining): if row["grantor"] not in authorized[row["privilege"]]: continue ordered.append(row) remaining.remove(row) if row["is_grantable"] and row["grantee"] != "PUBLIC": authorized[row["privilege"]].add(row["grantee"]) progress = True if not progress: blocked = ", ".join(f"{row['grantor']}->{row['grantee']}:{row['privilege']}" for row in remaining) raise ValueError(f"source manifest database ACL grantor dependencies cannot be replayed: {blocked}") reset_sql = """DO $database_acl_reset$ DECLARE acl_grantee text; BEGIN FOR acl_grantee IN SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(acl.grantee) END FROM pg_database database_row CROSS JOIN LATERAL aclexplode( coalesce(database_row.datacl, acldefault('d', database_row.datdba)) ) acl WHERE database_row.datname = current_database() LOOP IF acl_grantee = 'PUBLIC' THEN EXECUTE format( 'REVOKE ALL PRIVILEGES ON DATABASE %I FROM PUBLIC CASCADE', current_database() ); ELSE EXECUTE format( 'REVOKE ALL PRIVILEGES ON DATABASE %I FROM %I CASCADE', current_database(), acl_grantee ); END IF; END LOOP; END $database_acl_reset$;""" statements = [reset_sql] target = _quote_identifier(database) for row in ordered: grantee = "PUBLIC" if row["grantee"] == "PUBLIC" else _quote_identifier(row["grantee"]) suffix = " WITH GRANT OPTION" if row["is_grantable"] else "" statements.extend( [ f"SET ROLE {_quote_identifier(row['grantor'])};", ( f"GRANT {row['privilege']} ON DATABASE {target} TO {grantee}{suffix} " f"GRANTED BY {_quote_identifier(row['grantor'])};" ), "RESET ROLE;", ] ) return {"sql": "\n".join(statements), "grant_count": len(ordered)} def database_acl_sql(source_manifest: dict[str, Any], database: str) -> str: return str(_database_acl_plan(source_manifest, database)["sql"]) def execute_database_acl_replay( docker_bin: str, container: str, database: str, source_manifest: dict[str, Any], *, timeout: float, ) -> int: plan = _database_acl_plan(source_manifest, database) sql = str(plan["sql"]) completed = _run( _psql_command(container, database, "-c", sql, docker_bin=docker_bin), timeout=timeout, ) _require_success(completed, "database ACL replay") return int(plan["grant_count"]) 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": [], "role_setting_statement_count": 0, "database_role_setting_statement_count": 0, "role_membership_statement_count": 0, "database_acl_statement_count": 0, "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" role_plan = _application_role_plan(source_manifest, args.database) receipt["application_roles_bootstrapped"] = execute_application_role_bootstrap( args.docker_bin, container, args.database, source_manifest, timeout=args.command_timeout, ) receipt["role_setting_statement_count"] = role_plan["role_setting_statement_count"] receipt["database_role_setting_statement_count"] = role_plan["database_role_setting_statement_count"] phase = "role_membership_replay" receipt["role_membership_statement_count"] = execute_role_membership_replay( 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", "--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) if source_manifest is not None: phase = "database_acl_replay" receipt["database_acl_statement_count"] = execute_database_acl_replay( args.docker_bin, container, args.database, source_manifest, timeout=args.command_timeout, ) 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())