feat(recovery): add offline Btrfs base checkpoints and resumable restore

2026-10-13 21:40:54 +03:00
parent 98022dd355
commit ae0d9cf08d
4 changed files with 1025 additions and 0 deletions
@@ -0,0 +1,562 @@
+"""Offline, live-ISO-only Btrfs base checkpoints. Never runs installed executables."""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import ctypes
+from datetime import datetime, timezone
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import uuid
+
+
+class RecoveryError(RuntimeError):
+    pass
+
+
+MARKER_KEYS = {'format', 'install_uuid', 'filesystem', 'root_fs_uuid', 'root_partuuid',
+               'esp_uuid', 'esp_partuuid', 'owned_efi_path', 'root_subvolume'}
+ID_RE = re.compile(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\Z')
+
+
+def require(condition, message):
+    if not condition:
+        raise RecoveryError(message)
+
+
+def valid_uuid(value):
+    return isinstance(value, str) and bool(ID_RE.fullmatch(value.lower()))
+
+
+def validate_marker(value):
+    require(isinstance(value, dict) and set(value) == MARKER_KEYS, 'Invalid storage marker fields')
+    require(type(value['format']) is int and value['format'] == 1, 'Unsupported storage marker format')
+    require(all(valid_uuid(value[k]) for k in ('install_uuid', 'root_fs_uuid', 'root_partuuid', 'esp_partuuid')),
+            'Invalid storage UUID')
+    require(isinstance(value['esp_uuid'], str) and re.fullmatch(r'[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}', value['esp_uuid']),
+            'Invalid FAT filesystem UUID')
+    require(value['filesystem'] == 'btrfs' and value['root_subvolume'] == '@root'
+            and value['owned_efi_path'] == 'EFI/HydraOS', 'Unsupported storage layout')
+    return value
+
+
+def safe_path(base, relative, *, missing=False):
+    """Refuse links/special files at every component; caller owns private mount."""
+    base = Path(base)
+    require(not base.is_symlink() and base.is_dir(), 'Unsafe base directory')
+    parts = Path(relative).parts
+    require(parts and not Path(relative).is_absolute() and all(p not in ('.', '..') for p in parts), 'Unsafe path')
+    path = base
+    for part in parts:
+        path = path / part
+        try:
+            mode = path.lstat().st_mode
+        except FileNotFoundError:
+            require(missing, 'Missing path: ' + str(path))
+            continue
+        require(stat.S_ISDIR(mode) or stat.S_ISREG(mode), 'Link or special file refused: ' + str(path))
+    return path
+
+
+def read_json(path):
+    path = Path(path)
+    info = path.lstat()
+    require(stat.S_ISREG(info.st_mode) and info.st_nlink == 1 and info.st_uid == os.geteuid()
+            and not info.st_mode & 0o022, 'Metadata must be a trusted regular file: ' + str(path))
+    require(info.st_size <= 16 * 1024 * 1024, 'Metadata too large')
+    def unique(pairs):
+        out = {}
+        for key, value in pairs:
+            require(key not in out, 'Duplicate JSON field')
+            out[key] = value
+        return out
+    try:
+        return json.loads(path.read_text(), object_pairs_hook=unique)
+    except (ValueError, UnicodeError) as exc:
+        raise RecoveryError('Malformed metadata: ' + str(path)) from exc
+
+
+def fsync_dir(path):
+    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def durable_json(path, value):
+    path = Path(path)
+    fd, name = tempfile.mkstemp(prefix='.metadata-', dir=path.parent)
+    try:
+        with os.fdopen(fd, 'w') as stream:
+            json.dump(value, stream, sort_keys=True, indent=2)
+            stream.write('\n'); stream.flush(); os.fsync(stream.fileno())
+        os.replace(name, path)
+        fsync_dir(path.parent)
+    finally:
+        if os.path.exists(name): os.unlink(name)
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+
+
+def file_hash(path):
+    with Path(path).open('rb') as stream:
+        return hashlib.file_digest(stream, 'sha256').hexdigest()
+
+
+def tree_manifest(root):
+    """FAT-compatible names, regular files/directories, exact backup metadata."""
+    root = Path(root)
+    require(root.is_dir() and not root.is_symlink(), 'Owned EFI tree must be a directory')
+    entries = []
+    def walk(directory):
+        folded = set()
+        for path in sorted(directory.iterdir()):
+            name = path.name
+            require(re.fullmatch(r'[A-Za-z0-9_+., -]+', name) and name not in ('.', '..')
+                    and not name.endswith((' ', '.')), 'Unsafe EFI name')
+            require(name.casefold() not in folded, 'Ambiguous FAT filename')
+            folded.add(name.casefold())
+            info = path.lstat()
+            require(stat.S_ISDIR(info.st_mode) or stat.S_ISREG(info.st_mode), 'Unsafe EFI file type')
+            require(not stat.S_ISREG(info.st_mode) or info.st_nlink == 1, 'EFI hardlink refused')
+            entry = dict(path=path.relative_to(root).as_posix(), kind='dir' if path.is_dir() else 'file',
+                         mode=stat.S_IMODE(info.st_mode), mtime_ns=info.st_mtime_ns)
+            if path.is_file():
+                entry.update(size=info.st_size, sha256=file_hash(path))
+            entries.append(entry)
+            if path.is_dir(): walk(path)
+    walk(root)
+    return entries
+
+
+def sync_tree(root):
+    for path in Path(root).rglob('*'):
+        if path.is_file():
+            with path.open('rb') as stream: os.fsync(stream.fileno())
+    for directory, _, _ in os.walk(root, topdown=False): fsync_dir(directory)
+
+
+class Btrfs:
+    def run(self, argv):
+        return subprocess.run(argv, check=True, text=True, stdout=subprocess.PIPE,
+                              stderr=subprocess.PIPE, env=dict(os.environ, LC_ALL='C')).stdout
+    def info(self, path):
+        output = self.run(['btrfs', 'subvolume', 'show', str(path)])
+        fields = dict(line.strip().split(':', 1) for line in output.splitlines() if ':' in line)
+        value = fields.get('UUID', '').strip()
+        parent = fields.get('Parent UUID', '').strip()
+        require(valid_uuid(value), 'Not a Btrfs subvolume: ' + str(path))
+        ro = self.run(['btrfs', 'property', 'get', '-ts', str(path), 'ro']).strip()
+        require(ro in ('ro=true', 'ro=false'), 'Cannot establish snapshot readonly state')
+        return dict(uuid=value, parent_uuid=None if parent == '-' else parent, readonly=ro == 'ro=true')
+    def nested(self, path):
+        return self.run(['btrfs', 'subvolume', 'list', '-o', str(path)]).splitlines()
+    def snapshot(self, source, dest, readonly):
+        self.run(['btrfs', 'subvolume', 'snapshot', *(['-r'] if readonly else []), str(source), str(dest)])
+        return self.info(dest)
+    def sync(self, path):
+        fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        try:
+            libc = ctypes.CDLL(None, use_errno=True)
+            if libc.syncfs(fd) != 0: raise OSError(ctypes.get_errno(), 'syncfs failed')
+        finally:
+            os.close(fd)
+
+
+class RecoverySession:
+    """Mounted engine; CLI supplies verified identities and private offline mounts."""
+    def __init__(self, top, esp, backend, identities):
+        self.top, self.esp, self.backend = Path(top), Path(esp), backend
+        self.identities = dict(identities)
+        self.snapshots = safe_path(self.top, '@snapshots')
+        self.journal = self.snapshots / 'restore.json'
+        for name in ('@snapshots', '@home'):
+            path = safe_path(self.top, name)
+            require(not self.backend.info(path)['readonly'], 'Layout subvolume is readonly')
+
+    def marker(self, root):
+        marker = validate_marker(read_json(safe_path(root, 'etc/hydra/storage.json')))
+        for key, expected in self.identities.items():
+            require(marker.get(key) == expected, 'Storage identity mismatch: ' + key)
+        require(not self.backend.nested(root), 'Nested subvolumes under root are unsupported')
+        return marker
+
+    def current_root(self):
+        root = safe_path(self.top, '@root')
+        require(not self.backend.info(root)['readonly'], 'Installed root is readonly')
+        return root, self.marker(root)
+
+    def efi_root(self, *, missing=False):
+        # EFI itself must exist. Only the owned directory may be recreated.
+        safe_path(self.esp, 'EFI')
+        root = safe_path(self.esp, 'EFI/HydraOS', missing=missing)
+        for parent, expected in ((self.esp, 'EFI'), (self.esp / 'EFI', 'HydraOS')):
+            require(not any(p.name.casefold() == expected.casefold() and p.name != expected for p in parent.iterdir()),
+                    'Case alias of owned EFI directory refused')
+            names = [p.name.casefold() for p in parent.iterdir()]
+            require(len(names) == len(set(names)), 'Ambiguous EFI directory names')
+        return root
+
+    def checkpoint(self, name, *, dry_run=False):
+        safe_path(self.snapshots, 'restore.json', missing=True)
+        require(not self.journal.exists(), 'Resume pending restore before checkpointing')
+        require(isinstance(name, str) and 1 <= len(name) <= 100 and all(ord(c) >= 32 for c in name), 'Invalid checkpoint name')
+        root, storage = self.current_root()
+        efi = self.efi_root()
+        manifest = tree_manifest(efi)
+        source = self.backend.info(root)
+        if dry_run:
+            return dict(operation='checkpoint', execute=False, name=name, storage=storage)
+        ident = str(uuid.uuid4())
+        directory = self.snapshots / ident
+        directory.mkdir(mode=0o700)
+        fsync_dir(self.snapshots)
+        snapshot = self.backend.snapshot(root, directory / 'root', True)
+        require(snapshot['readonly'] and snapshot['parent_uuid'] == source['uuid'], 'Invalid new snapshot')
+        shutil.copytree(efi, directory / 'efi', copy_function=shutil.copy2)
+        require(tree_manifest(directory / 'efi') == manifest, 'EFI changed during checkpoint')
+        sync_tree(directory / 'efi')
+        self.backend.sync(self.top)
+        descriptor = dict(format=1, id=ident, name=name, created=datetime.now(timezone.utc).isoformat(),
+                          storage=storage, source_root_uuid=source['uuid'], snapshot_uuid=snapshot['uuid'],
+                          efi_manifest=manifest)
+        durable_json(directory / 'checkpoint.json', descriptor)
+        self.backend.sync(self.top)
+        return descriptor
+
+    def descriptor(self, ident):
+        require(valid_uuid(ident), 'Invalid checkpoint ID')
+        directory = safe_path(self.snapshots, ident)
+        require({p.name for p in directory.iterdir()} == {'root', 'efi', 'checkpoint.json'}, 'Unexpected checkpoint files')
+        value = read_json(safe_path(directory, 'checkpoint.json'))
+        require(isinstance(value, dict) and set(value) == {'format', 'id', 'name', 'created', 'storage',
+                'source_root_uuid', 'snapshot_uuid', 'efi_manifest'}, 'Invalid checkpoint fields')
+        require(type(value['format']) is int and value['format'] == 1 and value['id'] == ident,
+                'Unsupported checkpoint descriptor')
+        require(isinstance(value['name'], str) and isinstance(value['created'], str)
+                and valid_uuid(value['source_root_uuid']) and valid_uuid(value['snapshot_uuid']), 'Invalid checkpoint metadata')
+        validate_marker(value['storage'])
+        snapshot = safe_path(directory, 'root')
+        info = self.backend.info(snapshot)
+        require(info == dict(uuid=value['snapshot_uuid'], parent_uuid=value['source_root_uuid'], readonly=True),
+                'Checkpoint snapshot identity or readonly state changed')
+        require(self.marker(snapshot) == value['storage'], 'Checkpoint storage identity changed')
+        require(tree_manifest(safe_path(directory, 'efi')) == value['efi_manifest'], 'EFI backup metadata/hash mismatch')
+        return value
+
+    def list(self):
+        result = []
+        for directory in sorted(self.snapshots.iterdir()):
+            if valid_uuid(directory.name) and not directory.is_symlink() and (directory / 'checkpoint.json').exists():
+                result.append(self.descriptor(directory.name))
+        return result
+
+    def save_journal(self, value, phase):
+        value['phase'] = phase
+        durable_json(self.journal, value)
+        self.backend.sync(self.top)
+
+    def restore(self, ident, *, dry_run=False):
+        safe_path(self.snapshots, 'restore.json', missing=True)
+        descriptor = self.descriptor(ident)  # ALL backup validation precedes any write.
+        source = self.snapshots / ident / 'root'
+        owned = self.efi_root(missing=True)
+        if owned.exists(): tree_manifest(owned)
+        if self.journal.exists():
+            tx = read_json(self.journal)
+            require(isinstance(tx, dict) and set(tx) == {'format', 'id', 'checkpoint', 'checkpoint_digest',
+                    'storage', 'old_uuid', 'new_uuid', 'prepared_root', 'failed_root', 'phase'}, 'Invalid restore journal')
+            require(type(tx['format']) is int and tx['format'] == 1 and valid_uuid(tx['id'])
+                    and tx['checkpoint'] == ident and tx['checkpoint_digest'] == digest(descriptor)
+                    and tx['storage'] == descriptor['storage'] and valid_uuid(tx['old_uuid'])
+                    and (tx['new_uuid'] is None or valid_uuid(tx['new_uuid']))
+                    and tx['phase'] in ('preparing', 'prepared', 'root-retained', 'root-installed', 'efi-restored')
+                    and tx['prepared_root'] == '@restore-' + tx['id'] and tx['failed_root'] == '@failed-' + tx['id'],
+                    'Restore journal does not match requested checkpoint')
+        else:
+            root, marker = self.current_root()
+            require(marker == descriptor['storage'], 'Current root belongs to a different installation')
+            require(not any(p.name.startswith('@restore-') for p in self.top.iterdir()), 'Unexplained prepared root')
+            txid = str(uuid.uuid4())
+            tx = dict(format=1, id=txid, checkpoint=ident, checkpoint_digest=digest(descriptor),
+                      storage=marker, old_uuid=self.backend.info(root)['uuid'], new_uuid=None,
+                      prepared_root='@restore-' + txid, failed_root='@failed-' + txid, phase='preparing')
+        prepared = safe_path(self.top, tx['prepared_root'], missing=True)
+        failed = safe_path(self.top, tx['failed_root'], missing=True)
+        root = safe_path(self.top, '@root', missing=True)
+        def identity(path):
+            if not path.exists(): return None
+            info = self.backend.info(path)
+            require(not info['readonly'] and self.marker(path) == tx['storage'], 'Unexpected root identity or layout')
+            return info
+        actual_root, actual_prepared, actual_failed = identity(root), identity(prepared), identity(failed)
+        # Enumerate the only three legitimate rename arrangements. No phase-based guessing.
+        if tx['new_uuid'] is None:
+            require(tx['phase'] == 'preparing' and actual_root and actual_root['uuid'] == tx['old_uuid']
+                    and actual_failed is None, 'Unexplained initial restore state')
+            if actual_prepared is None:
+                if dry_run:
+                    return dict(operation='restore', execute=False, checkpoint=ident, storage=descriptor['storage'])
+                if not self.journal.exists(): self.save_journal(tx, 'preparing')
+                actual_prepared = self.backend.snapshot(source, prepared, False)
+                fsync_dir(self.top); self.backend.sync(self.top)
+            require(actual_prepared['parent_uuid'] == descriptor['snapshot_uuid'] and not actual_prepared['readonly'],
+                    'Prepared snapshot source mismatch')
+            tx['new_uuid'] = actual_prepared['uuid']
+            if not dry_run: self.save_journal(tx, 'prepared')
+        new_uuid, old_uuid = tx['new_uuid'], tx['old_uuid']
+        state = (actual_root['uuid'] if actual_root else None, actual_prepared['uuid'] if actual_prepared else None,
+                 actual_failed['uuid'] if actual_failed else None)
+        require(state in ((old_uuid, new_uuid, None), (None, new_uuid, old_uuid), (new_uuid, None, old_uuid)),
+                'Unexplained root rename state; preserve all roots and inspect offline')
+        active_new = actual_prepared if actual_prepared else actual_root
+        require(active_new['parent_uuid'] == descriptor['snapshot_uuid'], 'Prepared root parent UUID mismatch')
+        if dry_run:
+            return dict(operation='restore', execute=False, checkpoint=ident, storage=descriptor['storage'])
+        if state == (old_uuid, new_uuid, None):
+            os.rename(root, failed); fsync_dir(self.top); self.backend.sync(self.top)
+            self.save_journal(tx, 'root-retained')
+            state = (None, new_uuid, old_uuid)
+        if state == (None, new_uuid, old_uuid):
+            os.rename(prepared, root); fsync_dir(self.top); self.backend.sync(self.top)
+            self.save_journal(tx, 'root-installed')
+        self.restore_efi(self.snapshots / ident / 'efi', owned)
+        self.backend.sync(self.esp)
+        self.save_journal(tx, 'efi-restored')
+        # Archive rather than discard the identity record and retained root path.
+        os.replace(self.journal, self.snapshots / ('restored-' + tx['id'] + '.json'))
+        fsync_dir(self.snapshots); self.backend.sync(self.top)
+        return dict(checkpoint=ident, failed_root=tx['failed_root'], restored_root_uuid=new_uuid)
+
+    def restore_efi(self, source, dest):
+        # Backup and existing destination have already passed complete path validation.
+        # Repeating this copy is safe after interruption at any individual FAT write.
+        if dest.exists():
+            for child in list(dest.iterdir()):
+                if child.is_dir(): shutil.rmtree(child)
+                else: child.unlink()
+        else:
+            dest.mkdir()
+        shutil.copytree(source, dest, dirs_exist_ok=True, copy_function=shutil.copy2)
+        sync_tree(dest); fsync_dir(dest.parent)
+        expected = [(e['path'], e.get('sha256')) for e in tree_manifest(source)]
+        actual = [(e['path'], e.get('sha256')) for e in tree_manifest(dest)]
+        require(actual == expected, 'Restored EFI verification failed; rerun the same restore')
+
+
+def flatten(nodes):
+    result = {}
+    def visit(node):
+        require(isinstance(node, dict) and isinstance(node.get('path'), str), 'Malformed device topology')
+        path = node['path']
+        require(re.fullmatch(r'/dev/[A-Za-z0-9_./-]+', path) and '..' not in Path(path).parts, 'Invalid device path')
+        require(path not in result or result[path] == node, 'Conflicting device topology')
+        result[path] = node
+        require(isinstance(node.get('children', []), list), 'Malformed device children')
+        for child in node.get('children', []): visit(child)
+    for node in nodes: visit(node)
+    return result
+
+
+def resolve_partition(nodes, key, value, filesystem):
+    devices = flatten(nodes)
+    matches = [n for n in devices.values() if str(n.get(key, '')).lower() == value.lower()]
+    require(len(matches) == 1, 'Missing or ambiguous ' + key + ': ' + value)
+    selected = matches[0]
+    require(selected.get('type') == 'part' and selected.get('fstype') == filesystem
+            and not selected.get('children'), 'Target must be an ordinary partition of type ' + filesystem)
+    require(isinstance(selected.get('mountpoints'), list) and not any(selected['mountpoints']), 'Target is mounted or active swap')
+    for field in ('uuid', 'partuuid'):
+        identity = selected.get(field)
+        require(isinstance(identity, str) and identity and sum(str(n.get(field, '')).lower() == identity.lower()
+                for n in devices.values()) == 1, 'Missing or duplicate target ' + field)
+    parent = selected.get('pkname', '')
+    if not parent.startswith('/dev/'): parent = '/dev/' + parent
+    require(parent in devices and devices[parent].get('type') == 'disk', 'Target parent must be an ordinary disk')
+    require(not any(devices[parent].get('mountpoints', [])), 'Parent disk is mounted')
+    require(re.fullmatch(r'\d+:\d+', selected.get('maj:min', '')), 'Missing target block identity')
+    return selected
+
+
+def validate_live(base=Path('/'), cmdline=None, mountinfo=None):
+    base = Path(base)
+    path = safe_path(base, 'etc/hydra/live-mode/active')
+    info = path.lstat()
+    require(stat.S_ISREG(info.st_mode) and info.st_uid == os.geteuid() and info.st_nlink == 1
+            and not info.st_mode & 0o022, 'Untrusted live marker')
+    for directory in (path.parent, path.parent.parent, path.parent.parent.parent):
+        info = directory.stat()
+        require(info.st_uid == os.geteuid() and not info.st_mode & 0o022, 'Untrusted live marker directory')
+    if cmdline is None: cmdline = Path('/proc/cmdline').read_text()
+    if mountinfo is None: mountinfo = Path('/proc/self/mountinfo').read_text()
+    words = cmdline.split()
+    require('rd.live.image' in words and any(w.startswith('root=live:') for w in words), 'Recovery requires a Hydra live boot')
+    mounts = [line.split() for line in mountinfo.splitlines()]
+    require(any(len(parts) > 8 and parts[4] == '/' and '-' in parts
+                and parts[parts.index('-') + 1] == 'overlay' for parts in mounts)
+            and any(' - squashfs ' in line for line in mountinfo.splitlines()), 'Live root must use overlay and squashfs')
+
+
+class DeviceProbe:
+    def __init__(self, backend, *, proc=Path('/proc'), sysroot=Path('/sys')):
+        self.backend, self.proc, self.sysroot = backend, Path(proc), Path(sysroot)
+    def nodes(self):
+        data = json.loads(self.backend.run(['lsblk', '--json', '--paths', '--output',
+                                          'PATH,TYPE,PKNAME,MAJ:MIN,FSTYPE,UUID,PARTUUID,MOUNTPOINTS']))
+        require(isinstance(data, dict) and isinstance(data.get('blockdevices'), list), 'Malformed lsblk result')
+        nodes = data['blockdevices']
+        # Probe signatures directly; do not trust stale udev UUID caches.
+        for node in flatten(nodes).values():
+            result = subprocess.run(['blkid', '-p', '-o', 'export', node['path']], text=True,
+                                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            require(result.returncode in (0, 2), 'Ambiguous or unreadable device signatures')
+            fields = dict(line.split('=', 1) for line in result.stdout.splitlines() if '=' in line)
+            node.update(uuid=fields.get('UUID'), partuuid=fields.get('PART_ENTRY_UUID'), fstype=fields.get('TYPE'))
+        return nodes
+    def check(self, node):
+        path = Path(node['path'])
+        info = path.stat()
+        require(stat.S_ISBLK(info.st_mode) and f'{os.major(info.st_rdev)}:{os.minor(info.st_rdev)}' == node['maj:min'],
+                'Device identity changed')
+        sysnode = self.sysroot / 'dev/block' / node['maj:min']
+        require((sysnode / 'partition').is_file() and not any((sysnode / 'holders').iterdir()), 'Target has holders or is not a partition')
+        parent = sysnode.resolve().parent
+        require(not any((parent / 'holders').iterdir()), 'Parent disk has active holders')
+        for line in (self.proc / 'self/mountinfo').read_text().splitlines():
+            fields = line.split(); separator = fields.index('-')
+            source = fields[separator + 2]
+            require(fields[2] != node['maj:min'], 'Target is mounted')
+            if source.startswith('/dev/'):
+                try: mounted = Path(source).stat()
+                except FileNotFoundError: raise RecoveryError('Cannot resolve mounted block source')
+                require(mounted.st_rdev != info.st_rdev, 'Target backs a mounted filesystem')
+        for line in (self.proc / 'swaps').read_text().splitlines()[1:]:
+            source = Path(line.split()[0])
+            swap = source.stat()
+            require((swap.st_rdev if stat.S_ISBLK(swap.st_mode) else swap.st_dev) != info.st_rdev,
+                    'Target has active swap')
+
+
+@contextlib.contextmanager
+def operation_lock():
+    fd = os.open('/run/hydra-system.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+    try:
+        info = os.fstat(fd)
+        require(stat.S_ISREG(info.st_mode) and info.st_uid == 0 and info.st_nlink == 1 and not info.st_mode & 0o077,
+                'Untrusted operation lock')
+        try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError: raise RecoveryError('Another hydra-system operation is running')
+        yield
+    finally: os.close(fd)
+
+
+def private_namespace(backend):
+    libc = ctypes.CDLL(None, use_errno=True)
+    if libc.unshare(0x00020000) != 0: raise OSError(ctypes.get_errno(), 'Cannot create private mount namespace')
+    backend.run(['mount', '--make-rprivate', '/'])
+
+
+@contextlib.contextmanager
+def mounted(backend, device, filesystem, writable=False):
+    directory = Path(tempfile.mkdtemp(prefix='hydra-system-', dir='/run'))
+    did_mount = False
+    try:
+        options = ['rw' if writable else 'ro', 'nosuid', 'nodev', 'noexec']
+        if filesystem == 'btrfs':
+            options.append('subvolid=5')
+            if not writable: options.append('nologreplay')
+        backend.run(['mount', '-t', filesystem, '-o', ','.join(options), device, str(directory)])
+        did_mount = True
+        if filesystem == 'btrfs':
+            show = backend.run(['btrfs', 'filesystem', 'show', '--raw', str(directory)])
+            require(re.search(r'Total devices\s+1\b', show) and len(re.findall(r'\bdevid\s+\d+', show)) == 1,
+                    'Recovery requires a single-device Btrfs filesystem')
+        yield directory
+    finally:
+        if did_mount:
+            # A failed unmount deliberately leaves the directory intact; never recurse into a mount.
+            backend.run(['umount', str(directory)])
+        directory.rmdir()
+
+
+def mounted_storage(top):
+    if (top / '@root').exists():
+        return validate_marker(read_json(safe_path(top, '@root/etc/hydra/storage.json')))
+    journal = read_json(safe_path(top, '@snapshots/restore.json'))
+    require(isinstance(journal, dict) and isinstance(journal.get('storage'), dict), 'Missing root without valid recovery journal')
+    return validate_marker(journal['storage'])
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest='command', required=True)
+    for name in ('checkpoint', 'list', 'restore'):
+        command = commands.add_parser(name)
+        command.add_argument('--root-uuid', required=True)
+        if name != 'list':
+            command.add_argument('--execute', action='store_true')
+            command.add_argument('--confirm-root-uuid')
+        if name == 'checkpoint':
+            command.add_argument('--esp-partuuid', required=True)
+            command.add_argument('--name', required=True)
+        if name == 'restore': command.add_argument('--checkpoint', required=True)
+    args = parser.parse_args(argv)
+    try:
+        require(valid_uuid(args.root_uuid), 'Invalid root filesystem UUID')
+        execute = getattr(args, 'execute', False)
+        if execute:
+            require(args.confirm_root_uuid == args.root_uuid, 'Execution requires --confirm-root-uuid matching --root-uuid')
+        require(os.geteuid() == 0, 'Run from the Hydra live ISO as root; no automatic sudo')
+        validate_live()
+        backend = Btrfs(); probe = DeviceProbe(backend)
+        with operation_lock():
+            nodes = probe.nodes()
+            root = resolve_partition(nodes, 'uuid', args.root_uuid, 'btrfs'); probe.check(root)
+            private_namespace(backend)
+            # First inspect read-only, without log replay, before opening either filesystem writable.
+            with mounted(backend, root['path'], 'btrfs') as top:
+                marker = mounted_storage(top)
+                require(marker['root_fs_uuid'] == args.root_uuid and marker['root_partuuid'] == root['partuuid'], 'Root ownership mismatch')
+                if args.command == 'checkpoint': require(args.esp_partuuid == marker['esp_partuuid'], 'ESP differs from installation record')
+                esp = resolve_partition(nodes, 'partuuid', marker['esp_partuuid'], 'vfat'); probe.check(esp)
+                require(esp['uuid'] == marker['esp_uuid'], 'ESP filesystem identity mismatch')
+                with mounted(backend, esp['path'], 'vfat') as efi:
+                    session = RecoverySession(top, efi, backend, marker)
+                    if args.command == 'list': result = session.list()
+                    elif args.command == 'restore':
+                        result = session.restore(args.checkpoint, dry_run=True)
+                    else:
+                        result = session.checkpoint(args.name, dry_run=True)
+            if execute:
+                # Re-resolve and check after releasing inspection mounts, immediately before writes.
+                nodes = probe.nodes()
+                current_root = resolve_partition(nodes, 'uuid', args.root_uuid, 'btrfs')
+                current_esp = resolve_partition(nodes, 'partuuid', marker['esp_partuuid'], 'vfat')
+                for before, after in ((root, current_root), (esp, current_esp)):
+                    require(all(before.get(k) == after.get(k) for k in ('path', 'uuid', 'partuuid', 'maj:min', 'fstype')), 'Target changed during planning')
+                    probe.check(after)
+                with mounted(backend, root['path'], 'btrfs', True) as top:
+                    with mounted(backend, esp['path'], 'vfat', True) as efi:
+                        session = RecoverySession(top, efi, backend, marker)
+                        result = session.checkpoint(args.name) if args.command == 'checkpoint' else session.restore(args.checkpoint)
+            print(json.dumps(result, indent=2, sort_keys=True))
+        return 0
+    except (RecoveryError, OSError, ValueError, subprocess.CalledProcessError) as exc:
+        print('hydra-system: ' + str(exc), file=sys.stderr)
+        return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())