diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index e6ed822..704089f 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -5,6 +5,7 @@ import json import secrets import sys from pathlib import Path +from uuid import UUID import pytest @@ -31,6 +32,63 @@ proxy_runtime = importlib.import_module("scripts.release.proxy_runtime") proxy_service = importlib.import_module("scripts.release.proxy_service") 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") + + +def test_restoration_keeps_old_runtime_configuration_and_operations_inactive(): + directory = Path("/var/lib/kf-onlinejudge/recovery/rr-" + "1" * 32) + for domain in restore.HISTORICAL: + item = {"path": domain + "/old.json", "uid": 0, "gid": 0, "mode": 0o644, "bytes": 20} + path, uid, gid, mode = restore.target(item, directory) + assert path == directory / "history" / domain / "old.json" + assert (uid, gid, mode) == (0, 0, 0o600) + path, uid, gid, mode = restore.target( + {"path": "artifacts/object", "uid": 1200, "gid": 1200, "mode": 0o600, "bytes": 20}, + directory, + ) + assert path == Path("/var/lib/kf-onlinejudge/artifacts/object") + assert (uid, gid, mode) == (1200, 1200, 0o600) + + +def test_recovery_requires_new_worker_tokens_epoch_and_every_operational_password(): + from copy import deepcopy + + def context(offset): + return { + "operator_attested": True, + "worker_id": str(UUID(int=offset + 1)), + "measurement_nonce": format(offset + 10, "064x"), + "measurement_epoch": format(offset + 20, "064x"), + "worker_token": format(offset + 30, "064x"), + "platform_token": format(offset + 40, "064x"), + "endpoints": { + "database": { + "api": "postgresql://kf_platform:" + + str(offset) * 32 + + "@127.0.0.1:5432/kf_onlinejudge" + }, + "amqp": { + "worker": "amqp://kf_worker:" + str(offset + 1) * 32 + "@127.0.0.1:5672/kf" + }, + }, + } + + previous, current = context(1), context(5) + restore_policy.fresh_identities(previous, current) + for key in ( + "worker_id", + "measurement_nonce", + "measurement_epoch", + "worker_token", + "platform_token", + ): + with pytest.raises(policy.DeliveryError): + restore_policy.fresh_identities(previous, {**current, key: previous[key]}) + changed = deepcopy(current) + changed["endpoints"]["database"]["api"] = previous["endpoints"]["amqp"]["worker"] + with pytest.raises(policy.DeliveryError): + restore_policy.fresh_identities(previous, changed) def test_backup_retention_preserves_a_newer_unexpired_recovery_copy(): diff --git a/scripts/release/README.md b/scripts/release/README.md index c0914c3..671d229 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -376,6 +376,10 @@ 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. `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 @@ -427,3 +431,33 @@ and interruption behavior also still require actual VM acceptance. Backup data and manifests contain private credentials/source/testcase data and must stay outside Git and CI artifacts. Parsing, ordinary adapter tests or a manifest digest are not cold-backup/recovery acceptance evidence. + +`python3.13 -m scripts.release.restore --recovery-id rr-<32-lowercase-hex> +--backup-id ID --manifest-sha256 SHA` performs the cold **data** step on a +different clean boot/node. First install a verified public bundle, generate and +select an attested fresh configuration, and explicitly install the dependency +snapshot with a new broker cookie on empty preprovisioned volumes. Keep all +services stopped; do not initialize a new PostgreSQL cluster over restored data. +The target must have new Worker UUID, measurement nonce/epoch, control tokens +and all DB/MQ passwords. Cold restoration currently requires identical migration +bytes, target platform/toolchains and runtime images; schema-changing upgrades +need their separate Backend Alembic procedure. + +The fixed restore operator reads the backup volume through a read-only mount +view, under 512-MiB memory, zero swap, 32 tasks, 50% CPU and a 30-minute deadline. +It verifies the separate private receipt and every source byte before copying +to fixed targets. It preserves original data identities/modes and every retained +factory replay ID; a new boot's existing Sandbox preparation owns delegation +recreation. It copies old application/judge/dependency/proxy configuration and +maintenance receipts into private inactive `recovery/ID/history/`, never over +the selected current configuration. Unknown/nonempty data targets and differing +existing files are refused. Failed partial data stays stopped for explicit +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. diff --git a/scripts/release/backup.py b/scripts/release/backup.py index 458c7ca..e73df4b 100644 --- a/scripts/release/backup.py +++ b/scripts/release/backup.py @@ -113,7 +113,7 @@ def stopped(selected: dict, cut: dict) -> None: raise DeliveryError("PostgreSQL did not confirm clean cold shutdown") -def volume(*, permit_full: bool = False) -> None: +def volume(*, permit_full: bool = False, read_only: bool = False) -> None: preflight.protected(ROOT, directory=True) if stat.S_IMODE(ROOT.stat().st_mode) != 0o700: raise DeliveryError("private preprovisioned backup volume required") @@ -129,8 +129,9 @@ def volume(*, permit_full: bool = False) -> None: row[3] != "/" or row[sep + 1] != "ext4" or any(r.startswith(("shared:", "master:")) for r in row[6:sep]) - or not {"rw", "nosuid", "nodev", "noexec"} <= opts - or "ro" in opts + or not {"ro" if read_only else "rw", "nosuid", "nodev", "noexec"} <= opts + or (not read_only and "ro" in opts) + or any(other[2] == row[2] and other != row for other in rows) or not CAPACITY * 9 // 10 <= fs.f_frsize * fs.f_blocks <= CAPACITY ): raise DeliveryError("backup volume limits or mount isolation differ") @@ -154,6 +155,40 @@ def volume(*, permit_full: bool = False) -> None: ) +def broker_cookie_fingerprint() -> str: + path = Path("/var/lib/kf-onlinejudge/rabbitmq/.erlang.cookie") + info = path.lstat() + if ( + not stat.S_ISREG(info.st_mode) + or info.st_nlink != 1 + or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (1204, 1204, 0o400) + or info.st_size != 65 + ): + raise DeliveryError("private stopped broker cookie identity differs") + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + with os.fdopen(descriptor, "rb") as stream: + before = os.fstat(stream.fileno()) + contents = stream.read(66) + after = os.fstat(stream.fileno()) + keys = ( + "st_dev", + "st_ino", + "st_uid", + "st_gid", + "st_mode", + "st_size", + "st_mtime_ns", + "st_ctime_ns", + ) + if ( + re.fullmatch(rb"[0-9a-f]{64}\n", contents) is None + or any(getattr(info, key) != getattr(before, key) for key in keys) + or any(getattr(before, key) != getattr(after, key) for key in keys) + ): + raise DeliveryError("private stopped broker cookie changed") + return hashlib.sha256(contents).hexdigest() + + def guarded() -> None: if (os.getuid(), os.geteuid(), os.getgid()) != (0, 0, 0) or not set(os.getgroups()) <= {0}: raise DeliveryError("root backup operator required") @@ -216,6 +251,7 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: refs = backup_manifest.validate_references(backup_manifest.read_metadata(references_file)) cut = read_json(lifecycle.LAST, 32768) stopped(selected, cut) + cookie_sha = broker_cookie_fingerprint() artifact_root = SOURCES["artifacts"].stat() if refs["boot_id"] != cut["boot_id"] or refs["storage_identity"] != { "device": artifact_root.st_dev, @@ -314,11 +350,13 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: stopped(selected, cut) if install.selection() != selected or read_json(lifecycle.LAST, 32768) != cut: raise DeliveryError("selected cut changed during backup") + if broker_cookie_fingerprint() != cookie_sha: + raise DeliveryError("stopped broker cookie changed during backup") manifest = read_json(payload.parent / "manifest.json") now = int(time.time()) value = backup_manifest.validate( { - "schema_version": 1, + "schema_version": 2, "format": backup_manifest.FORMAT, "backup_id": backup_id, "source_sha": manifest["source_sha"], @@ -329,6 +367,7 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: "consistent_cut": cut, "selected": selected, "artifact_references": refs, + "broker_cookie_sha256": cookie_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 23172ff..bbbc33c 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.v1" +FORMAT = "kf.private.cold-backup.v2" MAX_BYTES = 7 * 1024**3 MAX_FILE_BYTES = 256 * 1024**2 MAX_ENTRIES = 50000 @@ -143,6 +143,7 @@ def validate(value: dict) -> dict: "consistent_cut", "selected", "artifact_references", + "broker_cookie_sha256", "images", "directories", "files", @@ -154,7 +155,7 @@ def validate(value: dict) -> dict: ) if ( type(value["schema_version"]) is not int - or value["schema_version"] != 1 + or value["schema_version"] != 2 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 @@ -167,6 +168,7 @@ def validate(value: dict) -> dict: ): raise DeliveryError("private backup identity or authority differs") sha256(value["manifest_sha256"]) + sha256(value["broker_cookie_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/restore.py b/scripts/release/restore.py new file mode 100644 index 0000000..961c0d3 --- /dev/null +++ b/scripts/release/restore.py @@ -0,0 +1,398 @@ +"""Explicit cold data restoration on a clean node; no service/schema/grade writes.""" + +import argparse +import fcntl +import hashlib +import json +import os +import re +import resource +import stat +import subprocess +import time +from pathlib import Path + +from . import ( + backup, + backup_manifest, + dependencies, + dependency_install, + install, + lifecycle, + manifest, + preflight, + restore_policy, +) +from .policy import DeliveryError, canonical, digest, sha256 +from .services import python_environment + +STATE = Path("/var/lib/kf-onlinejudge/recovery") +LIVE = { + name: backup.SOURCES[name] + for name in ("postgres", "artifacts", "imports", "factory", "releases", "images") +} +HISTORICAL = backup_manifest.DOMAINS - set(LIVE) + + +def identifier(value: str) -> str: + if not isinstance(value, str) or re.fullmatch(r"rr-[0-9a-f]{32}", value) is None: + raise DeliveryError("explicit cold recovery ID required") + return value + + +def guarded() -> None: + if (os.getuid(), os.geteuid(), os.getgid()) != (0, 0, 0) or not set(os.getgroups()) <= {0}: + raise DeliveryError("fixed root recovery operator required") + status = dict( + line.split(":", 1) + for line in Path("/proc/self/status").read_text().splitlines() + if ":" in line + ) + if status.get("NoNewPrivs", "").strip() != "1" or any( + int(status.get(key, "-1").strip(), 16) != 15 + for key in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb") + ): + raise DeliveryError("recovery read/copy/chown capability boundary differs") + group = preflight.current_cgroup() + if group != Path("/sys/fs/cgroup/kfonlinejudge.slice/kf-node-restore.service"): + raise DeliveryError("fixed recovery operator unit required") + for path, expected in ( + (group, ["536870912", "0", "32", "50000 100000"]), + (Path("/sys/fs/cgroup/kfonlinejudge.slice"), ["4026531840", "0", "384", "200000 100000"]), + ): + observed = [ + (path / key).read_text().strip() + for key in ("memory.max", "memory.swap.max", "pids.max", "cpu.max") + ] + if observed != expected: + raise DeliveryError("actual recovery operator/node budgets differ") + for key, maximum in ( + (resource.RLIMIT_CORE, 0), + (resource.RLIMIT_FSIZE, backup_manifest.MAX_FILE_BYTES), + (resource.RLIMIT_NOFILE, 256), + (resource.RLIMIT_STACK, 16 * 1024**2), + ): + if resource.getrlimit(key) != (maximum, maximum): + raise DeliveryError("actual recovery process limits differ") + backup.volume(permit_full=True, read_only=True) + + +def stopped() -> None: + install.require_stopped() + dependency_install.require_stopped() + for name in (*lifecycle.APPS, *lifecycle.DEPENDENCIES): + values = backup.manager( + "show", "kf-" + name + ".service", "--property=MainPID,ActiveState,ControlGroup" + ) + if values.get("MainPID") != "0" or values.get("ActiveState") != "inactive": + raise DeliveryError("cold recovery requires stopped required services") + lifecycle.Maintenance.empty_cgroup(name) + processes = [p for p in Path("/proc").iterdir() if p.name.isdecimal()] + if len(processes) > 1024: + raise DeliveryError("recovery process observation exceeds bound") + for process in processes: + try: + status = (process / "status").read_text() + except FileNotFoundError: + continue + row = next( + (line.split()[1:] for line in status.splitlines() if line.startswith("Uid:")), None + ) + if row is None or len(row) != 4: + raise DeliveryError("recovery process identity unavailable") + if any(int(uid) in {1200, 1201, 1202, 1203, 1204, 1205, 100000} for uid in row): + raise DeliveryError("recovery found an unmanaged data writer/task") + for name in ("tasks", "identities", "captures"): + path = Path("/var/lib/kf-judge") / name + if path.exists() and next(path.iterdir(), None) is not None: + raise DeliveryError("cold recovery task/lease/capture resources remain") + if Path("/var/lib/kf-judge/factory/control.sock").exists(): + raise DeliveryError("cold recovery factory socket remains") + + +def target(record: dict, recovery: Path) -> tuple[Path, int, int, int]: + name = backup_manifest.name(record["path"]) + domain, _, suffix = name.partition("/") + if domain in LIVE: + path = LIVE[domain] / suffix if suffix else LIVE[domain] + return path, record["uid"], record["gid"], record["mode"] + path = recovery / "history" / name + return path, 0, 0, 0o700 if "bytes" not in record else 0o600 + + +def member(path: Path, uid: int, gid: int, mode: int, *, directory: bool) -> None: + info = path.lstat() + if ( + (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (uid, gid, mode) + or (not stat.S_ISDIR(info.st_mode) if directory else not stat.S_ISREG(info.st_mode)) + or (not directory and info.st_nlink != 1) + ): + raise DeliveryError("existing recovery destination differs; no overwrite") + + +def write_file( + source: Path, path: Path, item: dict, uid: int, gid: int, mode: int, deadline: float +) -> None: + if path.exists() or path.is_symlink(): + member(path, uid, gid, mode, directory=False) + if path.stat().st_size != item["bytes"] or digest(path) != item["sha256"]: + raise DeliveryError("existing recovery file bytes differ; no overwrite") + return + with os.fdopen(os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC), "rb") as incoming: + with os.fdopen( + os.open( + path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600 + ), + "wb", + ) as outgoing: + total, h = 0, hashlib.sha256() + while block := incoming.read(65536): + total += len(block) + if total > item["bytes"] or time.monotonic() >= deadline: + raise DeliveryError( + "recovery input/deadline changed; private partial target retained" + ) + outgoing.write(block) + h.update(block) + if total != item["bytes"] or h.hexdigest() != item["sha256"]: + raise DeliveryError("recovery input bytes differ; private partial target retained") + outgoing.flush() + os.fchown(outgoing.fileno(), uid, gid) + os.fchmod(outgoing.fileno(), mode) + os.fsync(outgoing.fileno()) + install.sync_directory(path.parent) + + +def restore(recovery_id: str, backup_id: str, expected: str) -> dict: + guarded() + stopped() + selected = install.selection() + payload = install.RELEASES / selected["release_id"] / "payload" + for name in ("restore.py", "restore_policy.py", "backup.py", "backup_manifest.py"): + if digest(payload / "delivery/scripts/release" / name) != digest( + Path(__file__).parent / name + ): + raise DeliveryError("recovery operator differs from selected delivery") + root = backup.ROOT / backup.identifier(backup_id) + value = backup_manifest.verify(root, expected) + boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip() + if value["consistent_cut"]["boot_id"] == boot: + raise DeliveryError("cold recovery requires a different clean boot/node") + if not value["created_unix_seconds"] <= int(time.time()) < value["expires_unix_seconds"]: + raise DeliveryError("private recovery snapshot is future-dated or expired") + data = manifest.verify(payload.parent, selected["manifest_sha256"]) + restore_policy.compatible(value, payload, data, root) + restore_policy.identity_check(value, selected, root) + for name in ("postgres", "rabbitmq"): + dependencies.storage_observation(name) + member( + Path(dependencies.STORAGE[name]["path"]), + dependency_install.IDENTITIES[name][0], + dependency_install.IDENTITIES[name][0], + 0o700, + directory=True, + ) + install.directory(STATE, 0o700) + lock = STATE / "restore.lock" + install.exclusive_file(lock, b"", 0o600) + descriptor = os.open(lock, os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC) + recovery = STATE / identifier(recovery_id) + record = { + "schema_version": 1, + "recovery_id": recovery_id, + "backup_id": backup_id, + "private_backup_sha256": sha256(expected), + "source_sha": data["source_sha"], + "configuration_id": selected["configuration_id"], + "boot_id": boot, + "data_restored": False, + "database_credentials_rotated": False, + "services_started": False, + "execution_verified": False, + } + deadline = time.monotonic() + 1700 + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + stopped() + if recovery.exists() or recovery.is_symlink(): + raise DeliveryError( + "recovery ID exists; incomplete data retained for explicit inspection" + ) + for directory in (install.RELEASES, install.IMAGES): + if {path.name for path in directory.iterdir()} != {selected["release_id"]}: + raise DeliveryError( + "cold recovery node has foreign or historical installed versions" + ) + for domain in ("postgres", "artifacts", "imports", "factory"): + path = LIVE[domain] + if path.exists() or path.is_symlink(): + if path.is_symlink() or not path.is_dir() or next(path.iterdir(), None) is not None: + raise DeliveryError("cold recovery target is not clean") + recovery.mkdir(mode=0o700) + (recovery / "history").mkdir(mode=0o700) + install.exclusive_file(recovery / "prepared.json", canonical(record), 0o600) + for item in sorted( + value["directories"], key=lambda row: (row["path"].count("/"), row["path"]) + ): + path, uid, gid, mode = target(item, recovery) + if path.exists() or path.is_symlink(): + member(path, uid, gid, mode, directory=True) + else: + if path.parent.is_symlink() or not path.parent.is_dir(): + raise DeliveryError("recovery parent is absent or linked") + path.mkdir(mode=0o700) + # Keep new directories Root-owned and writable until verified + # children are complete. Final ownership/mode is applied below. + free = os.statvfs(Path("/var/lib/kf-onlinejudge")) + if free.f_frsize * free.f_bavail < value["payload_bytes"] + 128 * 1024**2: + raise DeliveryError("cold recovery destination reserve insufficient") + for item in value["files"]: + path, uid, gid, mode = target(item, recovery) + write_file(root / "payload" / item["path"], path, item, uid, gid, mode, deadline) + for item in sorted( + value["directories"], key=lambda row: row["path"].count("/"), reverse=True + ): + path, uid, gid, mode = target(item, recovery) + os.chown(path, uid, gid, follow_symlinks=False) + path.chmod(mode) + install.sync_directory(path) + member(path, uid, gid, mode, directory=True) + stopped() + if install.selection() != selected: + raise DeliveryError("selected recovery identity changed") + for item in value["files"]: + path, uid, gid, mode = target(item, recovery) + member(path, uid, gid, mode, directory=False) + if digest(path) != item["sha256"]: + raise DeliveryError("restored data byte verification failed") + record.update( + data_restored=True, + restored_payload_bytes=value["payload_bytes"], + restored_files=len(value["files"]), + restored_directories=len(value["directories"]), + historical_configuration_active=False, + factory_replay_history_retained=True, + ) + install.exclusive_file(recovery / "data-restored.json", canonical(record), 0o600) + install.sync_directory(recovery) + install.sync_directory(STATE) + return record + finally: + os.close(descriptor) + + +def launch(recovery_id: str, backup_id: str, expected: str) -> dict: + identifier(recovery_id) + backup.identifier(backup_id) + sha256(expected) + selected = install.selection() + payload = install.RELEASES / selected["release_id"] / "payload" + stopped() + install.directory(STATE, 0o700) + install.directory(LIVE["factory"], 0o711) + current = backup.manager( + "show", "kf-node-restore.service", "--property=LoadState,ActiveState,MainPID" + ) + if current != {"LoadState": "not-found", "ActiveState": "inactive", "MainPID": "0"}: + raise DeliveryError("unknown or active recovery operator") + command = [ + "/usr/bin/systemd-run", + "--quiet", + "--wait", + "--pipe", + "--collect", + "--unit=kf-node-restore.service", + "--slice=kfonlinejudge.slice", + ] + for key, setting in { + "User": "root", + "Group": "root", + "MemoryMax": "512M", + "MemorySwapMax": "0", + "TasksMax": "32", + "CPUQuota": "50%", + "RuntimeMaxSec": "1800", + "TimeoutStopSec": "15", + "LimitCORE": "0", + "LimitFSIZE": "256M", + "LimitNOFILE": "256", + "LimitSTACK": "16M", + "Restart": "no", + "KillMode": "control-group", + "UMask": "0077", + "NoNewPrivileges": "yes", + "CapabilityBoundingSet": "CAP_CHOWN CAP_DAC_OVERRIDE CAP_DAC_READ_SEARCH CAP_FOWNER", + "AmbientCapabilities": "CAP_CHOWN CAP_DAC_OVERRIDE CAP_DAC_READ_SEARCH CAP_FOWNER", + "ProtectSystem": "strict", + "ProtectHome": "yes", + "PrivateTmp": "yes", + "PrivateDevices": "yes", + "ProtectControlGroups": "yes", + "RestrictAddressFamilies": "AF_UNIX AF_INET", + "ReadOnlyPaths": str(backup.ROOT), + "ReadWritePaths": " ".join( + str(path) + for path in [ + Path(dependencies.STORAGE["postgres"]["path"]), + LIVE["artifacts"], + LIVE["imports"], + LIVE["factory"], + LIVE["releases"], + LIVE["images"], + STATE, + ] + ), + }.items(): + command.append("--property=" + key + "=" + setting) + command += [ + "/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.restore", + "--inside", + "--recovery-id", + recovery_id, + "--backup-id", + backup_id, + "--manifest-sha256", + expected, + ] + result = subprocess.run(command, capture_output=True, timeout=1820, check=False) + if result.returncode or len(result.stdout) > 4096: + raise DeliveryError( + "cold recovery failed; target remains stopped and partial data retained" + ) + return json.loads(result.stdout) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Private bounded cold data restoration") + parser.add_argument("--recovery-id", required=True) + parser.add_argument("--backup-id", required=True) + parser.add_argument("--manifest-sha256", required=True) + parser.add_argument("--inside", action="store_true", help=argparse.SUPPRESS) + args = parser.parse_args() + try: + if os.getuid() != 0 or os.geteuid() != 0: + raise DeliveryError("explicit root cold recovery operator required") + result = (restore if args.inside else launch)( + args.recovery_id, args.backup_id, args.manifest_sha256 + ) + print(json.dumps(result, sort_keys=True)) + except (OSError, ValueError, subprocess.SubprocessError) as error: + parser.exit( + 1, + "Cold recovery refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/restore_policy.py b/scripts/release/restore_policy.py new file mode 100644 index 0000000..9b4dd7f --- /dev/null +++ b/scripts/release/restore_policy.py @@ -0,0 +1,161 @@ +"""Restore compatibility and fresh operational identities; no data/state writes.""" + +import json +from pathlib import Path +from urllib.parse import urlsplit + +from . import configuration, install, manifest +from .policy import DeliveryError, digest, fields, read_json, sha256 + + +def snapshot(paths: dict[str, Path]) -> dict: + """Extract fixed credential roles without serializing any secret to a receipt.""" + if set(paths) != configuration.CONFIG_FILES: + raise DeliveryError("complete recovery configuration snapshot required") + env = { + name: configuration.environment_read(paths[name].read_bytes()) + for name in paths + if name.endswith(".env") + } + coordinator = read_json(paths["coordinator.json"], 32768) + worker = read_json(paths["worker.json"], 32768) + measurement = fields( + read_json(paths["measurement.json"], 32768), + {"binding", "epoch", "operator_attested"}, + "recovery measurement", + ) + + def database(name: str, key: str = "KF_OJ_DATABASE_URL") -> str: + value = env[name].get(key) + if not isinstance(value, str) or not value.startswith("postgresql+asyncpg:"): + raise DeliveryError("recovery backend driver differs") + return value.replace("postgresql+asyncpg:", "postgresql:", 1) + + endpoints = configuration.validate_connections( + { + "schema_version": 1, + "database": { + "api": database("api.env"), + "coordinator": coordinator.get("database_url"), + "import": database("import.env"), + "identity": database("identity.env", "KF_OJ_IDENTITY_OPERATOR_DATABASE_URL"), + "migration": database("migration.env"), + }, + "amqp": { + "outbox": env["outbox.env"].get("KF_OJ_AMQP_URL"), + "worker": worker.get("amqp_url"), + }, + } + ) + binding = measurement["binding"] + node = configuration.validate_profile( + { + "schema_version": 1, + "origin": json.loads(env["api.env"]["KF_OJ_BROWSER_ORIGINS"])[0], + "worker_id": binding["worker_id"], + "measurement_nonce": binding["measurement_nonce"], + } + ) + if worker.get("worker_id") != node["worker_id"]: + raise DeliveryError("recovery worker binding differs") + return { + "endpoints": endpoints, + "worker_id": node["worker_id"], + "measurement_nonce": node["measurement_nonce"], + "measurement_epoch": measurement["epoch"], + "operator_attested": measurement["operator_attested"], + "worker_token": sha256(worker.get("service_token")), + "platform_token": sha256(coordinator.get("platform_token")), + } + + +def fresh_identities(previous: dict, current: dict) -> None: + if ( + current["operator_attested"] is not True + or current["measurement_nonce"] is None + or current["measurement_epoch"] is None + or current["worker_id"] == previous["worker_id"] + or current["measurement_nonce"] == previous["measurement_nonce"] + or current["measurement_epoch"] == previous["measurement_epoch"] + or {current["worker_token"], current["platform_token"]} + & {previous["worker_token"], previous["platform_token"]} + ): + raise DeliveryError("fresh recovery worker/tokens/attested measurement epoch required") + sha256(current["measurement_nonce"]) + sha256(current["measurement_epoch"]) + old_secrets = { + urlsplit(value).password + for kind in ("database", "amqp") + for value in previous["endpoints"][kind].values() + } + new_secrets = { + urlsplit(value).password + for kind in ("database", "amqp") + for value in current["endpoints"][kind].values() + } + if old_secrets & new_secrets: + raise DeliveryError("recovery database/broker credentials must all be fresh") + + +def compatible(backup: dict, payload: Path, target_data: dict, private_root: Path) -> None: + """Cold data restoration requires unchanged migration bytes and runtime ABI. + + A migration-changing upgrade uses the dedicated Alembic upgrade procedure, + never a raw cold copy into an incompatible target schema/runtime. + """ + selected = backup["selected"] + old_root = private_root / "payload/releases" / selected["release_id"] + old = manifest.validate(read_json(old_root / "manifest.json")) + if ( + digest(old_root / "manifest.json") != backup["manifest_sha256"] + or old["source_sha"] != backup["source_sha"] + or old["images"] != backup["images"] + or old["images"] != target_data["images"] + or old["platform"] != target_data["platform"] + or old["toolchains"] != target_data["toolchains"] + ): + raise DeliveryError("cold recovery runtime/platform/toolchain identity differs") + + def migrations(value: dict) -> dict: + return { + row["path"]: (row["sha256"], row["bytes"]) + for row in value["files"] + if row["path"].startswith("backend/migrations/") + } + + if not migrations(old) or migrations(old) != migrations(target_data): + raise DeliveryError("cold recovery requires identical Backend migration inventory") + if (private_root / "payload/postgres/PG_VERSION").read_bytes() != b"18\n": + raise DeliveryError("cold recovery PostgreSQL major version differs") + # Every old data byte was verified by the closed private inventory. This + # check also binds the independently verified target public payload. + if payload != install.RELEASES / target_data["release_id"] / "payload": + raise DeliveryError("recovery target payload is not the installed release") + + +def identity_check(backup: dict, selected: dict, root: Path) -> dict: + paths = { + name: configuration.installed_path( + selected["release_id"], selected["configuration_id"], name + ) + for name in configuration.CONFIG_FILES + } + old = backup["selected"] + prior = { + name: root + / "payload" + / ("judge-config" if name == "runtime.json" else "app-config") + / configuration.installed_path(old["release_id"], old["configuration_id"], name).name + for name in configuration.CONFIG_FILES + } + inventory = {item["path"]: item for item in backup["files"]} + for path in prior.values(): + relative = path.relative_to(root / "payload").as_posix() + if relative not in inventory or digest(path) != inventory[relative]["sha256"]: + raise DeliveryError("old protected recovery configuration is missing") + previous, current = snapshot(prior), snapshot(paths) + fresh_identities(previous, current) + cookie = Path("/var/lib/kf-onlinejudge/rabbitmq/.erlang.cookie") + if digest(cookie) == backup["broker_cookie_sha256"]: + raise DeliveryError("recovery broker distribution cookie must be fresh") + return current