1192 lines
45 KiB
Python
Executable file
1192 lines
45 KiB
Python
Executable file
#!/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())
|