feat(recovery): rotate database identities before reopening admission

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)

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.

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"]),

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:

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 = {

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:

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"])

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):

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()

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()