diff --git a/backend/tests/test_delivery_backup.py b/backend/tests/test_delivery_backup.py new file mode 100644 index 0000000..bebcd0a --- /dev/null +++ b/backend/tests/test_delivery_backup.py @@ -0,0 +1,304 @@ +"""Trusted byte fixtures only; no privileged backup, recovery or RTO acceptance.""" + +import hashlib +import importlib +import os +import sys +import time +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) +backup_manifest = importlib.import_module("scripts.release.backup_manifest") +backup_objects = importlib.import_module("scripts.release.backup_objects") +policy = importlib.import_module("scripts.release.policy") + + +def empty_drain_projection(): + return { + "execution": { + "database_clock_unix_ms": 1791830000000, + **{ + name: {"count_capped": 0, "truncated": False} + for name in ("queued", "in_flight", "cleanup_unconfirmed") + }, + }, + "outbox": { + "schema_version": 1, + "pending_capped": 0, + "locked_capped": 0, + "truncated": False, + }, + "imports": { + "schema_version": 1, + "worker_ready": True, + "reserved_bytes": 0, + "uploaded_capped": 0, + "validating_capped": 0, + "truncated": False, + }, + "broker": { + "schema_version": 1, + "metadata_verified": True, + "messages_ready": 0, + "messages_unacknowledged": 0, + "message_bytes": 0, + "consumers": 1, + }, + "factory": { + "schema_version": 1, + "configuration_sha256": "b" * 64, + "supervisor_authorized": True, + "execution_gate": "required_per_execution", + "isolation_verified": False, + "max_concurrent_runs": 1, + "active_helpers": 0, + "cleanup_complete": True, + "remaining_requests": 4096, + "remaining_lifetime_seconds": 3600, + }, + } + + +@pytest.fixture +def snapshot(): + data = b"trusted repeated bytes\n" + digest = hashlib.sha256(data).hexdigest() + files = [ + { + "path": path, + "uid": uid, + "gid": uid, + "mode": mode, + "bytes": len(data), + "sha256": digest, + } + for path, uid, mode in [ + ("postgres/PG_VERSION", 1203, 0o600), + ("postgres/global/pg_control", 1203, 0o600), + ("releases/first", 0, 0o444), + ("releases/second", 0, 0o555), + ] + ] + boot = "12345678-1234-1234-1234-123456789abc" + selected = { + "schema_version": 1, + "release_id": "sd-" + "c" * 12, + "manifest_sha256": "b" * 64, + "configuration_id": "cfg-" + "d" * 24, + "receipt_sha256": "e" * 64, + } + return { + "schema_version": 3, + "format": backup_manifest.FORMAT, + "backup_id": "bk-" + "2" * 32, + "source_sha": "c" * 40, + "manifest_sha256": selected["manifest_sha256"], + "configuration_id": selected["configuration_id"], + "created_unix_seconds": 1793070000, + "expires_unix_seconds": 1793070000 + 7 * 86400, + "selected": selected, + "consistent_cut": { + "schema_version": 1, + "source_sha": "c" * 40, + "configuration_id": selected["configuration_id"], + "action": "stop", + "phase": "complete", + "success": True, + "events": [], + "root_cleanup_confirmed": True, + "boot_id": boot, + "drain_projection": empty_drain_projection(), + }, + "artifact_references": { + "schema_version": 1, + "kind": "kf.private.artifact-backup-inventory", + "read_only": True, + "quiescence_verified": False, + "boot_id": boot, + "storage_identity": {"device": 1, "inode": 2, "uid": 1200}, + "objects": [], + }, + "broker_cookie_sha256": "b" * 64, + "database_bootstrap_sha256": "e" * 64, + "images": { + key: { + "toolchain_id": key, + "image_kind": key, + "runtime_version": "1", + "image_revision": 1, + "image_digest": "f" * 64, + } + for key in ("gnu", "python313", "java21") + }, + "directories": [ + {"path": path, "uid": 0, "gid": 0, "mode": 0o700} + for path in sorted(backup_manifest.DOMAINS | {"postgres/global"}) + ], + "files": files, + "payload_bytes": len(data) * len(files), + "restored": False, + "execution_verified": False, + } + + +def object_format(value): + return { + **value, + "schema_version": 4, + "format": backup_manifest.OBJECT_FORMAT, + "logical_bytes": value["payload_bytes"], + "payload_bytes": value["files"][0]["bytes"], + } + + +def test_v3_reader_and_v4_logical_permissions_survive_content_deduplication(snapshot, tmp_path): + old = backup_manifest.validate(snapshot) + new = backup_manifest.validate(object_format(snapshot)) + assert len(backup_manifest.physical_files(old)) == 4 + assert len(backup_manifest.physical_files(new)) == 1 + assert backup_manifest.logical_bytes(old) == backup_manifest.logical_bytes(new) + assert new["payload_bytes"] * 4 == new["logical_bytes"] + assert new["files"][2]["mode"] != new["files"][3]["mode"] + for item in new["files"]: + assert backup_manifest.member_path(tmp_path, new, item) == ( + tmp_path / "objects" / item["sha256"] + ) + assert backup_manifest.named_member(tmp_path, old, item["path"]) == ( + tmp_path / "payload" / item["path"] + ) + + +@pytest.mark.parametrize( + "field,change", + [ + ("payload_bytes", 1), + ("logical_bytes", 1), + ("schema_version", True), + ("format", backup_manifest.FORMAT), + ], +) +def test_v4_cannot_confuse_stored_bytes_with_restored_bytes(snapshot, field, change): + value = object_format(snapshot) + value[field] = change + with pytest.raises(policy.DeliveryError): + backup_manifest.validate(value) + + +def test_v4_preserves_logical_versions_above_v3_limit_without_raising_physical_limit(snapshot): + value = object_format(snapshot) + size = backup_manifest.MAX_FILE_BYTES + value["files"] = [ + {**value["files"][0], "path": "releases/version-" + str(index), "bytes": size} + for index in range(32) + ] + [{**item, "bytes": size} for item in snapshot["files"][:2]] + value["logical_bytes"] = len(value["files"]) * size + value["payload_bytes"] = size + backup_manifest.validate(value) + assert value["logical_bytes"] > backup_manifest.MAX_BYTES + old = { + **value, + "schema_version": 3, + "format": backup_manifest.FORMAT, + "payload_bytes": value["logical_bytes"], + } + old.pop("logical_bytes") + with pytest.raises(policy.DeliveryError): + backup_manifest.validate(old) + # Distinct object digests still consume the unchanged physical budget. + for index, item in enumerate(value["files"]): + item["sha256"] = hashlib.sha256(str(index).encode()).hexdigest() + value["payload_bytes"] = value["logical_bytes"] + with pytest.raises(policy.DeliveryError): + backup_manifest.validate(value) + + +def test_store_reads_each_source_but_writes_one_single_link_object(tmp_path, monkeypatch): + source, repeated, objects = tmp_path / "source", tmp_path / "repeated", tmp_path / "objects" + objects.mkdir(mode=0o700) + data = b"trusted source\n" * 2000 + source.write_bytes(data) + repeated.write_bytes(data) + repeated.chmod(0o444) + original = backup_objects.read_stream + observed = [] + + def read(incoming, before, deadline, outgoing=None): + observed.append((before.st_ino, outgoing is not None)) + return original(incoming, before, deadline, outgoing) + + monkeypatch.setattr(backup_objects, "read_stream", read) + seen = {} + first, added = backup_objects.store( + source, source.stat(), objects, seen, time.monotonic() + 10, remaining_bytes=len(data) + ) + second, repeated_bytes = backup_objects.store( + repeated, repeated.stat(), objects, seen, time.monotonic() + 10, remaining_bytes=0 + ) + assert first == second == hashlib.sha256(data).hexdigest() + assert (added, repeated_bytes) == (len(data), 0) + assert observed == [ + (source.stat().st_ino, False), + (source.stat().st_ino, True), + (repeated.stat().st_ino, False), + ] + assert list(objects.iterdir()) == [objects / first] + assert (objects / first).read_bytes() == data + assert (objects / first).stat().st_nlink == 1 + assert (objects / first).stat().st_mode & 0o777 == 0o600 + + +def test_failed_second_source_read_retains_incomplete_private_copy(tmp_path, monkeypatch): + source, objects = tmp_path / "source", tmp_path / "objects" + source.write_bytes(b"trusted fixture") + objects.mkdir(mode=0o700) + original = backup_objects.read_stream + + def read(incoming, before, deadline, outgoing=None): + if outgoing is not None: + raise OSError("fixture write failure") + return original(incoming, before, deadline) + + monkeypatch.setattr(backup_objects, "read_stream", read) + seen = {} + with pytest.raises(OSError): + backup_objects.store( + source, source.stat(), objects, seen, time.monotonic() + 10, remaining_bytes=100 + ) + assert seen == {} + assert list(objects.iterdir()) == [objects / ".pending"] + assert not (tmp_path / "manifest.json").exists() + + +def test_restore_expands_repeated_content_into_independent_files_with_own_modes(tmp_path): + restore = importlib.import_module("scripts.release.restore") + content = tmp_path / "object" + data = b"trusted restoration fixture\n" + content.write_bytes(data) + row = {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()} + first, second = tmp_path / "first", tmp_path / "second" + for path, mode in ((first, 0o600), (second, 0o444)): + witness = restore.write_file( + content, path, row, os.getuid(), os.getgid(), mode, time.monotonic() + 10 + ) + restore.unchanged_file(path, witness, os.getuid(), os.getgid(), mode) + assert path.read_bytes() == data + assert path.stat().st_mode & 0o777 == mode + assert path.stat().st_nlink == 1 + assert len({path.stat().st_ino for path in (content, first, second)}) == 3 + + +def test_delayed_filesystem_write_error_is_not_success(tmp_path, monkeypatch): + class SyncFailure: + def __call__(self, descriptor): + assert os.fstat(descriptor).st_ino == tmp_path.stat().st_ino + return -1 + + class Library: + syncfs = SyncFailure() + + monkeypatch.setattr(backup_objects.ctypes, "CDLL", lambda *_, **__: Library()) + with pytest.raises(policy.DeliveryError, match="filesystem synchronization failed"): + backup_objects.sync_filesystem(tmp_path, time.monotonic() + 10) diff --git a/scripts/release/backup.py b/scripts/release/backup.py index c3a104b..a6cb115 100644 --- a/scripts/release/backup.py +++ b/scripts/release/backup.py @@ -15,12 +15,16 @@ from pathlib import Path from . import ( availability, backup_manifest, + backup_objects, configuration, install, lifecycle, preflight, service_install, ) +from . import ( + manifest as release_manifest, +) from .policy import DeliveryError, canonical, digest, read_json, sha256 from .services import python_environment @@ -237,6 +241,7 @@ def guarded() -> None: raise DeliveryError("backup process limit differs") +@release_manifest.verified_operation def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: from .recovery_database import bootstrap_fingerprint @@ -246,6 +251,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: for name in ( "backup.py", "backup_manifest.py", + "backup_objects.py", + "file_copy.py", "recovery_database.py", "recovery_identity.py", "dependency_initialize.py", @@ -282,9 +289,10 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: if target.exists() or target.is_symlink(): raise DeliveryError("backup target already exists; incomplete backups retained") target.mkdir(mode=0o700) - tree = target / "payload" + tree = target / "objects" tree.mkdir(mode=0o700) - dirs, files, total = [], [], 0 + dirs, files, total, stored = [], [], 0, 0 + objects = {} for domain, source_root in SOURCES.items(): if source_root.is_symlink() or not source_root.is_dir(): raise DeliveryError("a required backup source domain is absent or linked") @@ -310,10 +318,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: backup_manifest._member(entry, set(), directory=True) if not stat.S_ISDIR(info.st_mode): raise DeliveryError("backup source directory changed type") - destination = tree / relative - destination.mkdir(mode=0o700) dirs.append(entry) - if len(dirs) + len(files) > backup_manifest.MAX_ENTRIES: + if len(dirs) + len(files) > backup_manifest.MAX_LOGICAL_ENTRIES: raise DeliveryError("backup directory inventory exceeds bound") for member in names: if (source_dir / member).is_symlink(): @@ -335,29 +341,22 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: raise DeliveryError("backup individual file exceeds bound") total += before.st_size if ( - total > backup_manifest.MAX_BYTES - or len(dirs) + len(files) >= backup_manifest.MAX_ENTRIES + total > backup_manifest.MAX_LOGICAL_BYTES + or len(dirs) + len(files) >= backup_manifest.MAX_LOGICAL_ENTRIES or os.statvfs(ROOT).f_frsize * os.statvfs(ROOT).f_bavail < before.st_size + 64 * 1024**2 ): raise DeliveryError("private backup capacity exhausted") - h = hashlib.sha256() - with os.fdopen( - os.open(p, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC), - "rb", - ) as incoming: - with os.fdopen( - os.open( - destination / member, - os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, - 0o600, - ), - "wb", - ) as outgoing: - copy_file(incoming, outgoing, before, deadline, h) - entry["sha256"] = h.hexdigest() + entry["sha256"], added = backup_objects.store( + p, + before, + tree, + objects, + deadline, + remaining_bytes=backup_manifest.MAX_BYTES - stored, + ) + stored += added files.append(entry) - install.sync_directory(destination) stopped(selected, cut) if install.selection() != selected or read_json(lifecycle.LAST, 32768) != cut: raise DeliveryError("selected cut changed during backup") @@ -369,8 +368,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: now = int(time.time()) value = backup_manifest.validate( { - "schema_version": 3, - "format": backup_manifest.FORMAT, + "schema_version": 4, + "format": backup_manifest.OBJECT_FORMAT, "backup_id": backup_id, "source_sha": manifest["source_sha"], "manifest_sha256": selected["manifest_sha256"], @@ -385,7 +384,8 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: "images": manifest["images"], "directories": sorted(dirs, key=lambda x: x["path"]), "files": sorted(files, key=lambda x: x["path"]), - "payload_bytes": total, + "payload_bytes": stored, + "logical_bytes": total, "restored": False, "execution_verified": False, } @@ -393,16 +393,20 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: raw = canonical(value) if len(raw) > backup_manifest.MAX_METADATA_BYTES: raise DeliveryError("private backup manifest exceeds bound") + backup_objects.sync_filesystem(tree, deadline) install.exclusive_file(target / "manifest.json", raw, 0o600) install.sync_directory(tree) install.sync_directory(target) install.sync_directory(ROOT) fingerprint = digest(target / "manifest.json") + # Release construction-only inventories before the independent byte read. + del value, raw, dirs, files, objects backup_manifest.verify(target, fingerprint) return { "backup_id": backup_id, "private_manifest_sha256": fingerprint, - "payload_bytes": total, + "payload_bytes": stored, + "logical_bytes": total, "source_sha": manifest["source_sha"], "captured": True, "restored": False, @@ -412,26 +416,6 @@ def capture(backup_id: str, references_file: Path, references_sha: str) -> dict: os.close(descriptor) -def copy_file(incoming, outgoing, before, deadline: float, h) -> None: - keys = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns") - observed = os.fstat(incoming.fileno()) - if any(getattr(before, key) != getattr(observed, key) for key in keys): - raise DeliveryError("backup source changed before capture") - count = 0 - while block := incoming.read(65536): - count += len(block) - if count > before.st_size or time.monotonic() >= deadline: - raise DeliveryError("backup source or deadline changed") - h.update(block) - outgoing.write(block) - after = os.fstat(incoming.fileno()) - if count != before.st_size or any(getattr(before, key) != getattr(after, key) for key in keys): - raise DeliveryError("backup source changed during capture") - outgoing.flush() - os.fsync(outgoing.fileno()) - os.fchmod(outgoing.fileno(), 0o600) - - def retention_eligible(expired: dict, retained: dict, now: int) -> None: """An explicit expired snapshot never removes the last verified recovery copy.""" if ( @@ -461,10 +445,11 @@ def prune(backup_id: str, expected: str, keep_id: str, keep_expected: str) -> di retention_eligible(value, keep, int(time.time())) # Remove exactly the already verified inventory, never traverse or delete # an unknown tree. Any interruption leaves a private partial target. - for item in value["files"]: + storage = target / backup_manifest.storage_directory(value) + for item in backup_manifest.physical_files(value): if time.monotonic() >= deadline: raise DeliveryError("private backup pruning deadline") - path = target / "payload" / item["path"] + path = storage / item["path"] info = path.lstat() if ( not stat.S_ISREG(info.st_mode) @@ -476,15 +461,17 @@ def prune(backup_id: str, expected: str, keep_id: str, keep_expected: str) -> di path.unlink() install.sync_directory(path.parent) for item in sorted( - value["directories"], key=lambda item: item["path"].count("/"), reverse=True + value["directories"] if value["schema_version"] == 3 else [], + key=lambda item: item["path"].count("/"), + reverse=True, ): - path = target / "payload" / item["path"] + path = storage / item["path"] preflight.protected(path, directory=True) if stat.S_IMODE(path.stat().st_mode) != 0o700: raise DeliveryError("private backup directory changed before pruning") path.rmdir() install.sync_directory(path.parent) - (target / "payload").rmdir() + storage.rmdir() configuration.private(target / "manifest.json") if digest(target / "manifest.json") != sha256(expected): raise DeliveryError("private backup receipt changed before pruning") @@ -591,7 +578,11 @@ def launch(args) -> dict: result = subprocess.run(command, capture_output=True, timeout=1820, check=False) if result.returncode or len(result.stdout) > 4096: raise DeliveryError( - "private backup action failed; source unchanged and partial target retained" + "private backup action failed; exit=" + + str(result.returncode) + + "; stderr_sha256=" + + hashlib.sha256(result.stderr).hexdigest() + + "; source unchanged and partial target retained" ) return json.loads(result.stdout) @@ -599,13 +590,19 @@ def launch(args) -> dict: def selected_operator() -> None: 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", + "backup_objects.py", + "file_copy.py", + ): if digest(payload / "delivery/scripts/release" / name) != digest( Path(__file__).parent / name ): raise DeliveryError("backup operator differs from selected release") +@release_manifest.verified_operation def main() -> None: parser = argparse.ArgumentParser(description="Private bounded cold backup operator") sub = parser.add_subparsers(dest="action", required=True) diff --git a/scripts/release/backup_manifest.py b/scripts/release/backup_manifest.py index 82f1d1e..b4b869a 100644 --- a/scripts/release/backup_manifest.py +++ b/scripts/release/backup_manifest.py @@ -17,9 +17,13 @@ from .policy import ( ) FORMAT = "kf.private.cold-backup.v3" +OBJECT_FORMAT = "kf.private.cold-backup.v4" MAX_BYTES = 7 * 1024**3 MAX_FILE_BYTES = 256 * 1024**2 MAX_ENTRIES = 65536 +# v4 preserves logical versions independently of their deduplicated storage. +MAX_LOGICAL_BYTES = 12 * 1024**3 +MAX_LOGICAL_ENTRIES = 81920 MAX_METADATA_BYTES = 16 * 1024**2 DOMAINS = { "postgres", @@ -130,6 +134,8 @@ def validate_references(value: dict) -> dict: def validate(value: dict) -> dict: + version = value.get("schema_version") if isinstance(value, dict) else None + object_store = type(version) is int and version == 4 fields( value, { @@ -152,13 +158,14 @@ def validate(value: dict) -> dict: "payload_bytes", "restored", "execution_verified", - }, + } + | ({"logical_bytes"} if object_store else set()), "private backup", ) if ( type(value["schema_version"]) is not int - or value["schema_version"] != 3 - or value["format"] != FORMAT + or value["schema_version"] not in {3, 4} + or value["format"] != (OBJECT_FORMAT if object_store else FORMAT) or not isinstance(value["backup_id"], str) or re.fullmatch(r"bk-[0-9a-f]{32}", value["backup_id"]) is None or not isinstance(value["source_sha"], str) @@ -238,23 +245,35 @@ def validate(value: dict) -> dict: if ( not isinstance(dirs, list) or not isinstance(files, list) - or not 1 <= len(dirs) + len(files) <= MAX_ENTRIES + or not 1 <= len(dirs) + len(files) <= (MAX_LOGICAL_ENTRIES if object_store else MAX_ENTRIES) ): raise DeliveryError("backup inventory count exceeds bound") seen = set() for item in dirs: fields(item, {"path", "uid", "gid", "mode"}, "backup directory") _member(item, seen, directory=True) - total = 0 + total, objects = 0, {} for item in files: fields(item, {"path", "uid", "gid", "mode", "bytes", "sha256"}, "backup file") _member(item, seen, directory=False) total += integer(item["bytes"], 0, MAX_FILE_BYTES, "backup file bytes") sha256(item["sha256"]) + if item["sha256"] in objects and objects[item["sha256"]] != item["bytes"]: + raise DeliveryError("private backup object size conflict") + objects[item["sha256"]] = item["bytes"] + physical = sum(objects.values()) if object_store else total if ( type(value["payload_bytes"]) is not int - or value["payload_bytes"] != total - or total > MAX_BYTES + or value["payload_bytes"] != physical + or physical > MAX_BYTES + or ( + object_store + and ( + type(value["logical_bytes"]) is not int + or value["logical_bytes"] != total + or total > MAX_LOGICAL_BYTES + ) + ) ): raise DeliveryError("private backup aggregate bytes differ") directories = {item["path"] for item in dirs} @@ -278,6 +297,36 @@ def validate(value: dict) -> dict: return value +def logical_bytes(value: dict) -> int: + return value["logical_bytes"] if value["schema_version"] == 4 else value["payload_bytes"] + + +def storage_directory(value: dict) -> str: + return "objects" if value["schema_version"] == 4 else "payload" + + +def member_path(root: Path, value: dict, item: dict) -> Path: + return ( + root + / storage_directory(value) + / (item["sha256"] if value["schema_version"] == 4 else item["path"]) + ) + + +def named_member(root: Path, value: dict, name: str) -> Path: + for item in value["files"]: + if item["path"] == name: + return member_path(root, value, item) + raise DeliveryError("private backup required logical member missing") + + +def physical_files(value: dict) -> list[dict]: + if value["schema_version"] == 3: + return value["files"] + objects = {item["sha256"]: item["bytes"] for item in value["files"]} + return [{"path": key, "sha256": key, "bytes": size} for key, size in objects.items()] + + def _member(item: dict, seen: set[str], *, directory: bool) -> None: path = name(item["path"]) if path in seen: @@ -307,14 +356,14 @@ def verify(root: Path, expected: str) -> dict: raise DeliveryError("private backup directory required") path = root / "manifest.json" configuration.private(path) - if digest(path) != sha256(expected) or {p.name for p in root.iterdir()} != { - "manifest.json", - "payload", - }: + if digest(path) != sha256(expected): raise DeliveryError("private backup receipt or layout differs") value = validate(read_metadata(path)) + storage = storage_directory(value) + if {p.name for p in root.iterdir()} != {"manifest.json", storage}: + raise DeliveryError("private backup receipt or layout differs") found_files, found_dirs = set(), set() - payload = root / "payload" + payload = root / storage preflight.protected(payload, directory=True) if stat.S_IMODE(payload.stat().st_mode) != 0o700: raise DeliveryError("private backup payload directory required") @@ -322,7 +371,11 @@ def verify(root: Path, expected: str) -> dict: for member in names + files: p = Path(directory) / member info = p.lstat() - relative = name(p.relative_to(payload).as_posix()) + relative = p.relative_to(payload).as_posix() + if value["schema_version"] == 4: + sha256(relative) + else: + name(relative) if info.st_uid != 0 or info.st_gid != 0 or p.is_symlink(): raise DeliveryError("private backup member ownership differs") if stat.S_ISDIR(info.st_mode) and stat.S_IMODE(info.st_mode) == 0o700: @@ -335,14 +388,39 @@ def verify(root: Path, expected: str) -> dict: found_files.add(relative) else: raise DeliveryError("private backup special or non-private member") - if len(found_files) + len(found_dirs) > MAX_ENTRIES: + if len(found_files) + len(found_dirs) > ( + MAX_LOGICAL_ENTRIES if value["schema_version"] == 4 else MAX_ENTRIES + ): raise DeliveryError("private backup traversal exceeds bound") - if found_dirs != {x["path"] for x in value["directories"]} or found_files != { - x["path"] for x in value["files"] - }: + physical = physical_files(value) + if found_dirs != ( + {x["path"] for x in value["directories"]} if value["schema_version"] == 3 else set() + ) or found_files != {x["path"] for x in physical}: raise DeliveryError("private backup inventory differs") - for item in value["files"]: + from . import file_copy + + for item in physical: p = payload / item["path"] - if p.stat().st_size != item["bytes"] or digest(p) != item["sha256"]: + before = p.lstat() + if ( + not stat.S_ISREG(before.st_mode) + or before.st_nlink != 1 + or (before.st_uid, before.st_gid, stat.S_IMODE(before.st_mode)) != (0, 0, 0o600) + or before.st_size != item["bytes"] + ): + raise DeliveryError("private backup object protection changed") + with os.fdopen(os.open(p, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC), "rb") as stream: + if file_copy._identity(os.fstat(stream.fileno())) != file_copy._identity(before): + raise DeliveryError("private backup object changed before verification") + from hashlib import file_digest + + observed = file_digest(stream, "sha256").hexdigest() + after = os.fstat(stream.fileno()) + if ( + before.st_size != item["bytes"] + or observed != item["sha256"] + or file_copy._identity(after) != file_copy._identity(before) + or file_copy._identity(p.lstat()) != file_copy._identity(before) + ): raise DeliveryError("private backup byte verification failed") return value diff --git a/scripts/release/backup_objects.py b/scripts/release/backup_objects.py new file mode 100644 index 0000000..6bda20e --- /dev/null +++ b/scripts/release/backup_objects.py @@ -0,0 +1,117 @@ +"""Private content storage; every logical source byte is read before deduplication.""" + +import ctypes +import hashlib +import os +import stat +import time +from pathlib import Path + +from . import file_copy +from .policy import DeliveryError + +CHUNK = 1024 * 1024 + + +def identity(info: os.stat_result) -> tuple[int, ...]: + return file_copy._identity(info) + + +def read_stream(incoming, before, deadline: float, outgoing=None) -> str: + if identity(os.fstat(incoming.fileno())) != identity(before): + raise DeliveryError("backup source changed before capture") + count, digest = 0, hashlib.sha256() + while block := incoming.read(CHUNK): + count += len(block) + if count > before.st_size or time.monotonic() >= deadline: + raise DeliveryError("backup source or deadline changed") + digest.update(block) + if outgoing is not None: + outgoing.write(block) + if ( + count != before.st_size + or time.monotonic() >= deadline + or identity(os.fstat(incoming.fileno())) != identity(before) + ): + raise DeliveryError("backup source changed during capture") + return digest.hexdigest() + + +def store( + source: Path, + before, + objects: Path, + seen: dict[str, int], + deadline: float, + *, + remaining_bytes: int, +) -> tuple[str, int]: + """Hash every source; copy only new content, with a second full byte check. + + The one fixed pending file is exclusive. An interrupted copy remains private + and prevents adoption of the incomplete snapshot. Publishing never replaces + an existing object, and the final object has exactly one link. + """ + if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1: + raise DeliveryError("special or linked backup source refused") + with os.fdopen( + os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC), + "rb", + ) as incoming: + digest = read_stream(incoming, before, deadline) + added = 0 + if digest in seen: + if seen[digest] != before.st_size: + raise DeliveryError("backup content size conflict") + else: + if before.st_size > remaining_bytes: + raise DeliveryError("private backup physical capacity exhausted") + pending = objects / ".pending" + with os.fdopen( + os.open( + pending, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, + 0o600, + ), + "wb", + ) as outgoing: + os.fchmod(outgoing.fileno(), 0o600) + incoming.seek(0) + if read_stream(incoming, before, deadline, outgoing) != digest: + raise DeliveryError("backup source bytes changed between reads") + outgoing.flush() + written = os.fstat(outgoing.fileno()) + if identity(pending.lstat()) != identity(written): + raise DeliveryError("private backup pending object changed") + os.link(pending, objects / digest, follow_symlinks=False) + # Only this operation's completed, exclusively created staging link. + pending.unlink() + seen[digest] = before.st_size + added = before.st_size + if identity(source.lstat()) != identity(before): + raise DeliveryError("backup source path changed during capture") + return digest, added + + +def sync_filesystem(directory: Path, deadline: float) -> None: + """Flush the dedicated backup filesystem before publishing its manifest. + + Linux syncfs reports delayed write errors and waits for data and metadata. + The existing service deadline bounds a blocked filesystem operation too. + """ + if time.monotonic() >= deadline: + raise DeliveryError("private backup capture deadline before synchronization") + descriptor = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) + try: + libc = ctypes.CDLL(None, use_errno=True) + try: + syncfs = libc.syncfs + except AttributeError as error: + raise DeliveryError("private backup filesystem synchronization unavailable") from error + syncfs.argtypes, syncfs.restype = [ctypes.c_int], ctypes.c_int + if syncfs(descriptor) != 0: + raise DeliveryError("private backup filesystem synchronization failed") + if time.monotonic() >= deadline: + raise DeliveryError("private backup capture deadline after synchronization") + finally: + os.close(descriptor) diff --git a/scripts/release/recovery_database.py b/scripts/release/recovery_database.py index 9d89dfa..0cd2289 100644 --- a/scripts/release/recovery_database.py +++ b/scripts/release/recovery_database.py @@ -152,7 +152,11 @@ def admission_guard(): 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"): + for name in ( + "prepared.json", + "data-restored.json", + "identity-recovered.json", + ): path = directory / name preflight.protected(path) configuration.private(path) @@ -231,10 +235,18 @@ def validate_data_receipt( 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") + integer( + value["restored_payload_bytes"], + 1, + backup_manifest.MAX_LOGICAL_BYTES, + "restored bytes", + ) for key in ("restored_files", "restored_directories"): - integer(value[key], 1, 50000, "restored entries") - if value["restored_files"] + value["restored_directories"] > 50000: + integer(value[key], 1, backup_manifest.MAX_LOGICAL_ENTRIES, "restored entries") + if ( + value["restored_files"] + value["restored_directories"] + > backup_manifest.MAX_LOGICAL_ENTRIES + ): raise DeliveryError("restored inventory limit differs") return value @@ -395,7 +407,11 @@ def recover_locked(recovery_id: str, data_sha: str, password_file: Path, passwor try: dispatched = True completed = subprocess.run( - command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=290, check=False + 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") @@ -403,7 +419,9 @@ def recover_locked(recovery_id: str, data_sha: str, password_file: Path, passwor finally: if dispatched: active = backup.manager( - "show", unit, "--property=LoadState,ActiveState,MainPID,Description,Transient" + "show", + unit, + "--property=LoadState,ActiveState,MainPID,Description,Transient", ) if active.get("LoadState") != "not-found": if active.get("Description") != description or active.get("Transient") != "yes": @@ -491,10 +509,17 @@ def read_result(raw: bytes) -> dict: ): 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" + 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") + 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 ( @@ -520,7 +545,10 @@ def main() -> None: args = parser.parse_args() try: value = recover( - args.recovery_id, args.data_receipt_sha256, args.password_file, args.password_sha256 + 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: diff --git a/scripts/release/restore.py b/scripts/release/restore.py index 672ae2c..4ebf7b9 100644 --- a/scripts/release/restore.py +++ b/scripts/release/restore.py @@ -62,7 +62,10 @@ def guarded() -> None: raise DeliveryError("fixed recovery operator unit required") for path, expected in ( (group, ["536870912", "0", "32", f"{CPU_PERCENT * 1000} 100000"]), - (Path("/sys/fs/cgroup/kfonlinejudge.slice"), ["4026531840", "0", "384", "200000 100000"]), + ( + Path("/sys/fs/cgroup/kfonlinejudge.slice"), + ["4026531840", "0", "384", "200000 100000"], + ), ): observed = [ (path / key).read_text().strip() @@ -86,7 +89,9 @@ def stopped() -> None: dependency_install.require_stopped() for name in (*lifecycle.APPS, *lifecycle.DEPENDENCIES): values = backup.manager( - "show", "kf-" + name + ".service", "--property=MainPID,ActiveState,ControlGroup" + "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") @@ -100,7 +105,8 @@ def stopped() -> None: except FileNotFoundError: continue row = next( - (line.split()[1:] for line in status.splitlines() if line.startswith("Uid:")), None + (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") @@ -182,18 +188,22 @@ def write_file( if path.exists() or path.is_symlink(): # Existing destinations are read and checked, never overwritten. return readback(path, item, uid, gid, mode, deadline) - with os.fdopen(os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC), "rb") as incoming: - with os.fdopen( + with ( + os.fdopen(os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC), "rb") as incoming, + os.fdopen( os.open( - path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600 + path, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, + 0o600, ), "wb", - ) as outgoing: - file_copy.transfer(incoming, outgoing, item["bytes"], deadline=deadline) - outgoing.flush() - os.fchown(outgoing.fileno(), uid, gid) - os.fchmod(outgoing.fileno(), mode) - os.fsync(outgoing.fileno()) + ) as outgoing, + ): + file_copy.transfer(incoming, outgoing, item["bytes"], deadline=deadline) + outgoing.flush() + os.fchown(outgoing.fileno(), uid, gid) + os.fchmod(outgoing.fileno(), mode) + os.fsync(outgoing.fileno()) # The backup was fully verified before copying. Independently read every # flushed destination while its pages are still local, then bind that proof # to its unchanged inode/metadata through completion on the stopped node. @@ -211,6 +221,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: "restore_policy.py", "backup.py", "backup_manifest.py", + "backup_objects.py", "manifest.py", "file_copy.py", ): @@ -271,9 +282,10 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: ) 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") + if (path.exists() or path.is_symlink()) and ( + 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) @@ -290,12 +302,20 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: # 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: + if free.f_frsize * free.f_bavail < backup_manifest.logical_bytes(value) + 128 * 1024**2: raise DeliveryError("cold recovery destination reserve insufficient") def copy(item: dict) -> tuple[int, ...]: path, uid, gid, mode = target(item, recovery) - return write_file(root / "payload" / item["path"], path, item, uid, gid, mode, deadline) + return write_file( + backup_manifest.member_path(root, value, item), + path, + item, + uid, + gid, + mode, + deadline, + ) witnesses = file_copy.bounded_copies(value["files"], copy, deadline=deadline) verified = { @@ -326,7 +346,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: unchanged_file(path, verified[item["path"]], uid, gid, mode) record.update( data_restored=True, - restored_payload_bytes=value["payload_bytes"], + restored_payload_bytes=backup_manifest.logical_bytes(value), restored_files=len(value["files"]), restored_directories=len(value["directories"]), historical_configuration_active=False, diff --git a/scripts/release/restore_policy.py b/scripts/release/restore_policy.py index 9b4dd7f..b2fa0ba 100644 --- a/scripts/release/restore_policy.py +++ b/scripts/release/restore_policy.py @@ -4,7 +4,7 @@ import json from pathlib import Path from urllib.parse import urlsplit -from . import configuration, install, manifest +from . import backup_manifest, configuration, install, manifest from .policy import DeliveryError, digest, fields, read_json, sha256 @@ -104,10 +104,12 @@ def compatible(backup: dict, payload: Path, target_data: dict, private_root: Pat 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")) + old_manifest = backup_manifest.named_member( + private_root, backup, "releases/" + selected["release_id"] + "/manifest.json" + ) + old = manifest.validate(read_json(old_manifest)) if ( - digest(old_root / "manifest.json") != backup["manifest_sha256"] + digest(old_manifest) != backup["manifest_sha256"] or old["source_sha"] != backup["source_sha"] or old["images"] != backup["images"] or old["images"] != target_data["images"] @@ -125,7 +127,10 @@ def compatible(backup: dict, payload: Path, target_data: dict, private_root: Pat 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": + if ( + backup_manifest.named_member(private_root, backup, "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. @@ -142,18 +147,28 @@ def identity_check(backup: dict, selected: dict, root: Path) -> dict: } 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 + name: ("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"]: + for relative in prior.values(): + if ( + relative not in inventory + or digest(backup_manifest.member_path(root, backup, inventory[relative])) + != inventory[relative]["sha256"] + ): raise DeliveryError("old protected recovery configuration is missing") - previous, current = snapshot(prior), snapshot(paths) + previous, current = ( + snapshot( + { + name: backup_manifest.member_path(root, backup, inventory[relative]) + for name, relative in prior.items() + } + ), + snapshot(paths), + ) fresh_identities(previous, current) cookie = Path("/var/lib/kf-onlinejudge/rabbitmq/.erlang.cookie") if digest(cookie) == backup["broker_cookie_sha256"]: