From 2e201ed7fb6c02d405be40ac41914d43b09e2a94 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Tue, 13 Oct 2026 10:08:58 +0800 Subject: [PATCH] feat(recovery): rotate database identities before reopening admission --- backend/tests/test_delivery.py | 164 +++++++ scripts/release/README.md | 60 ++- scripts/release/backup.py | 18 +- scripts/release/backup_manifest.py | 7 +- scripts/release/dependencies.py | 12 +- scripts/release/dependency_initialize.py | 13 +- scripts/release/install.py | 2 + scripts/release/lifecycle.py | 15 +- scripts/release/recovery_database.py | 534 +++++++++++++++++++++++ scripts/release/recovery_identity.py | 391 +++++++++++++++++ 10 files changed, 1194 insertions(+), 22 deletions(-) create mode 100644 scripts/release/recovery_database.py create mode 100644 scripts/release/recovery_identity.py diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index 704089f..5c2a7b1 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -4,6 +4,7 @@ import importlib import json import secrets import sys +from contextlib import nullcontext from pathlib import Path from uuid import UUID @@ -34,6 +35,168 @@ proxy_guard = importlib.import_module("scripts.release.proxy_guard") backup = importlib.import_module("scripts.release.backup") restore = importlib.import_module("scripts.release.restore") restore_policy = importlib.import_module("scripts.release.restore_policy") +recovery_identity = importlib.import_module("scripts.release.recovery_identity") +recovery_database = importlib.import_module("scripts.release.recovery_database") + + +def test_postgres_recovery_retains_narrow_existing_role_authorities(): + assert recovery_identity.properties("kf_platform") == ( + True, + False, + False, + False, + False, + False, + 8, + True, + None, + None, + ) + assert recovery_identity.memberships("kf_migrator") == [ + ("kf_oj_identity_authority_owner", False, True, True) + ] + assert recovery_identity.memberships("kf_coordinator") == [ + ("kf_oj_coordinator", False, True, False) + ] + assert recovery_identity.memberships("postgres") == [] + + +def postgres_recovery_settings_shape(): + values = recovery_identity.recovery_settings() + rows = [{"name": key, "setting": value, "unit": None} for key, value in values.items()] + normalized = { + "shared_buffers": ("8192", "8kB"), + "work_mem": ("1024", "kB"), + "maintenance_work_mem": ("32768", "kB"), + "statement_timeout": ("5000", "ms"), + "lock_timeout": ("1000", "ms"), + "idle_in_transaction_session_timeout": ("5000", "ms"), + "authentication_timeout": ("5", "s"), + "temp_file_limit": ("16384", "kB"), + "log_parameter_max_length": ("0", "B"), + "log_parameter_max_length_on_error": ("0", "B"), + "log_min_duration_statement": ("-1", "ms"), + "log_min_duration_sample": ("-1", "ms"), + } + for row in rows: + if row["name"] in normalized: + row["setting"], row["unit"] = normalized[row["name"]] + return rows + + +def test_postgres_recovery_readback_compares_effective_units_instead_of_cli_strings(): + rows = postgres_recovery_settings_shape() + recovered = recovery_identity.check_settings(rows) + assert recovered["listen_addresses"]["setting"] == "" + assert recovered["unix_socket_permissions"]["setting"] == "0700" + for row in rows: + if row["name"] == "shared_buffers": + row["setting"] = "16384" + with pytest.raises(policy.DeliveryError): + recovery_identity.check_settings(rows) + + +def recovery_receipts_shape(): + # Pure closed-contract example; no PG connection, Root or Linux acceptance. + prepared = { + "schema_version": 1, + "recovery_id": "rr-" + "1" * 32, + "backup_id": "bk-" + "2" * 32, + "private_backup_sha256": "b" * 64, + "source_sha": "c" * 40, + "configuration_id": "cfg-" + "d" * 24, + "boot_id": "e" * 8 + "-" + "e" * 4 + "-" + "e" * 4 + "-" + "e" * 4 + "-" + "e" * 12, + "data_restored": False, + "database_credentials_rotated": False, + "services_started": False, + "execution_verified": False, + } + data = { + **prepared, + "data_restored": True, + "restored_payload_bytes": 1024, + "restored_files": 10, + "restored_directories": 12, + "historical_configuration_active": False, + "factory_replay_history_retained": True, + } + identity = { + **{ + key: prepared[key] + for key in ( + "schema_version", + "recovery_id", + "source_sha", + "configuration_id", + "boot_id", + ) + }, + "data_receipt_sha256": "f" * 64, + "bootstrap_sha256": "b" * 64, + "cluster_identity": {"device": 1, "inode": 10, "major_version": 18, "uid": 1203}, + "temporary_server_stopped": True, + "database_credentials_rotated": True, + "services_started": False, + "schema_migrated": False, + "execution_state_changed": False, + "execution_verified": False, + "ordinary_identity_readback": { + "dependency": "postgres", + "actual_cgroup_limits": { + "memory.max": "536870912", + "memory.swap.max": "0", + "pids.max": "64", + "cpu.max": "100000 100000", + }, + "data_filesystem_bytes": 2 * 1024**3, + "execution_verified": False, + "rotated_roles": sorted(["postgres", *configuration.PROFILES.values()]), + "role_authority_sha256": "c" * 64, + "authenticated": True, + "schema_migrated": False, + "execution_state_changed": False, + "effective_settings": recovery_identity.check_settings( + postgres_recovery_settings_shape() + ), + "temporary_server_stopped": True, + "services_started": False, + }, + } + return prepared, data, identity + + +@pytest.mark.parametrize("failure", ["unrotated", "running", "wrong_data", "wrong_config", "extra"]) +def test_recovery_completion_refuses_partial_or_unbound_steps(failure): + prepared, data, identity = recovery_receipts_shape() + recovery_database.validate_complete(prepared, data, identity, prepared["recovery_id"], "f" * 64) + if failure == "unrotated": + identity["database_credentials_rotated"] = False + elif failure == "running": + identity["temporary_server_stopped"] = False + elif failure == "wrong_data": + identity["data_receipt_sha256"] = "b" * 64 + elif failure == "wrong_config": + identity["configuration_id"] = "cfg-" + "f" * 24 + else: + identity["skip_verification"] = True + with pytest.raises(policy.DeliveryError): + recovery_database.validate_complete( + prepared, data, identity, prepared["recovery_id"], "f" * 64 + ) + + +def test_startup_cannot_launch_services_when_recovery_admission_is_refused(monkeypatch): + maintenance = object.__new__(lifecycle.Maintenance) + calls = [] + monkeypatch.setattr(maintenance, "start_locked", lambda **_: calls.append("start")) + + def refuse(): + raise policy.DeliveryError("partial recovery") + + monkeypatch.setattr(recovery_database, "admission_guard", refuse) + with pytest.raises(policy.DeliveryError): + maintenance.start(resume=True) + assert calls == [] def test_restoration_keeps_old_runtime_configuration_and_operations_inactive(): @@ -1124,6 +1287,7 @@ def test_maintenance_does_not_leave_admission_open_when_final_confirmation_fails maintenance.selected = {} maintenance.payload = Path("/unused") monkeypatch.setattr(maintenance, "change", lambda *_: None) + monkeypatch.setattr(recovery_database, "admission_guard", nullcontext) def publish(_, state): gate.append(state) diff --git a/scripts/release/README.md b/scripts/release/README.md index 671d229..93dc9f0 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -376,10 +376,12 @@ liveness check. Other health, internal, execution and control routes remain outside the public proxy allowlist; liveness does not attest judging readiness. Private cold backups use a separate format from public release bundles. -The private v2 format also binds the stopped broker cookie fingerprint, so a -restore can reject distribution-secret reuse. It changes no public judging -v1/v2 canonical bytes or historical report. The earlier private v1 prototype -was never system-accepted and is refused by this restore format. +The private v3 format binds the stopped broker cookie and current DB bootstrap +fingerprints and retains +recovery history as another private domain. A restore can reject distribution-secret +reuse without dropping the inactive configuration from an earlier recovery. It +changes no public judging v1/v2 canonical bytes or historical report. The earlier +private v1/v2 prototypes were never system-accepted and are refused by this format. `python3.13 -m kf_oj.infra.artifacts.backup` is an ordinary read-only artifact inventory command. Run it with the selected API database/storage identity after the owner-confirmed drain and while PostgreSQL is still available. It reads every @@ -456,8 +458,48 @@ inspection; it is never silently adopted or erased. The `data-restored.json` receipt proves only byte/identity restoration. The operator starts no service, changes no schema/grant/grade/lease and reports DB -credential rotation and execution verification as false. Recovery still needs -an explicit ordinary PostgreSQL credential-recovery step, Backend Alembic -verification, new TLS inputs, service readiness and actual four-language/history -verification in the second VM before it can be considered complete. Those -steps and the 30-minute end-to-end recovery target remain pending acceptance. +credential rotation and execution verification as false. + +`python3.13 -m scripts.release.recovery_database --recovery-id ID +--data-receipt-sha256 SHA --password-file PRIVATE_NEW_BOOTSTRAP +--password-sha256 SHA` dispatches the credential step only after that boot's +selected complete data receipt. Supply a new 256-bit bootstrap password; its +fingerprint must differ from the separately verified private backup's current +bootstrap fingerprint. Backup capture requires one complete initialization or +recovery marker for the actual cluster, so a restored node can be backed up again. +Current +service endpoints are loaded from the new protected configuration, not backup +URLs. Missing/different authority, an active cluster, unknown unit or incomplete +cleanup is refused; this is an explicit operation, not service startup behavior. + +One ordinary `kf-postgres` temporary unit gets 512-MiB memory, zero swap, 64 +tasks, 100% CPU, a four-minute deadline, cap0/no-new-privileges and private PID/ +network/runtime namespaces. It uses a private 0700 socket directory with TCP +listening disabled. The [PostgreSQL peer mechanism](https://www.postgresql.org/docs/18/auth-peer.html) +maps only this local OS identity to the fixed maintenance database login; fixed +service logins use SCRAM for their credential checks. No trust authentication, +external database listener or runtime Root business credential is introduced. + +The process checks effective PostgreSQL memory/time/socket/logging settings and +the exact existing roles/memberships/database owner. It rotates only passwords +for the five fixed service logins and `postgres` in one transaction, verifies +unchanged role authority and authenticates each new credential. It grants no +membership, creates no database/table, invokes no business repair and does not +rewrite grades, leases or measurement history. SQL/error output containing +utility statements is suppressed. Temporary-server exit and empty resources +must be confirmed before the dispatcher writes `identity-recovered.json`. + +Startup and credential recovery share the cold data step's fixed lock. Startup +holds a read-only shared lock through admission and validates the linked prepared, +data and identity receipts in each bounded Root-private recovery directory. +Missing/incomplete/foreign receipts refuse startup; a committed password change +without a verified shutdown receipt cannot open admission. Complete historical +receipts remain valid after reboot or a later selected version; they do not make +the current services ready. Previous recovery history is backed up and restored +into inactive history, never used as the new node's completion receipt. + +Recovery still needs Backend Alembic verification, new TLS inputs, service +readiness and actual four-language/history checks in the second VM. Neither +data nor credential receipts assert execution safety or healthy public service. +These system steps and the 30-minute end-to-end recovery target remain pending +independent VM acceptance. diff --git a/scripts/release/backup.py b/scripts/release/backup.py index e73df4b..8fb55e3 100644 --- a/scripts/release/backup.py +++ b/scripts/release/backup.py @@ -38,6 +38,7 @@ SOURCES = { "images": install.IMAGES, "dependencies": Path("/etc/kf-dependencies"), "proxy": Path("/etc/kf-proxy"), + "recovery": Path("/var/lib/kf-onlinejudge/recovery"), } @@ -237,10 +238,19 @@ def guarded() -> None: def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: + from .recovery_database import bootstrap_fingerprint + guarded() selected = install.selection() payload = install.RELEASES / selected["release_id"] / "payload" - for name in ("backup.py", "backup_manifest.py"): + for name in ( + "backup.py", + "backup_manifest.py", + "recovery_database.py", + "recovery_identity.py", + "dependency_initialize.py", + "restore.py", + ): if digest(payload / "delivery/scripts/release" / name) != digest( Path(__file__).parent / name ): @@ -252,6 +262,7 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: cut = read_json(lifecycle.LAST, 32768) stopped(selected, cut) cookie_sha = broker_cookie_fingerprint() + bootstrap_sha = bootstrap_fingerprint() artifact_root = SOURCES["artifacts"].stat() if refs["boot_id"] != cut["boot_id"] or refs["storage_identity"] != { "device": artifact_root.st_dev, @@ -352,11 +363,13 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: raise DeliveryError("selected cut changed during backup") if broker_cookie_fingerprint() != cookie_sha: raise DeliveryError("stopped broker cookie changed during backup") + if bootstrap_fingerprint() != bootstrap_sha: + raise DeliveryError("stopped DB credential marker changed during backup") manifest = read_json(payload.parent / "manifest.json") now = int(time.time()) value = backup_manifest.validate( { - "schema_version": 2, + "schema_version": 3, "format": backup_manifest.FORMAT, "backup_id": backup_id, "source_sha": manifest["source_sha"], @@ -368,6 +381,7 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: "selected": selected, "artifact_references": refs, "broker_cookie_sha256": cookie_sha, + "database_bootstrap_sha256": bootstrap_sha, "images": manifest["images"], "directories": sorted(dirs, key=lambda x: x["path"]), "files": sorted(files, key=lambda x: x["path"]), diff --git a/scripts/release/backup_manifest.py b/scripts/release/backup_manifest.py index bbbc33c..d6f7070 100644 --- a/scripts/release/backup_manifest.py +++ b/scripts/release/backup_manifest.py @@ -16,7 +16,7 @@ from .policy import ( unique_object, ) -FORMAT = "kf.private.cold-backup.v2" +FORMAT = "kf.private.cold-backup.v3" MAX_BYTES = 7 * 1024**3 MAX_FILE_BYTES = 256 * 1024**2 MAX_ENTRIES = 50000 @@ -33,6 +33,7 @@ DOMAINS = { "images", "dependencies", "proxy", + "recovery", } @@ -144,6 +145,7 @@ def validate(value: dict) -> dict: "selected", "artifact_references", "broker_cookie_sha256", + "database_bootstrap_sha256", "images", "directories", "files", @@ -155,7 +157,7 @@ def validate(value: dict) -> dict: ) if ( type(value["schema_version"]) is not int - or value["schema_version"] != 2 + or value["schema_version"] != 3 or value["format"] != FORMAT or not isinstance(value["backup_id"], str) or re.fullmatch(r"bk-[0-9a-f]{32}", value["backup_id"]) is None @@ -169,6 +171,7 @@ def validate(value: dict) -> dict: raise DeliveryError("private backup identity or authority differs") sha256(value["manifest_sha256"]) sha256(value["broker_cookie_sha256"]) + sha256(value["database_bootstrap_sha256"]) start = integer(value["created_unix_seconds"], 1, 2**63 - 1, "backup time") end = integer(value["expires_unix_seconds"], start + 1, start + 7 * 86400, "backup retention") if end - start != 7 * 86400: diff --git a/scripts/release/dependencies.py b/scripts/release/dependencies.py index 0a8f79b..3f2f410 100644 --- a/scripts/release/dependencies.py +++ b/scripts/release/dependencies.py @@ -75,8 +75,14 @@ def validate_storage(name: str, observation: dict) -> None: raise DeliveryError("dedicated hard dependency storage budget differs") -def check_runtime(name: str, *, initializing: bool = False) -> dict: - if initializing and name != "postgres": +def check_runtime(name: str, *, initializing: bool = False, recovering: bool = False) -> dict: + if ( + type(initializing) is not bool + or type(recovering) is not bool + or (initializing and recovering) + ): + raise DeliveryError("unexpected dependency maintenance purpose") + if (initializing or recovering) and name != "postgres": raise DeliveryError("unexpected dependency initializer") uid = {"postgres": 1203, "rabbitmq": 1204}[name] if ( @@ -94,7 +100,7 @@ def check_runtime(name: str, *, initializing: bool = False) -> dict: ): raise DeliveryError("dependency privilege boundary differs") group = preflight.current_cgroup() - unit = name + "-initialize" if initializing else name + unit = name + ("-initialize" if initializing else "-recover" if recovering else "") if str(group) != "/sys/fs/cgroup/kfonlinejudge.slice/kf-" + unit + ".service": raise DeliveryError("dependency cgroup identity differs") limits = { diff --git a/scripts/release/dependency_initialize.py b/scripts/release/dependency_initialize.py index ff028c3..32a9b13 100644 --- a/scripts/release/dependency_initialize.py +++ b/scripts/release/dependency_initialize.py @@ -72,16 +72,19 @@ def validate_receipt(value: dict) -> dict: ): raise DeliveryError("initialization receipt binding differs") sha256(value["bootstrap_sha256"]) - identity = fields( - value["cluster_identity"], {"device", "inode", "major_version", "uid"}, "cluster identity" - ) + validate_cluster_identity(value["cluster_identity"]) + return value + + +def validate_cluster_identity(value: dict) -> dict: + identity = fields(value, {"device", "inode", "major_version", "uid"}, "cluster identity") if ( any(type(identity[key]) is not int or identity[key] < 1 for key in identity) or identity["major_version"] != 18 or identity["uid"] != 1203 ): - raise DeliveryError("initialization receipt cluster identity differs") - return value + raise DeliveryError("recorded PostgreSQL cluster identity differs") + return identity def run_in_unit() -> dict: diff --git a/scripts/release/install.py b/scripts/release/install.py index 004a26c..fa439cb 100644 --- a/scripts/release/install.py +++ b/scripts/release/install.py @@ -207,11 +207,13 @@ def files(bundle: Path, expected: str) -> dict: ("/etc/kf-onlinejudge", 0o700), ("/etc/kf-judge", 0o700), ("/var/lib/kf-onlinejudge", 0o755), + ("/var/lib/kf-onlinejudge/recovery", 0o700), ("/var/lib/kf-judge", 0o755), ("/var/lib/kf-judge/tasks", 0o755), (str(IMAGES), 0o755), ): directory(Path(name), mode) + exclusive_file(Path("/var/lib/kf-onlinejudge/recovery/restore.lock"), b"", 0o600) for name in ("artifacts", "imports"): data_directory(Path("/var/lib/kf-onlinejudge") / name, preflight.ACCOUNTS["kf-oj"]) data_directory(Path("/var/lib/kf-onlinejudge/worker"), preflight.ACCOUNTS["kf-worker"]) diff --git a/scripts/release/lifecycle.py b/scripts/release/lifecycle.py index fb27b0b..5dc35b8 100644 --- a/scripts/release/lifecycle.py +++ b/scripts/release/lifecycle.py @@ -471,6 +471,12 @@ class Maintenance: return True def start(self, *, resume: bool) -> None: + from .recovery_database import admission_guard + + with admission_guard(): + self.start_locked(resume=resume) + + def start_locked(self, *, resume: bool) -> None: if not resume: for name in APPS: value = self.unit(name) @@ -522,7 +528,14 @@ class Maintenance: def run(action: str) -> dict: operator_units.guarded("node-maintenance") selected, paths, payload = node_health.binding() - for name in ("lifecycle.py", "availability.py"): + for name in ( + "lifecycle.py", + "availability.py", + "recovery_database.py", + "recovery_identity.py", + "restore.py", + "dependency_initialize.py", + ): shipped = payload / "delivery/scripts/release" / name preflight.protected(shipped) if digest(shipped) != digest(Path(__file__).parent / name): diff --git a/scripts/release/recovery_database.py b/scripts/release/recovery_database.py new file mode 100644 index 0000000..79a609a --- /dev/null +++ b/scripts/release/recovery_database.py @@ -0,0 +1,534 @@ +"""Root dispatch/receipt for an ordinary, finite and isolated DB identity recovery.""" + +import argparse +import fcntl +import json +import os +import re +import stat +import subprocess +from contextlib import contextmanager +from pathlib import Path +from urllib.parse import urlsplit + +from . import ( + backup, + backup_manifest, + configuration, + dependency_initialize, + install, + manifest, + preflight, + recovery_identity, + restore, + restore_policy, +) +from .policy import DeliveryError, canonical, digest, fields, integer, read_json, sha256 +from .services import python_environment + + +def validate_complete( + prepared: dict, data: dict, identity: dict, recovery_id: str, data_sha: str +) -> None: + """A completed historical recovery survives reboot; an incomplete one never does.""" + fields( + prepared, + { + "schema_version", + "recovery_id", + "backup_id", + "private_backup_sha256", + "source_sha", + "configuration_id", + "boot_id", + "data_restored", + "database_credentials_rotated", + "services_started", + "execution_verified", + }, + "prepared recovery", + ) + if ( + type(prepared["schema_version"]) is not int + or prepared["schema_version"] != 1 + or prepared["recovery_id"] != restore.identifier(recovery_id) + or any( + prepared[key] is not False + for key in ( + "data_restored", + "database_credentials_rotated", + "services_started", + "execution_verified", + ) + ) + or not isinstance(prepared["source_sha"], str) + or re.fullmatch(r"[0-9a-f]{40}", prepared["source_sha"]) is None + or not isinstance(prepared["configuration_id"], str) + or re.fullmatch(r"cfg-[0-9a-f]{24}", prepared["configuration_id"]) is None + or not isinstance(prepared["boot_id"], str) + or re.fullmatch( + r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", + prepared["boot_id"], + ) + is None + ): + raise DeliveryError("incomplete or foreign prepared recovery; admission refused") + validate_data_receipt(data, recovery_id, prepared, prepared["source_sha"], prepared["boot_id"]) + if any(data[key] != value for key, value in prepared.items() if key != "data_restored"): + raise DeliveryError("recovery data receipt differs from prepared binding") + fields( + identity, + { + "schema_version", + "recovery_id", + "source_sha", + "configuration_id", + "boot_id", + "data_receipt_sha256", + "bootstrap_sha256", + "cluster_identity", + "temporary_server_stopped", + "database_credentials_rotated", + "ordinary_identity_readback", + "services_started", + "schema_migrated", + "execution_state_changed", + "execution_verified", + }, + "completed recovery identity", + ) + if ( + any( + identity[key] != prepared[key] + for key in ( + "schema_version", + "recovery_id", + "source_sha", + "configuration_id", + "boot_id", + ) + ) + or type(identity["schema_version"]) is not int + or identity["data_receipt_sha256"] != sha256(data_sha) + or identity["temporary_server_stopped"] is not True + or identity["database_credentials_rotated"] is not True + or any( + identity[key] is not False + for key in ( + "services_started", + "schema_migrated", + "execution_state_changed", + "execution_verified", + ) + ) + ): + raise DeliveryError("recovery identity is incomplete or unbound; admission refused") + sha256(identity["bootstrap_sha256"]) + dependency_initialize.validate_cluster_identity(identity["cluster_identity"]) + read_result(canonical(identity["ordinary_identity_readback"])) + + +@contextmanager +def admission_guard(): + """Hold a read-only shared lock through startup, refusing all partial recoveries.""" + preflight.protected(restore.STATE, directory=True) + if stat.S_IMODE(restore.STATE.stat().st_mode) != 0o700: + raise DeliveryError("private recovery state required before admission") + lock = restore.STATE / "restore.lock" + preflight.protected(lock) + configuration.private(lock) + if lock.stat().st_size != 0: + raise DeliveryError("unknown recovery lock contents") + descriptor = os.open(lock, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + try: + fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB) + for count, directory in enumerate(restore.STATE.iterdir(), start=1): + if count > 9: + raise DeliveryError("recovery history exceeds startup observation bound") + if directory == lock: + continue + restore.identifier(directory.name) + preflight.protected(directory, directory=True) + if stat.S_IMODE(directory.stat().st_mode) != 0o700: + raise DeliveryError("private recovery record directory required") + records = {} + for name in ("prepared.json", "data-restored.json", "identity-recovered.json"): + path = directory / name + preflight.protected(path) + configuration.private(path) + records[name] = read_json(path, 16384) + validate_complete( + records["prepared.json"], + records["data-restored.json"], + records["identity-recovered.json"], + directory.name, + digest(directory / "data-restored.json"), + ) + yield + finally: + os.close(descriptor) + + +def bootstrap_fingerprint() -> str: + """Read the current cluster's proven credential marker, including restored nodes.""" + candidates = [] + cluster = dependency_initialize.cluster_identity() + with admission_guard(): + initial = dependency_initialize.RECEIPT + if initial.exists() or initial.is_symlink(): + preflight.protected(initial) + configuration.private(initial) + candidates.append(dependency_initialize.validate_receipt(read_json(initial, 8192))) + for directory in restore.STATE.iterdir(): + if directory.name == "restore.lock": + continue + # admission_guard already validated all linked completion receipts. + candidates.append(read_json(directory / "identity-recovered.json", 16384)) + if len(candidates) != 1 or candidates[0]["cluster_identity"] != cluster: + raise DeliveryError("current DB bootstrap fingerprint is absent or ambiguous") + return sha256(candidates[0]["bootstrap_sha256"]) + + +def validate_data_receipt( + value: dict, recovery_id: str, selected: dict, source_sha: str, boot: str +) -> dict: + fields( + value, + { + "schema_version", + "recovery_id", + "backup_id", + "private_backup_sha256", + "source_sha", + "configuration_id", + "boot_id", + "data_restored", + "database_credentials_rotated", + "services_started", + "execution_verified", + "restored_payload_bytes", + "restored_files", + "restored_directories", + "historical_configuration_active", + "factory_replay_history_retained", + }, + "cold data restoration receipt", + ) + if ( + type(value["schema_version"]) is not int + or value["schema_version"] != 1 + or value["recovery_id"] != restore.identifier(recovery_id) + or value["source_sha"] != source_sha + or value["configuration_id"] != selected["configuration_id"] + or value["boot_id"] != boot + or value["data_restored"] is not True + or value["database_credentials_rotated"] is not False + or value["services_started"] is not False + or value["execution_verified"] is not False + or value["historical_configuration_active"] is not False + or value["factory_replay_history_retained"] is not True + ): + raise DeliveryError("cold data restoration is not this boot's selected complete data step") + backup.identifier(value["backup_id"]) + sha256(value["private_backup_sha256"]) + integer(value["restored_payload_bytes"], 1, 7 * 1024**3, "restored bytes") + for key in ("restored_files", "restored_directories"): + integer(value[key], 1, 50000, "restored entries") + if value["restored_files"] + value["restored_directories"] > 50000: + raise DeliveryError("restored inventory limit differs") + return value + + +def recover(recovery_id: str, data_sha: str, password_file: Path, password_sha: str) -> dict: + # Share the data step's lock with admission. A partial password rotation + # must not race service startup or another restoration operation. + lock = restore.STATE / "restore.lock" + configuration.private(lock) + if lock.stat().st_size != 0: + raise DeliveryError("unknown recovery lock contents") + descriptor = os.open(lock, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + return recover_locked(recovery_id, data_sha, password_file, password_sha) + finally: + os.close(descriptor) + + +def recover_locked(recovery_id: str, data_sha: str, password_file: Path, password_sha: str) -> dict: + if (os.getuid(), os.geteuid(), os.getgid()) != (0, 0, 0): + raise DeliveryError("explicit root recovery identity dispatcher required") + restore.identifier(recovery_id) + restore.stopped() + selected = install.selection() + payload = install.RELEASES / selected["release_id"] / "payload" + data = manifest.verify(payload.parent, selected["manifest_sha256"]) + for name in ( + "recovery_database.py", + "recovery_identity.py", + "restore.py", + "restore_policy.py", + "dependencies.py", + ): + if digest(payload / "delivery/scripts/release" / name) != digest( + Path(__file__).parent / name + ): + raise DeliveryError("recovery identity program differs from selected delivery") + directory = restore.STATE / recovery_id + preflight.protected(directory, directory=True) + preflight_path = directory / "data-restored.json" + preflight.protected(preflight_path) + configuration.private(preflight_path) + if digest(preflight_path) != sha256(data_sha): + raise DeliveryError("separate cold data restoration fingerprint differs") + boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip() + recorded = validate_data_receipt( + read_json(preflight_path, 8192), recovery_id, selected, data["source_sha"], boot + ) + preflight.protected(password_file) + configuration.private(password_file) + if re.fullmatch(r"/[A-Za-z0-9_./-]+", str(password_file)) is None or digest( + password_file + ) != sha256(password_sha): + raise DeliveryError("private fresh DB bootstrap fingerprint/path differs") + dependency_initialize.validate_password(password_file.read_bytes()) + # Bind the latest cluster fingerprint through the separately hashed private + # snapshot. Restored nodes need not have an initial-cluster receipt. + snapshot_path = backup.ROOT / recorded["backup_id"] / "manifest.json" + preflight.protected(snapshot_path) + configuration.private(snapshot_path) + if digest(snapshot_path) != recorded["private_backup_sha256"]: + raise DeliveryError("private recovery snapshot fingerprint changed") + previous = backup_manifest.validate(backup_manifest.read_metadata(snapshot_path)) + if previous["backup_id"] != recorded["backup_id"]: + raise DeliveryError("private recovery snapshot ID differs") + if previous["database_bootstrap_sha256"] == password_sha: + raise DeliveryError("new recovery DB bootstrap secret required") + current = restore_policy.snapshot( + { + name: configuration.installed_path( + selected["release_id"], selected["configuration_id"], name + ) + for name in configuration.CONFIG_FILES + } + ) + fresh_bootstrap = password_file.read_text().strip() + if any( + fresh_bootstrap == urlsplit(endpoint).password + for kind in ("database", "amqp") + for endpoint in current["endpoints"][kind].values() + ): + raise DeliveryError("recovery bootstrap must be distinct from all service credentials") + credential = directory / "identity-connections.json" + install.exclusive_file(credential, canonical(current["endpoints"]), 0o600) + if (directory / "identity-recovered.json").exists(): + raise DeliveryError("DB identities already recovered; no implicit repeat/reset") + unit = recovery_identity.UNIT + status = backup.manager("show", unit, "--property=LoadState,ActiveState,MainPID") + if status != {"LoadState": "not-found", "ActiveState": "inactive", "MainPID": "0"}: + raise DeliveryError("unknown or active DB identity recovery unit") + if recovery_identity.RUNTIME.exists() or recovery_identity.RUNTIME.is_symlink(): + raise DeliveryError("unknown residual DB identity recovery runtime directory") + identity = dependency_initialize.cluster_identity() + description = "KF fixed DB recovery " + recovery_id + "." + selected["configuration_id"] + command = [ + "/usr/bin/systemd-run", + "--quiet", + "--wait", + "--pipe", + "--collect", + "--unit=" + unit, + "--description=" + description, + "--slice=kfonlinejudge.slice", + ] + for key, setting in { + "Type": "exec", + "User": "kf-postgres", + "Group": "kf-postgres", + "MemoryMax": "512M", + "MemorySwapMax": "0", + "CPUQuota": "100%", + "TasksMax": "64", + "RuntimeMaxSec": "240", + "TimeoutStopSec": "45", + "KillMode": "control-group", + "KillSignal": "SIGINT", + "Restart": "no", + "LimitCORE": "0", + "LimitFSIZE": "256M", + "LimitNOFILE": "256", + "LimitSTACK": "16M", + "NoNewPrivileges": "yes", + "CapabilityBoundingSet": "", + "AmbientCapabilities": "", + "UMask": "0077", + "PrivateTmp": "yes", + "PrivateDevices": "yes", + "PrivateNetwork": "yes", + "PrivatePIDs": "yes", + "ProtectHome": "yes", + "ProtectSystem": "strict", + "ProtectControlGroups": "yes", + "RestrictAddressFamilies": "AF_UNIX", + "RuntimeDirectory": recovery_identity.RUNTIME.name, + "RuntimeDirectoryMode": "0700", + "ReadWritePaths": str(dependency_initialize.DATA), + }.items(): + command.append("--property=" + key + "=" + setting) + command += [ + "--property=LoadCredential=connections:" + str(credential), + "--property=LoadCredential=bootstrap:" + str(password_file), + "/usr/bin/env", + "-i", + "PATH=/usr/bin:/bin", + "LC_ALL=C.UTF-8", + *python_environment(payload).split(), + str(payload / "python/bin/python3.13"), + "-s", + "-P", + "-m", + "scripts.release.recovery_identity", + ] + result = None + dispatched = False + try: + dispatched = True + completed = subprocess.run( + command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=290, check=False + ) + if completed.returncode or len(completed.stdout) > 16384: + raise DeliveryError("DB identity recovery failed; target remains stopped") + result = read_result(completed.stdout) + finally: + if dispatched: + active = backup.manager( + "show", unit, "--property=LoadState,ActiveState,MainPID,Description,Transient" + ) + if active.get("LoadState") != "not-found": + if active.get("Description") != description or active.get("Transient") != "yes": + raise DeliveryError("foreign DB recovery unit; refused cleanup") + from . import service_install + + service_install.systemctl("stop", unit) + restore.stopped() + if ( + recovery_identity.RUNTIME.exists() + or dependency_initialize.DATA.joinpath("postmaster.pid").exists() + ): + raise DeliveryError("DB identity recovery shutdown/runtime cleanup unconfirmed") + if install.selection() != selected or dependency_initialize.cluster_identity() != identity: + raise DeliveryError("recovery configuration or cluster identity changed") + if digest(preflight_path) != data_sha: + raise DeliveryError("cold data restoration receipt changed") + record = { + "schema_version": 1, + "recovery_id": recovery_id, + "source_sha": data["source_sha"], + "configuration_id": selected["configuration_id"], + "boot_id": boot, + "data_receipt_sha256": data_sha, + "bootstrap_sha256": password_sha, + "cluster_identity": identity, + "temporary_server_stopped": True, + "database_credentials_rotated": True, + "ordinary_identity_readback": result, + "services_started": False, + "schema_migrated": False, + "execution_state_changed": False, + "execution_verified": False, + } + install.exclusive_file(directory / "identity-recovered.json", canonical(record), 0o600) + return { + "recovery_id": recovery_id, + "database_credentials_rotated": True, + "receipt_sha256": digest(directory / "identity-recovered.json"), + "temporary_server_stopped": True, + "schema_migrated": False, + "services_started": False, + "execution_verified": False, + } + + +def read_result(raw: bytes) -> dict: + from .operator_units import decode + + value = decode(raw, 16384) + fields( + value, + { + "dependency", + "actual_cgroup_limits", + "data_filesystem_bytes", + "execution_verified", + "rotated_roles", + "role_authority_sha256", + "authenticated", + "schema_migrated", + "execution_state_changed", + "effective_settings", + "temporary_server_stopped", + "services_started", + }, + "ordinary DB recovery result", + ) + if ( + value["dependency"] != "postgres" + or value["rotated_roles"] != sorted(["postgres", *configuration.PROFILES.values()]) + or value["actual_cgroup_limits"] + != { + "memory.max": "536870912", + "memory.swap.max": "0", + "pids.max": "64", + "cpu.max": "100000 100000", + } + or value["authenticated"] is not True + or value["temporary_server_stopped"] is not True + or value["schema_migrated"] is not False + or value["execution_state_changed"] is not False + or value["services_started"] is not False + or value["execution_verified"] is not False + ): + raise DeliveryError("ordinary DB recovery result is incomplete or unproven") + integer( + value["data_filesystem_bytes"], 2 * 1024**3 * 9 // 10, 2 * 1024**3, "DB recovery volume" + ) + sha256(value["role_authority_sha256"]) + fields(value["effective_settings"], set(recovery_identity.recovery_settings()), "PG settings") + for item in value["effective_settings"].values(): + fields(item, {"setting", "unit"}, "PG effective setting") + if ( + not isinstance(item["setting"], str) + or item["unit"] is not None + and not isinstance(item["unit"], str) + ): + raise DeliveryError("invalid effective PostgreSQL setting") + recovery_identity.check_settings( + [{"name": key, **item} for key, item in value["effective_settings"].items()] + ) + return value + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Explicit ordinary DB identity recovery after cold restoration" + ) + parser.add_argument("--recovery-id", required=True) + parser.add_argument("--data-receipt-sha256", required=True) + parser.add_argument("--password-file", type=Path, required=True) + parser.add_argument("--password-sha256", required=True) + args = parser.parse_args() + try: + value = recover( + args.recovery_id, args.data_receipt_sha256, args.password_file, args.password_sha256 + ) + print(json.dumps(value, sort_keys=True)) + except (OSError, ValueError, subprocess.SubprocessError) as error: + parser.exit( + 1, + "DB recovery refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/recovery_identity.py b/scripts/release/recovery_identity.py new file mode 100644 index 0000000..616a0e0 --- /dev/null +++ b/scripts/release/recovery_identity.py @@ -0,0 +1,391 @@ +"""Ordinary isolated PostgreSQL role-password recovery, never business-state repair.""" + +import asyncio +import hashlib +import json +import os +import signal +import stat +import subprocess +import time +from pathlib import Path +from urllib.parse import urlsplit + +from . import configuration, database, dependencies, dependency_initialize, preflight +from .policy import DeliveryError, canonical, read_json + +UNIT = "kf-postgres-recover.service" +RUNTIME = Path("/run/kf-postgres-recovery") +CREDENTIALS = Path("/run/credentials") / UNIT +PORT = 5440 +CLIENT_SETTINGS = { + **database.SETTINGS, + "search_path": "pg_catalog", + "session_preload_libraries": "", + "local_preload_libraries": "", + "log_statement": "none", + "log_min_error_statement": "panic", + "log_min_duration_statement": "-1", + "log_min_duration_sample": "-1", + "log_transaction_sample_rate": "0", + "log_parameter_max_length_on_error": "0", +} + + +def properties(name: str) -> tuple: + login = name in configuration.PROFILES.values() + if name == "postgres": + return True, True, True, True, True, True, -1, True, None, None + return login, False, False, False, False, False, 8 if login else -1, True, None, None + + +def memberships(name: str) -> list: + return ( + [] + if name not in database.MEMBERSHIP + else [(database.MEMBERSHIP[name], False, True, name == "kf_migrator")] + ) + + +async def authority(connection) -> dict: + result = {} + for name in ("postgres", *database.GROUPS, *configuration.PROFILES.values()): + row = await connection.fetchrow( + "SELECT rolcanlogin,rolsuper,rolcreatedb,rolcreaterole,rolreplication," + "rolbypassrls,rolconnlimit,rolinherit,rolconfig,rolvaliduntil " + "FROM pg_roles WHERE rolname=$1", + name, + ) + if row is None or tuple(row) != properties(name): + raise DeliveryError("restored DB role authority differs; no overwrite") + actual = sorted( + tuple(row) + for row in await connection.fetch( + "SELECT r.rolname,m.admin_option,m.inherit_option,m.set_option " + "FROM pg_auth_members m JOIN pg_roles r ON r.oid=m.roleid " + "JOIN pg_roles u ON u.oid=m.member WHERE u.rolname=$1", + name, + ) + ) + if actual != memberships(name): + raise DeliveryError("restored DB membership authority differs; no overwrite") + result[name] = {"attributes": tuple(row), "memberships": actual} + logins = await connection.fetch("SELECT rolname FROM pg_roles WHERE rolcanlogin") + if {row[0] for row in logins} != {"postgres", *configuration.PROFILES.values()}: + raise DeliveryError("restored DB has unexpected login authority") + owner = await connection.fetchval( + "SELECT r.rolname FROM pg_database d JOIN pg_roles r ON r.oid=d.datdba WHERE d.datname=$1", + database.DATABASE, + ) + if owner != "kf_migrator": + raise DeliveryError("restored DB owner differs; no overwrite") + if await connection.fetchval( + "SELECT current_user" + ) != "postgres" or not await connection.fetchval( + "SELECT rolsuper FROM pg_roles WHERE rolname=current_user" + ): + raise DeliveryError("local peer recovery authority unavailable") + return result + + +def recovery_settings() -> dict: + return { + "listen_addresses": "", + "port": str(PORT), + "unix_socket_directories": str(RUNTIME), + "unix_socket_permissions": "0700", + "hba_file": str(RUNTIME / "pg_hba.conf"), + "ident_file": str(RUNTIME / "pg_ident.conf"), + "logging_collector": "off", + "log_destination": "stderr", + "log_statement": "none", + "log_min_error_statement": "panic", + "log_min_duration_statement": "-1", + "log_min_duration_sample": "-1", + "log_transaction_sample_rate": "0", + "log_connections": "", + "log_disconnections": "off", + "log_duration": "off", + "log_error_verbosity": "terse", + "log_parameter_max_length": "0", + "log_parameter_max_length_on_error": "0", + "debug_print_parse": "off", + "debug_print_rewritten": "off", + "debug_print_plan": "off", + "log_line_prefix": "", + "shared_preload_libraries": "", + "session_preload_libraries": "", + "local_preload_libraries": "", + "archive_mode": "off", + "archive_command": "", + "archive_library": "", + "ssl": "off", + "fsync": "on", + "full_page_writes": "on", + "restore_command": "", + "archive_cleanup_command": "", + "recovery_end_command": "", + "autovacuum": "off", + "restart_after_crash": "off", + "password_encryption": "scram-sha-256", + "shared_buffers": "64MB", + "max_connections": "16", + "work_mem": "1MB", + "maintenance_work_mem": "32MB", + "max_wal_senders": "0", + "max_worker_processes": "0", + "max_parallel_workers": "0", + "jit": "off", + "statement_timeout": "5000", + "lock_timeout": "1000", + "idle_in_transaction_session_timeout": "5000", + "authentication_timeout": "5", + "temp_file_limit": "16384", + "max_files_per_process": "128", + } + + +def check_settings(rows) -> dict: + actual = {row["name"]: (row["setting"], row["unit"]) for row in rows} + wanted = recovery_settings() + if set(actual) != set(wanted): + raise DeliveryError("temporary recovery setting inventory differs") + byte_values = { + "shared_buffers": 64 * 1024**2, + "work_mem": 1024**2, + "maintenance_work_mem": 32 * 1024**2, + "log_parameter_max_length": 0, + "log_parameter_max_length_on_error": 0, + } + time_values = { + "statement_timeout": 5000, + "lock_timeout": 1000, + "idle_in_transaction_session_timeout": 5000, + "authentication_timeout": 5000, + "log_min_duration_statement": -1, + "log_min_duration_sample": -1, + } + for key, expected in wanted.items(): + value, unit = actual[key] + if key in byte_values: + good = ( + unit in {"B", "kB", "8kB"} + and int(value) * {"B": 1, "kB": 1024, "8kB": 8192}[unit] == byte_values[key] + ) + elif key in time_values: + good = ( + unit in {"ms", "s"} and int(value) * {"ms": 1, "s": 1000}[unit] == time_values[key] + ) + elif key == "unix_socket_permissions": + good = unit is None and int(value, 8 if value.startswith("0") else 10) == 0o700 + elif key == "temp_file_limit": + good = unit == "kB" and value == expected + else: + good = unit is None and value == expected + if not good: + raise DeliveryError("temporary recovery effective setting differs") + return {key: {"setting": value, "unit": unit} for key, (value, unit) in actual.items()} + + +def write_private(path: Path, contents: bytes) -> None: + descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600) + with os.fdopen(descriptor, "wb") as stream: + stream.write(contents) + stream.flush() + os.fsync(stream.fileno()) + + +async def rotate(process: subprocess.Popen, endpoints: dict, password: str) -> dict: + import asyncpg + + deadline = time.monotonic() + 60 + connection = None + while time.monotonic() < deadline: + if process.poll() is not None: + raise DeliveryError("temporary recovery server exited") + try: + connection = await asyncpg.connect( + host=str(RUNTIME), + port=PORT, + user="postgres", + database="postgres", + timeout=3, + command_timeout=5, + server_settings=CLIENT_SETTINGS, + ) + break + except (OSError, asyncpg.PostgresError): + await asyncio.sleep(0.25) + if connection is None: + raise DeliveryError("temporary peer recovery readiness deadline") + try: + before = await authority(connection) + effective = check_settings( + await connection.fetch( + "SELECT name,setting,unit FROM pg_settings WHERE name=ANY($1::text[])", + list(recovery_settings()), + ) + ) + async with connection.transaction(): + await authority(connection) + for key, name in configuration.PROFILES.items(): + secret = urlsplit(endpoints["database"][key]).password + # Fixed names and the validated secret alphabet contain no SQL + # delimiters. Error/statement logging remains disabled above. + await connection.execute("ALTER ROLE " + name + " PASSWORD '" + secret + "'") + await connection.execute("ALTER ROLE postgres PASSWORD '" + password + "'") + if await authority(connection) != before: + raise DeliveryError("DB authority changed during credential recovery") + for key, name in configuration.PROFILES.items(): + client = await asyncpg.connect( + host=str(RUNTIME), + port=PORT, + user=name, + database=database.DATABASE, + password=urlsplit(endpoints["database"][key]).password, + timeout=3, + command_timeout=5, + server_settings=database.SETTINGS, + ) + try: + if tuple(await client.fetchrow("SELECT current_user,current_database()")) != ( + name, + database.DATABASE, + ): + raise DeliveryError("recovered service credential authentication differs") + finally: + await client.close(timeout=3) + client = await asyncpg.connect( + host=str(RUNTIME), + port=PORT, + user="postgres", + database="template1", + password=password, + timeout=3, + command_timeout=5, + server_settings=database.SETTINGS, + ) + try: + if tuple(await client.fetchrow("SELECT current_user,current_database()")) != ( + "postgres", + "template1", + ): + raise DeliveryError("recovered bootstrap credential authentication differs") + finally: + await client.close(timeout=3) + return { + "rotated_roles": sorted(["postgres", *configuration.PROFILES.values()]), + "role_authority_sha256": hashlib.sha256(canonical(before)).hexdigest(), + "effective_settings": effective, + "authenticated": True, + "schema_migrated": False, + "execution_state_changed": False, + } + finally: + await connection.close(timeout=3) + + +def run_in_unit() -> dict: + observed = dependencies.check_runtime("postgres", recovering=True) + if { + line.split(":", 1)[0].strip() for line in Path("/proc/net/dev").read_text().splitlines()[2:] + } != {"lo"}: + raise DeliveryError("network-isolated recovery identity process required") + if dependency_initialize.DATA.joinpath("postmaster.pid").exists(): + raise DeliveryError("restored PostgreSQL cluster is already active") + if any( + (dependency_initialize.DATA / name).exists() + for name in ("standby.signal", "recovery.signal") + ): + raise DeliveryError("identity recovery refuses standby/replay signal files") + dependency_initialize.cluster_identity() + for name, maximum in (("connections", 16384), ("bootstrap", 65)): + info = (CREDENTIALS / name).lstat() + if ( + not stat.S_ISREG(info.st_mode) + or info.st_nlink != 1 + or info.st_uid not in {0, 1203} + or stat.S_IMODE(info.st_mode) not in {0o400, 0o600} + or info.st_size > maximum + ): + raise DeliveryError("private recovery credential identity differs") + endpoints = configuration.validate_connections(read_json(CREDENTIALS / "connections", 16384)) + raw = (CREDENTIALS / "bootstrap").read_bytes() + dependency_initialize.validate_password(raw) + password = raw.decode().strip() + info = RUNTIME.lstat() + if (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (1203, 1203, 0o700) or next( + RUNTIME.iterdir(), None + ) is not None: + raise DeliveryError("private empty recovery runtime directory required") + write_private( + RUNTIME / "pg_hba.conf", + ( + "local postgres postgres peer map=kf_recovery\n" + "local template1 postgres scram-sha-256\n" + "local kf_onlinejudge " + ",".join(configuration.PROFILES.values()) + " scram-sha-256\n" + "local all all reject\n" + ).encode(), + ) + write_private(RUNTIME / "pg_ident.conf", b"kf_recovery kf-postgres postgres\n") + preflight.protected(Path("/usr/bin/postgres")) + version = subprocess.run( + ["/usr/bin/postgres", "--version"], + capture_output=True, + timeout=5, + env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8"}, + ) + if version.returncode or version.stdout.strip() != b"postgres (PostgreSQL) 18.6": + raise DeliveryError("recovery PostgreSQL binary version differs") + command = ["/usr/bin/postgres", "-D", str(dependency_initialize.DATA)] + for key, value in recovery_settings().items(): + command += ["-c", key + "=" + value] + process = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "HOME": "/nonexistent"}, + ) + try: + result = asyncio.run(asyncio.wait_for(rotate(process, endpoints, password), timeout=155)) + finally: + if process.poll() is None: + process.send_signal(signal.SIGINT) + try: + status = process.wait(timeout=40) + except subprocess.TimeoutExpired as error: + process.kill() + process.wait(timeout=5) + raise DeliveryError("temporary recovery shutdown unconfirmed") from error + if status or dependency_initialize.DATA.joinpath("postmaster.pid").exists(): + raise DeliveryError("temporary recovery PostgreSQL clean shutdown unconfirmed") + return { + **observed, + **result, + "temporary_server_stopped": True, + "services_started": False, + "execution_verified": False, + } + + +def main() -> None: + import asyncpg + + try: + print(json.dumps(run_in_unit(), sort_keys=True)) + except ( + OSError, + ValueError, + subprocess.SubprocessError, + asyncpg.PostgresError, + asyncpg.InterfaceError, + ): + raise SystemExit( + "PostgreSQL identity recovery refused; target remains stopped; " + "no schema/grade repair performed." + ) from None + + +if __name__ == "__main__": + main()