"""Offline recovery tests; fake Btrfs only, never operate on block devices.""" import copy import io import stat from types import SimpleNamespace import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest import uuid from unittest.mock import patch try: from hydra import system_recovery as r except ImportError: r = None ROOT_UUID = '11111111-2222-3333-4444-555555555555' ROOT_PART = '21111111-2222-3333-4444-555555555555' ESP_PART = '31111111-2222-3333-4444-555555555555' MARKER = dict(format=1, install_uuid='41111111-2222-3333-4444-555555555555', filesystem='btrfs', root_fs_uuid=ROOT_UUID, root_partuuid=ROOT_PART, esp_uuid='1234-ABCD', esp_partuuid=ESP_PART, owned_efi_path='EFI/HydraOS', root_subvolume='@root') class FakeBtrfs: """Persist identity inside fake subvolumes so a fresh process can recover.""" def info(self, path): return json.loads((Path(path) / '.fake-subvolume.json').read_text()) def create(self, path, parent=None, readonly=False): Path(path).mkdir(parents=True, exist_ok=True) info = dict(uuid=str(uuid.uuid4()), parent_uuid=parent, readonly=readonly) (Path(path) / '.fake-subvolume.json').write_text(json.dumps(info)) return info def snapshot(self, source, dest, readonly): shutil.copytree(source, dest) return self.create(dest, self.info(source)['uuid'], readonly) def nested(self, path): return [str(p) for p in Path(path).glob('**/.fake-subvolume.json') if p.parent != Path(path)] def sync(self, path): pass class RecoveryTests(unittest.TestCase): def setUp(self): self.assertIsNotNone(r, 'offline recovery module must exist') self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.base = Path(self.temp.name) self.top, self.esp = self.base / 'top', self.base / 'esp' self.top.mkdir(); self.esp.mkdir() self.b = FakeBtrfs() for name in ('@root', '@home', '@snapshots'): self.b.create(self.top / name) marker = self.top / '@root/etc/hydra/storage.json' marker.parent.mkdir(parents=True) marker.write_text(json.dumps(MARKER)) (self.top / '@root/kernel').write_text('old kernel') (self.top / '@home/user-data').write_text('user data') (self.esp / 'EFI/HydraOS').mkdir(parents=True) (self.esp / 'EFI/HydraOS/grubx64.efi').write_bytes(b'old EFI') (self.esp / 'EFI/OtherOS').mkdir() (self.esp / 'EFI/OtherOS/sentinel').write_bytes(b'foreign') self.session = r.RecoverySession(self.top, self.esp, self.b, MARKER) def checkpoint(self): return self.session.checkpoint('before-update') def test_exact_marker(self): self.assertEqual(r.validate_marker(MARKER), MARKER) for change in ({'format': True}, {'filesystem': 'ext4'}, {'root_subvolume': '@other'}, {'owned_efi_path': '../EFI'}, {'unexpected': 'field'}, {'root_partuuid': 'bad'}): with self.subTest(change=change), self.assertRaises(r.RecoveryError): r.validate_marker(dict(MARKER, **change)) def test_checkpoint_completed_and_readonly(self): result = self.checkpoint() self.assertTrue(self.b.info(self.top / '@snapshots' / result['id'] / 'root')['readonly']) self.assertEqual(self.session.list(), [result]) self.assertEqual(result['storage'], MARKER) def test_snapshot_failure_never_publishes_descriptor(self): with patch.object(self.b, 'snapshot', side_effect=OSError('interrupted')): with self.assertRaises(OSError): self.checkpoint() self.assertEqual(self.session.list(), []) def test_nested_subvolume_refused(self): self.b.create(self.top / '@root/nested') with self.assertRaises(r.RecoveryError): self.checkpoint() def test_restore_preserves_home_foreign_and_failed_root(self): cp = self.checkpoint() old_uuid = self.b.info(self.top / '@root')['uuid'] (self.top / '@root/kernel').write_text('broken kernel') (self.top / '@home/user-data').write_text('new home') (self.esp / 'EFI/HydraOS/grubx64.efi').write_bytes(b'broken') (self.esp / 'EFI/HydraOS/new-file').write_text('remove') result = self.session.restore(cp['id']) self.assertEqual((self.top / '@root/kernel').read_text(), 'old kernel') self.assertEqual((self.top / '@home/user-data').read_text(), 'new home') self.assertEqual((self.esp / 'EFI/OtherOS/sentinel').read_bytes(), b'foreign') self.assertEqual((self.esp / 'EFI/HydraOS/grubx64.efi').read_bytes(), b'old EFI') self.assertFalse((self.esp / 'EFI/HydraOS/new-file').exists()) self.assertEqual(self.b.info(self.top / result['failed_root'])['uuid'], old_uuid) def test_tampered_backup_rejected_before_root_mutation(self): cp = self.checkpoint() original = self.b.info(self.top / '@root') (self.top / '@snapshots' / cp['id'] / 'efi/grubx64.efi').write_bytes(b'bad') with self.assertRaises(r.RecoveryError): self.session.restore(cp['id']) self.assertEqual(self.b.info(self.top / '@root'), original) self.assertFalse((self.top / '@snapshots/restore.json').exists()) def test_symlink_efi_and_metadata_refused(self): (self.esp / 'EFI/HydraOS/link').symlink_to('/etc/passwd') with self.assertRaises(r.RecoveryError): self.checkpoint() def test_rw_snapshot_refused(self): cp = self.checkpoint() path = self.top / '@snapshots' / cp['id'] / 'root/.fake-subvolume.json' data = json.loads(path.read_text()); data['readonly'] = False; path.write_text(json.dumps(data)) with self.assertRaises(r.RecoveryError): self.session.restore(cp['id']) def test_wrong_installed_identity_refused(self): marker = self.top / '@root/etc/hydra/storage.json' marker.write_text(json.dumps(dict(MARKER, esp_partuuid=ROOT_PART))) with self.assertRaises(r.RecoveryError): self.checkpoint() def test_fresh_process_resume_each_transition(self): cp = self.checkpoint() for phase in ('prepared', 'root-retained', 'root-installed', 'efi-restored'): with self.subTest(phase=phase): (self.top / '@root/kernel').write_text('broken again') script = '''import json, os, sys from pathlib import Path from hydra import system_recovery as r sys.path.insert(0, 'tests') from test_system_recovery import FakeBtrfs, MARKER s = r.RecoverySession(Path(sys.argv[1]), Path(sys.argv[2]), FakeBtrfs(), MARKER) save = s.save_journal def crash(value, phase): save(value, phase) if phase == sys.argv[4]: os._exit(77) s.save_journal = crash s.restore(sys.argv[3]) ''' result = subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], phase]) self.assertEqual(result.returncode, 77) subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], 'none'], check=True) self.assertEqual((self.top / '@root/kernel').read_text(), 'old kernel') class GuardTests(unittest.TestCase): def devices(self): return [dict(path='/dev/vda', type='disk', **{'maj:min': '8:0'}, mountpoints=[], children=[ dict(path='/dev/vda1', type='part', fstype='vfat', uuid='1234-ABCD', partuuid=ESP_PART, pkname='/dev/vda', **{'maj:min': '8:1'}, mountpoints=[], children=[]), dict(path='/dev/vda2', type='part', fstype='btrfs', uuid=ROOT_UUID, partuuid=ROOT_PART, pkname='/dev/vda', **{'maj:min': '8:2'}, mountpoints=[], children=[])])] def test_unique_ordinary_unmounted_partition(self): nodes = self.devices() self.assertEqual(r.resolve_partition(nodes, 'uuid', ROOT_UUID, 'btrfs')['path'], '/dev/vda2') for change in ('duplicate', 'mounted', 'holder', 'not-partition', 'same-partuuid'): bad = copy.deepcopy(nodes) root = bad[0]['children'][1] if change == 'duplicate': extra = dict(root, path='/dev/vdb2', **{'maj:min': '8:18'}) bad.append(extra) elif change == 'mounted': root['mountpoints'] = ['/mnt/target'] elif change == 'holder': root['children'] = [dict(path='/dev/dm-0', type='crypt')] elif change == 'not-partition': root['type'] = 'disk' else: bad[0]['children'][0]['partuuid'] = ROOT_PART with self.subTest(change=change), self.assertRaises(r.RecoveryError): r.resolve_partition(bad, 'uuid', ROOT_UUID, 'btrfs') def test_live_marker_must_be_regular_trusted_and_live_boot(self): with tempfile.TemporaryDirectory() as temp: base = Path(temp); marker = base / 'etc/hydra/live-mode/active' marker.parent.mkdir(parents=True); marker.write_text('') cmdline = 'root=live:CDLABEL=HYDRA_OS rd.live.image rd.live.overlay.overlayfs=1' mountinfo = '1 0 0:1 / / rw - overlay overlay rw\n2 1 7:0 / /run/rootfsbase ro - squashfs /dev/loop0 ro\n' r.validate_live(base, cmdline, mountinfo) for command, mounts in (('', mountinfo), (cmdline, '')): with self.assertRaises(r.RecoveryError): r.validate_live(base, command, mounts) marker.unlink(); marker.symlink_to('/etc/passwd') with self.assertRaises(r.RecoveryError): r.validate_live(base, cmdline, mountinfo) def test_untrusted_journal_link_refused_before_write(self): case = RecoveryTests('test_exact_marker'); case.setUp() try: cp = case.checkpoint() (case.top / '@snapshots/restore.json').symlink_to(case.base / 'absent') with self.assertRaises(r.RecoveryError): case.session.restore(cp['id']) self.assertFalse((case.base / 'absent').exists()) finally: case.doCleanups() class HardeningTests(unittest.TestCase): setUp = RecoveryTests.setUp checkpoint = RecoveryTests.checkpoint def test_checkpoint_plan_validates_pending_transaction_without_writes(self): before = sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*')) self.session.checkpoint('planned', dry_run=True) self.assertEqual(before, sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*'))) (self.top / '@snapshots/restore.json').write_text('{}') with self.assertRaises(r.RecoveryError): self.session.checkpoint('planned', dry_run=True) def test_dry_restore_verifies_destination_and_writes_nothing(self): cp = self.checkpoint() before = sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*')) self.session.restore(cp['id'], dry_run=True) self.assertEqual(before, sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*'))) (self.esp / 'EFI/HydraOS/escape').symlink_to(self.base / 'foreign') with self.assertRaises(r.RecoveryError): self.session.restore(cp['id'], dry_run=True) def test_case_alias_of_owned_directory_refused(self): cp = self.checkpoint() (self.esp / 'EFI/HydraOS').rename(self.esp / 'EFI/hydraos') with self.assertRaises(r.RecoveryError): self.session.restore(cp['id']) self.assertFalse((self.top / '@snapshots/restore.json').exists()) def test_unexpected_checkpoint_file_refused(self): cp = self.checkpoint() (self.top / '@snapshots' / cp['id'] / 'unexpected').write_text('bad') with self.assertRaises(r.RecoveryError): self.session.restore(cp['id']) def test_fresh_process_death_between_mutation_and_journal(self): cp = self.checkpoint() for point in ('preparing', 'snapshot', 'first-rename', 'second-rename', 'efi-copy'): script = '''import os, sys from pathlib import Path from hydra import system_recovery as r sys.path.insert(0, 'tests') from test_system_recovery import FakeBtrfs, MARKER b = FakeBtrfs() s = r.RecoverySession(Path(sys.argv[1]), Path(sys.argv[2]), b, MARKER) point = sys.argv[4] save = s.save_journal def crash(value, phase): save(value, phase) if phase == point: os._exit(77) s.save_journal = crash snapshot = b.snapshot def interrupted_snapshot(*args): result = snapshot(*args) if point == 'snapshot': os._exit(77) return result b.snapshot = interrupted_snapshot rename = r.os.rename def interrupted_rename(source, dest): rename(source, dest) if (point == 'first-rename' and Path(source).name == '@root') or (point == 'second-rename' and Path(dest).name == '@root'): os._exit(77) r.os.rename = interrupted_rename copytree = r.shutil.copytree def interrupted_copytree(*args, **kwargs): if point == 'efi-copy' and Path(args[0]).name == 'efi': os._exit(77) return copytree(*args, **kwargs) if point == 'efi-copy': r.shutil.copytree = interrupted_copytree s.restore(sys.argv[3]) ''' with self.subTest(point=point): result = subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], point]) self.assertEqual(result.returncode, 77) # A different checkpoint must never take over this transaction. with self.assertRaises(r.RecoveryError): self.session.restore(str(uuid.uuid4())) subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], 'none'], check=True) self.assertEqual((self.top / '@root/kernel').read_text(), 'old kernel') class NativeBoundaryTests(unittest.TestCase): def test_failed_native_command_includes_bounded_stderr(self): backend = r.Btrfs() error = subprocess.CalledProcessError(32, ['mount'], stderr='specific mount failure\n' + 'x' * 5000) with patch.object(r.subprocess, 'run', side_effect=error): with self.assertRaisesRegex(r.RecoveryError, 'specific mount failure') as caught: backend.run(['mount', '-t', 'btrfs', '/dev/example', '/run/example']) self.assertLess(len(str(caught.exception)), 2300) def test_btrfs_show_parser_and_readonly_property(self): backend = r.Btrfs() sample = '''@snapshots/example/root Name: root UUID: 51111111-2222-3333-4444-555555555555 Parent UUID: 61111111-2222-3333-4444-555555555555 Received UUID: - Creation time: 2026-10-13 10:15:03 +0000 Subvolume ID: 259 Flags: readonly Snapshot(s): ''' with patch.object(backend, 'run', side_effect=[sample, 'ro=true\n']): self.assertEqual(backend.info('/mnt/snapshot'), dict(uuid='51111111-2222-3333-4444-555555555555', parent_uuid='61111111-2222-3333-4444-555555555555', readonly=True)) def test_failed_mount_never_unmounts_unowned_path(self): backend = r.Btrfs() calls = [] def fail(argv): calls.append(argv) raise r.RecoveryError('mount failed') with tempfile.TemporaryDirectory() as temp: own = Path(temp) / 'own'; own.mkdir() with patch.object(r.tempfile, 'mkdtemp', return_value=str(own)), patch.object(backend, 'run', side_effect=fail): with self.assertRaises(r.RecoveryError): with r.mounted(backend, '/dev/fake', 'btrfs'): pass self.assertEqual([argv[0] for argv in calls], ['mount']) self.assertFalse(own.exists()) def test_multidevice_mount_is_rejected_and_owned_mount_cleaned(self): backend = r.Btrfs() calls = [] def run(argv): calls.append(argv) return 'Total devices 2\n devid 1 size 100\n devid 2 size 100' if argv[0] == 'btrfs' else '' with tempfile.TemporaryDirectory() as temp: own = Path(temp) / 'own'; own.mkdir() with patch.object(r.tempfile, 'mkdtemp', return_value=str(own)), patch.object(backend, 'run', side_effect=run): with self.assertRaises(r.RecoveryError): with r.mounted(backend, '/dev/fake', 'btrfs'): pass self.assertEqual([argv[0] for argv in calls], ['mount', 'btrfs', 'umount']) self.assertIn('ro,nosuid,nodev,noexec,subvolid=5,rescue=nologreplay', calls[0]) def test_cli_requires_matching_confirmation_before_probe(self): with patch.object(r, 'DeviceProbe') as probe, patch.object(sys, 'stderr', new_callable=io.StringIO): self.assertEqual(r.main(['restore', '--root-uuid', ROOT_UUID, '--checkpoint', ROOT_PART, '--execute']), 1) probe.assert_not_called() class DeviceStateTests(unittest.TestCase): def test_fake_proc_sys_mount_swap_and_holders_fail_closed(self): with tempfile.TemporaryDirectory() as temp: base = Path(temp) proc, sysroot = base / 'proc', base / 'sys' proc.mkdir(); (proc / 'self').mkdir() (proc / 'self/mountinfo').write_text('') (proc / 'swaps').write_text('Filename Type Size Used Priority\n') physical = sysroot / 'devices/disk/part' physical.mkdir(parents=True) (physical / 'partition').write_text('2') (physical / 'holders').mkdir(); (physical.parent / 'holders').mkdir() (sysroot / 'dev/block').mkdir(parents=True) (sysroot / 'dev/block/8:2').symlink_to(physical) node = GuardTests().devices()[0]['children'][1] original_stat = Path.stat def device_stat(path, *args, **kwargs): if str(path) in ('/dev/vda2', '/dev/alias'): return SimpleNamespace(st_mode=stat.S_IFBLK, st_rdev=os.makedev(8, 2)) return original_stat(path, *args, **kwargs) with patch.object(Path, 'stat', device_stat): probe = r.DeviceProbe(r.Btrfs(), proc=proc, sysroot=sysroot) probe.check(node) (proc / 'self/mountinfo').write_text('1 0 0:99 / /mnt rw - btrfs /dev/alias rw\n') with self.assertRaises(r.RecoveryError): probe.check(node) (proc / 'self/mountinfo').write_text('') (proc / 'swaps').write_text('Filename Type Size Used Priority\n/dev/vda2 partition 100 0 -2\n') with self.assertRaises(r.RecoveryError): probe.check(node) (proc / 'swaps').write_text('Filename Type Size Used Priority\n') (physical / 'holders/dm-0').touch() with self.assertRaises(r.RecoveryError): probe.check(node) if __name__ == '__main__': unittest.main()