335 lines
18 KiB
Python
335 lines
18 KiB
Python
"""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_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,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()
|