560 lines
28 KiB
Python
560 lines
28 KiB
Python
"""Installer tests never invoke a real partition, mount or bootloader command."""
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import pty
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
try:
|
|
from hydra import installer as i
|
|
except ImportError:
|
|
i = None
|
|
|
|
|
|
def disk(path='/dev/vda', children=None, mounts=None, kind='disk', pkname=None):
|
|
return {'path': path, 'type': kind, 'pkname': pkname, 'mountpoints': [] if mounts is None else mounts,
|
|
'size': 32 * 1024 ** 3, 'maj:min': '252:' + str(sum((n + 1) * ord(c) for n, c in enumerate(path))), 'children': children or []}
|
|
|
|
|
|
def topology(*nodes):
|
|
return i.Topology.from_json(json.dumps({'blockdevices': list(nodes)}))
|
|
|
|
|
|
class FakeProbe:
|
|
def __init__(self, topo=None):
|
|
self.topo = topo or topology(disk())
|
|
self.calls = 0
|
|
|
|
def snapshot(self, payload):
|
|
self.calls += 1
|
|
return self.topo
|
|
|
|
def resolve(self, path):
|
|
return {'/dev/disk/by-id/test': '/dev/vda'}.get(path, path)
|
|
|
|
def physical_disk(self, path):
|
|
return path == '/dev/vda'
|
|
|
|
|
|
class FakeRunner:
|
|
def __init__(self, root, fail=None):
|
|
self.root, self.fail, self.calls, self.inputs = root, fail, [], []
|
|
|
|
def run(self, argv, *, input=None, sensitive=False):
|
|
self.calls.append(list(argv))
|
|
self.inputs.append((input, sensitive))
|
|
if self.fail == argv[0] or (argv[0] == 'chroot' and self.fail == argv[2]) or (self.fail == 'unsquashfs-extract' and argv[0] == 'unsquashfs' and '-d' in argv):
|
|
raise i.InstallError('command failed: ' + argv[0])
|
|
if argv[0] == 'unsquashfs' and '-d' in argv:
|
|
for directory in ('etc/hydra/live-mode', 'etc/systemd/system/getty@tty1.service.d',
|
|
'etc/sudoers.d', 'usr/share/zoneinfo/Europe', 'etc/locale.gen', 'lib/modules/6.12.1', 'boot'):
|
|
if directory != 'etc/locale.gen':
|
|
(self.root / directory).mkdir(parents=True, exist_ok=True)
|
|
(self.root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\n')
|
|
(self.root / 'usr/share/zoneinfo/Europe/Moscow').touch()
|
|
(self.root / 'etc/locale.gen').write_text('# en_US.UTF-8 UTF-8\n')
|
|
(self.root / 'etc/hydra/live-mode/active').touch()
|
|
(self.root / 'etc/hydra/live-mode/remove-on-install.json').write_text(json.dumps([
|
|
'/etc/systemd/system/getty@tty1.service.d/90-hydra-live.conf',
|
|
'/etc/sudoers.d/90-hydra-live']))
|
|
(self.root / 'etc/systemd/system/getty@tty1.service.d/90-hydra-live.conf').touch()
|
|
(self.root / 'etc/sudoers.d/90-hydra-live').touch()
|
|
(self.root / 'boot/vmlinuz-6.12.1').touch()
|
|
if argv[0] == 'blkid':
|
|
return '11111111-2222-3333-4444-555555555555\n'
|
|
return ''
|
|
|
|
|
|
class GuardTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertIsNotNone(i, 'installer module must implement the guarded installer')
|
|
|
|
def request(self, **kw):
|
|
return i.Request(**({'disk': '/dev/vda', 'user': 'emil'} | kw))
|
|
|
|
def test_confirmation_missing_and_mismatched(self):
|
|
for confirmation in (None, '/dev/vdb', '/dev/disk/by-id/test'):
|
|
with self.subTest(confirmation=confirmation), self.assertRaises(i.InstallError):
|
|
i.validate_request(self.request(execute=True, confirm_erase=confirmation), FakeProbe())
|
|
self.assertEqual(i.validate_request(self.request(execute=True, confirm_erase='/dev/vda'), FakeProbe()), '/dev/vda')
|
|
|
|
def test_alias_resolved_but_confirmation_must_name_resolved_target(self):
|
|
self.assertEqual(i.validate_request(self.request(disk='/dev/disk/by-id/test', execute=True,
|
|
confirm_erase='/dev/vda'), FakeProbe()), '/dev/vda')
|
|
|
|
def test_disk_mounted_child_root_lvm_and_live_source_rejected(self):
|
|
examples = [
|
|
topology(disk(mounts=['/mnt'])),
|
|
topology(disk(children=[disk('/dev/vda1', kind='part', mounts=['/'])])),
|
|
topology(disk(children=[disk('/dev/vda2', kind='part', children=[disk('/dev/mapper/root', kind='lvm', mounts=['/'])])])),
|
|
]
|
|
live = topology(disk(children=[disk('/dev/vda1', kind='part')]))
|
|
live.protect('/dev/vda1')
|
|
examples.append(live)
|
|
for topo in examples:
|
|
with self.subTest(topo=topo), self.assertRaises(i.InstallError):
|
|
i.validate_request(self.request(), FakeProbe(topo))
|
|
|
|
def test_loop_backing_mount_protects_parent_disk(self):
|
|
topo = topology(disk(children=[disk('/dev/vda1', kind='part')]), disk('/dev/loop0', kind='loop', mounts=['/run/live']))
|
|
topo.link('/dev/loop0', '/dev/vda1')
|
|
with self.assertRaises(i.InstallError):
|
|
i.validate_request(self.request(), FakeProbe(topo))
|
|
|
|
def test_malformed_topology_fails_closed(self):
|
|
bad = ['{}', '{', '{"blockdevices":null}', '{"blockdevices":[]}',
|
|
json.dumps({'blockdevices': [disk(kind='mystery')]}),
|
|
json.dumps({'blockdevices': [disk(mounts='')] }),
|
|
json.dumps({'blockdevices': [disk(pkname='/dev/missing')]}),
|
|
json.dumps({'blockdevices': [disk(children=[disk('/dev/vda', kind='part')])]}),
|
|
json.dumps({'blockdevices': [disk('/dev/vda;reboot')]}),
|
|
json.dumps({'blockdevices': [disk(children='bad')]} )]
|
|
for data in bad:
|
|
with self.subTest(data=data), self.assertRaises(i.InstallError):
|
|
i.Topology.from_json(data)
|
|
|
|
def test_nonphysical_unknown_or_partition_target_rejected(self):
|
|
for path in ('/dev/vdb', '/tmp/fake', '/dev/vda1', '/dev/loop0', '/dev/vda;touch /tmp/x'):
|
|
with self.subTest(path=path), self.assertRaises(i.InstallError):
|
|
i.validate_request(self.request(disk=path), FakeProbe())
|
|
|
|
def test_username_timezone_and_payload_injection_rejected(self):
|
|
for username in ('root', '-u', 'a;reboot', 'a\nb', '../x', 'A', ''):
|
|
with self.subTest(username=username), self.assertRaises(i.InstallError):
|
|
i.validate_request(self.request(user=username), FakeProbe())
|
|
for timezone in ('../../etc/passwd', '/etc/passwd', 'a;reboot'):
|
|
with self.subTest(timezone=timezone), self.assertRaises(i.InstallError):
|
|
i.validate_request(self.request(timezone=timezone), FakeProbe())
|
|
for payload in ('-evil', '/tmp/a\nb'):
|
|
with self.subTest(payload=payload), self.assertRaises(i.InstallError):
|
|
i.validate_request(self.request(payload=payload), FakeProbe())
|
|
|
|
def test_plan_is_nonsecret_and_has_exact_partition_boot_commands(self):
|
|
plan = i.render_plan(self.request(), '/dev/vda', '/mnt/hydra-target')
|
|
self.assertIn('1MiB', plan)
|
|
self.assertIn('1025MiB', plan)
|
|
self.assertIn('--removable --no-nvram', plan)
|
|
self.assertIn('/run/initramfs/live/LiveOS/squashfs.img', plan)
|
|
self.assertIn('dry-run', plan)
|
|
self.assertNotIn('chpasswd:', plan)
|
|
|
|
|
|
class PipelineTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertIsNotNone(i, 'installer module must implement the guarded installer')
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name) / 'target'
|
|
self.root.mkdir()
|
|
self.probe = FakeProbe()
|
|
self.log = io.StringIO()
|
|
self.request = i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
|
|
|
|
def install(self, runner, password='distinctive-private-password'):
|
|
i.install(self.request, runner, self.probe, self.root, password, self.log)
|
|
|
|
def test_full_fake_pipeline_sets_installed_defaults_and_cleans_live_files(self):
|
|
runner = FakeRunner(self.root)
|
|
self.install(runner)
|
|
self.assertIn('UUID=', (self.root / 'etc/fstab').read_text())
|
|
self.assertIn('ru_RU.UTF-8 UTF-8', (self.root / 'etc/locale.gen').read_text())
|
|
self.assertFalse((self.root / 'etc/hydra/live-mode').exists())
|
|
self.assertFalse((self.root / 'etc/sudoers.d/90-hydra-live').exists())
|
|
self.assertIn(['chroot', str(self.root), 'grub-install', '--target=x86_64-efi', '--efi-directory=/boot/efi', '--bootloader-id=HydraOS', '--removable', '--no-nvram'], runner.calls)
|
|
password_commands = [(args, data) for args, data in zip(runner.calls, runner.inputs) if 'chpasswd' in args]
|
|
self.assertEqual(len(password_commands), 1)
|
|
self.assertEqual(password_commands[0][1], ('emil:distinctive-private-password\n', True))
|
|
self.assertNotIn('distinctive-private-password', self.log.getvalue())
|
|
self.assertGreaterEqual(self.probe.calls, 2)
|
|
self.assertIn('hostonly_cmdline="no"', (self.root / 'etc/dracut.conf.d/90-hydra-installed.conf').read_text())
|
|
self.assertIn('omit_dracutmodules+=" dmsquash-live "', (self.root / 'etc/dracut.conf.d/90-hydra-installed.conf').read_text())
|
|
self.assertNotIn('root=live:', (self.root / 'etc/default/grub').read_text())
|
|
dracut = next(c for c in runner.calls if 'dracut' in c)
|
|
self.assertIn('6.12.1', dracut)
|
|
self.assertEqual(runner.calls[-1], ['umount', str(self.root)])
|
|
|
|
def test_partition_format_and_extract_failures_abort_and_cleanup_only_mounted(self):
|
|
for failing in ('parted', 'mkfs.vfat', 'mkfs.ext4', 'unsquashfs-extract'):
|
|
with self.subTest(failing=failing):
|
|
runner = FakeRunner(self.root, failing)
|
|
with self.assertRaises(i.InstallError):
|
|
self.install(runner)
|
|
self.assertFalse(any('grub-install' in c for c in runner.calls))
|
|
index = next(n for n, c in enumerate(runner.calls) if c[0] == failing or (failing == 'unsquashfs-extract' and c[0] == 'unsquashfs' and '-d' in c))
|
|
self.assertTrue(all(c[0] == 'umount' for c in runner.calls[index + 1:]))
|
|
if failing == 'unsquashfs-extract':
|
|
self.assertEqual(runner.calls[-1], ['umount', str(self.root)])
|
|
self.assertIn(failing.split('-extract')[0], self.log.getvalue())
|
|
|
|
def test_unmount_failure_is_visible(self):
|
|
runner = FakeRunner(self.root, 'umount')
|
|
with self.assertRaises(i.InstallError):
|
|
self.install(runner)
|
|
self.assertIn('cleanup', self.log.getvalue())
|
|
|
|
def test_password_newlines_rejected_before_commands(self):
|
|
for password in ('x\ny', 'x\ry', '', 'x\x00y'):
|
|
runner = FakeRunner(self.root)
|
|
with self.subTest(password=password), self.assertRaises(i.InstallError):
|
|
self.install(runner, password)
|
|
self.assertEqual(runner.calls, [])
|
|
|
|
def test_dry_run_cannot_enter_install_pipeline(self):
|
|
runner = FakeRunner(self.root)
|
|
self.request.execute = False
|
|
with self.assertRaises(i.InstallError):
|
|
self.install(runner)
|
|
self.assertEqual(runner.calls, [])
|
|
|
|
def test_live_manifest_arbitrary_paths_and_symlink_parents_refused(self):
|
|
for path in ('/etc/passwd', '/home/emil/data', '/../etc/sudoers.d/90-hydra-live'):
|
|
directory = self.root / 'etc/hydra/live-mode'
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
(directory / 'active').touch()
|
|
(directory / 'remove-on-install.json').write_text(json.dumps([path]))
|
|
with self.subTest(path=path), self.assertRaises(i.InstallError):
|
|
i.remove_live_files(self.root)
|
|
outside = Path(self.tmp.name) / 'outside'
|
|
outside.mkdir()
|
|
(outside / '90-hydra-live').write_text('must remain')
|
|
(self.root / 'etc/sudoers.d').symlink_to(outside)
|
|
(directory / 'remove-on-install.json').write_text(json.dumps(['/etc/sudoers.d/90-hydra-live']))
|
|
with self.assertRaises(i.InstallError):
|
|
i.remove_live_files(self.root)
|
|
self.assertTrue((outside / '90-hydra-live').exists())
|
|
|
|
def test_missing_or_multiple_kernels_are_explicit_errors(self):
|
|
with self.assertRaises(i.InstallError):
|
|
i.select_kernel(self.root, None)
|
|
for version in ('6.12.1', '6.12.2'):
|
|
(self.root / 'lib/modules' / version).mkdir(parents=True)
|
|
(self.root / 'boot').mkdir(exist_ok=True)
|
|
(self.root / 'boot' / ('vmlinuz-' + version)).touch()
|
|
with self.assertRaises(i.InstallError):
|
|
i.select_kernel(self.root, None)
|
|
self.assertEqual(i.select_kernel(self.root, '6.12.2'), '6.12.2')
|
|
|
|
|
|
class RunnerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertIsNotNone(i, 'installer module must implement the guarded installer')
|
|
|
|
def test_sensitive_failure_never_includes_stderr_or_stdin(self):
|
|
with patch('subprocess.run', return_value=subprocess.CompletedProcess([], 1, '', 'private-password')):
|
|
with self.assertRaises(i.InstallError) as caught:
|
|
i.Runner().run(['chroot', '/target', 'chpasswd'], input='user:private-password\n', sensitive=True)
|
|
self.assertNotIn('private-password', str(caught.exception))
|
|
|
|
def test_subprocess_uses_argv_without_shell_and_checked_returncode(self):
|
|
with patch('subprocess.run', return_value=subprocess.CompletedProcess([], 0, 'ok', '')) as call:
|
|
self.assertEqual(i.Runner().run(['tool', '/tmp/path;echo bad']), 'ok')
|
|
self.assertEqual(call.call_args.args[0], ['tool', '/tmp/path;echo bad'])
|
|
self.assertFalse(call.call_args.kwargs.get('shell', False))
|
|
|
|
|
|
class ProbeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertIsNotNone(i, 'installer module required')
|
|
|
|
def test_device_mapper_alias_and_dev_root_resolve_by_identity(self):
|
|
topo = topology(disk('/dev/vda', children=[disk('/dev/dm-0', kind='lvm')]))
|
|
mount = {'source': '/dev/root', 'target': '/', 'fstype': 'ext4', 'maj:min': topo.nodes['/dev/vda']['maj:min'], 'options': 'rw'}
|
|
probe = i.SystemProbe()
|
|
self.assertEqual(probe._block_source(mount, topo), '/dev/vda')
|
|
self.assertIn('/dev/vda', topo.protected)
|
|
|
|
def test_parent_reference_cycles_rejected(self):
|
|
data = json.dumps({'blockdevices': [disk('/dev/a', pkname='/dev/b'), disk('/dev/b', pkname='/dev/a')]})
|
|
with self.assertRaises(i.InstallError):
|
|
i.Topology.from_json(data)
|
|
|
|
def test_malformed_findmnt_rejected(self):
|
|
for data in ('{}', '{', '{"filesystems":[]}', '{"filesystems":[{"source":"overlay"}]}'):
|
|
with self.subTest(data=data), self.assertRaises(i.InstallError):
|
|
i.SystemProbe._mounts(data)
|
|
|
|
def test_unknown_mounted_block_source_refused(self):
|
|
mount = {'source': '/dev/missing', 'target': '/', 'fstype': 'ext4', 'maj:min': '100:1', 'options': 'rw'}
|
|
with self.assertRaises(i.InstallError):
|
|
i.SystemProbe()._block_source(mount, topology(disk()))
|
|
|
|
def test_payload_symlink_never_resolves_into_host_from_target(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / 'usr/lib/modules/6.12.1').mkdir(parents=True)
|
|
(root / 'lib').symlink_to('/usr/lib')
|
|
(root / 'boot').mkdir()
|
|
(root / 'boot/vmlinuz-6.12.1').touch()
|
|
self.assertEqual(i.select_kernel(root, None), '6.12.1')
|
|
with self.assertRaises(i.InstallError):
|
|
i.target_path(root, '/lib/modules')
|
|
|
|
def test_leaf_symlink_removal_does_not_follow_link(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp) / 'root'
|
|
outside = Path(tmp) / 'keep'
|
|
outside.write_text('keep')
|
|
(root / 'etc/hydra/live-mode').mkdir(parents=True)
|
|
(root / 'etc/sudoers.d').mkdir()
|
|
(root / 'etc/hydra/live-mode/active').touch()
|
|
(root / 'etc/hydra/live-mode/remove-on-install.json').write_text(json.dumps(['/etc/sudoers.d/90-hydra-live']))
|
|
(root / 'etc/sudoers.d/90-hydra-live').symlink_to(outside)
|
|
i.remove_live_files(root)
|
|
self.assertEqual(outside.read_text(), 'keep')
|
|
self.assertFalse((root / 'etc/sudoers.d/90-hydra-live').is_symlink())
|
|
|
|
def test_read_only_cli_invalid_target_and_confirmation_never_prompt_password(self):
|
|
with patch.object(i, 'getpass') as password, contextlib.redirect_stderr(io.StringIO()):
|
|
self.assertEqual(i.main(['--disk', '/dev/hydra-this-device-does-not-exist', '--user', 'emil']), 1)
|
|
self.assertEqual(i.main(['--execute']), 1)
|
|
password.getpass.assert_not_called()
|
|
|
|
def test_preflight_refuses_container_before_execution(self):
|
|
with patch('os.geteuid', return_value=0), patch.object(i.sys.stdin, 'isatty', return_value=True), \
|
|
patch.object(i.Path, 'exists', return_value=True):
|
|
with self.assertRaisesRegex(i.InstallError, 'container'):
|
|
i.preflight(i.Request('/dev/vda', 'emil'), FakeRunner(Path('/unused')))
|
|
|
|
|
|
class AccountTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertIsNotNone(i)
|
|
|
|
def test_live_account_removed_without_touching_other_home(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / 'etc').mkdir()
|
|
(root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\nhydra-live:x:1000:1000:Live:/home/hydra-live:/bin/bash\n')
|
|
(root / 'home/hydra-live').mkdir(parents=True)
|
|
(root / 'home/hydra-live/config').touch()
|
|
(root / 'home/keep').mkdir()
|
|
runner = FakeRunner(root)
|
|
i.remove_live_account(root, runner.run)
|
|
self.assertIn(['chroot', str(root), 'userdel', 'hydra-live'], runner.calls)
|
|
self.assertFalse((root / 'home/hydra-live').exists())
|
|
self.assertTrue((root / 'home/keep').exists())
|
|
|
|
def test_live_account_unexpected_home_or_root_uid_refused(self):
|
|
for row in ('hydra-live:x:1000:1000:Live:/home/other:/bin/bash',
|
|
'hydra-live:x:0:0:Live:/home/hydra-live:/bin/bash'):
|
|
with self.subTest(row=row), tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / 'etc').mkdir()
|
|
(root / 'etc/passwd').write_text(row + '\n')
|
|
runner = FakeRunner(root)
|
|
with self.assertRaises(i.InstallError):
|
|
i.remove_live_account(root, runner.run)
|
|
self.assertEqual(runner.calls, [])
|
|
|
|
def test_live_manifest_nested_unit_paths_rejected(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
live = root / 'etc/hydra/live-mode'
|
|
live.mkdir(parents=True)
|
|
(live / 'active').touch()
|
|
(live / 'remove-on-install.json').write_text(json.dumps(['/etc/systemd/system/getty@foo/bar.service.d/90-hydra-live.conf']))
|
|
with self.assertRaises(i.InstallError):
|
|
i.remove_live_files(root)
|
|
|
|
|
|
class AdditionalFailureTests(unittest.TestCase):
|
|
setUp = PipelineTests.setUp
|
|
install = PipelineTests.install
|
|
def test_later_chroot_failure_aborts_bootloader_and_unmounts_all(self):
|
|
runner = FakeRunner(self.root, 'dracut')
|
|
with self.assertRaises(i.InstallError):
|
|
self.install(runner)
|
|
self.assertFalse(any('grub-install' in c for c in runner.calls))
|
|
self.assertEqual([c for c in runner.calls if c[0] == 'umount'], [
|
|
['umount', '--recursive', str(self.root / 'sys')],
|
|
['umount', str(self.root / 'proc')],
|
|
['umount', '--recursive', str(self.root / 'dev')],
|
|
['umount', str(self.root / 'boot/efi')],
|
|
['umount', str(self.root)],
|
|
])
|
|
|
|
def test_revalidation_detects_mounted_disk_before_partitioning(self):
|
|
class ChangedProbe(FakeProbe):
|
|
def snapshot(self, payload):
|
|
self.calls += 1
|
|
return topology(disk(mounts=['/mounted'])) if self.calls > 1 else self.topo
|
|
self.probe = ChangedProbe()
|
|
runner = FakeRunner(self.root)
|
|
with self.assertRaises(i.InstallError):
|
|
self.install(runner)
|
|
self.assertFalse(any(c[0] == 'parted' for c in runner.calls))
|
|
|
|
def test_unknown_multidevice_filesystem_refused(self):
|
|
topo = topology(disk())
|
|
for kind in ('zfs', 'btrfs'):
|
|
mount = {'source': 'pool' if kind == 'zfs' else '/dev/vda',
|
|
'target': '/', 'fstype': kind, 'maj:min': '0:22', 'options': 'rw'}
|
|
with self.subTest(kind=kind), self.assertRaises(i.InstallError):
|
|
i.SystemProbe()._block_source(mount, topo)
|
|
|
|
def test_prompt_requires_controlling_tty(self):
|
|
with patch('builtins.open', side_effect=OSError('no tty')):
|
|
with self.assertRaises(i.InstallError):
|
|
i.prompt_password()
|
|
|
|
|
|
class SnapshotTests(unittest.TestCase):
|
|
def test_snapshot_connects_live_loop_to_its_physical_backing(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
payload = Path(tmp) / 'squashfs.img'
|
|
payload.touch()
|
|
data = {'blockdevices': [disk(), disk('/dev/vdb', children=[disk('/dev/vdb1', kind='part', mounts=['/run/initramfs/live'])]), disk('/dev/loop0', kind='loop', mounts=['/'])]}
|
|
media = {'source': '/dev/vdb1', 'target': '/run/initramfs/live', 'fstype': 'iso9660', 'maj:min': data['blockdevices'][1]['children'][0]['maj:min'], 'options': 'ro'}
|
|
loop = {'source': '/dev/loop0', 'target': '/', 'fstype': 'squashfs', 'maj:min': data['blockdevices'][2]['maj:min'], 'options': 'ro'}
|
|
class ReadOnlyRunner:
|
|
def run(self, args):
|
|
if args[0] == 'lsblk':
|
|
return json.dumps(data)
|
|
return json.dumps({'filesystems': [media] if '--target' in args else [media, loop]})
|
|
real_read = Path.read_text
|
|
def read(path, *args, **kw):
|
|
if str(path).startswith('/sys/dev/block/'):
|
|
return str(payload)
|
|
return real_read(path, *args, **kw)
|
|
with patch.object(Path, 'read_text', read):
|
|
topo = i.SystemProbe(ReadOnlyRunner()).snapshot(str(payload))
|
|
self.assertFalse(topo.component('/dev/vda') & topo.protected)
|
|
self.assertIn('/dev/loop0', topo.component('/dev/vdb'))
|
|
self.assertTrue(topo.component('/dev/vdb') & topo.protected)
|
|
|
|
def test_failed_extraction_retains_diagnostic_on_target(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp) / 'target'
|
|
root.mkdir()
|
|
request = i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
|
|
runner = FakeRunner(root, 'unsquashfs-extract')
|
|
with self.assertRaises(i.InstallError):
|
|
i.install(request, runner, FakeProbe(), root, 'secret', io.StringIO())
|
|
diagnostic = root / 'var/log/hydra-install.log'
|
|
self.assertTrue(diagnostic.is_file())
|
|
self.assertIn('unsquashfs', diagnostic.read_text())
|
|
self.assertNotIn('secret', diagnostic.read_text())
|
|
|
|
|
|
class TargetPathTests(unittest.TestCase):
|
|
def test_live_mail_parent_symlink_is_resolved_within_target(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / 'etc').mkdir()
|
|
(root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\n')
|
|
(root / 'var/spool').mkdir(parents=True)
|
|
(root / 'var/mail').mkdir()
|
|
(root / 'var/spool/mail').symlink_to('/var/mail')
|
|
i.remove_live_account(root, FakeRunner(root).run)
|
|
|
|
def test_kernel_version_symlink_is_not_followed_into_host(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp) / 'root'
|
|
external = Path(tmp) / 'external'
|
|
external.mkdir()
|
|
(root / 'lib/modules').mkdir(parents=True)
|
|
(root / 'lib/modules/6.12.1').symlink_to(external)
|
|
(root / 'boot').mkdir()
|
|
(root / 'boot/vmlinuz-6.12.1').touch()
|
|
with self.assertRaises(i.InstallError):
|
|
i.select_kernel(root, None)
|
|
|
|
|
|
class IdentityTests(unittest.TestCase):
|
|
def test_conflicting_device_identities_rejected(self):
|
|
a, b = disk(), disk('/dev/vdb')
|
|
b['maj:min'] = a['maj:min']
|
|
with self.assertRaises(i.InstallError):
|
|
i.Topology.from_json(json.dumps({'blockdevices': [a, b]}))
|
|
|
|
|
|
class TtyTests(unittest.TestCase):
|
|
def test_password_stream_opens_real_nonseekable_tty(self):
|
|
master, slave = pty.openpty()
|
|
try:
|
|
slave_name = os.ttyname(slave)
|
|
real_open = open
|
|
def tty_open(path, mode):
|
|
self.assertEqual(path, '/dev/tty')
|
|
return real_open(slave_name, mode, opener=lambda path, flags: os.open(path, flags | os.O_NOCTTY))
|
|
with patch('builtins.open', tty_open), patch.object(i.getpass, 'getpass', return_value='test-only-secret'):
|
|
self.assertEqual(i.prompt_password(), 'test-only-secret')
|
|
finally:
|
|
os.close(slave)
|
|
os.close(master)
|
|
|
|
|
|
class ReviewRegressionTests(unittest.TestCase):
|
|
def request(self):
|
|
return i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
|
|
|
|
def test_parentless_protected_stacked_root_refuses_all_targets(self):
|
|
for kind in ('lvm', 'crypt', 'dm', 'mpath', 'md', 'raid0', 'raid1', 'raid4', 'raid5', 'raid6', 'raid10'):
|
|
with self.subTest(kind=kind), self.assertRaisesRegex(i.InstallError, 'backing parent'):
|
|
topo = topology(disk(), disk('/dev/dm-0', kind=kind, mounts=['/']))
|
|
i.validate_request(self.request(), FakeProbe(topo))
|
|
|
|
def test_child_relationship_is_not_evidence_of_backing_parent(self):
|
|
for kind in ('part', 'lvm', 'crypt', 'raid1'):
|
|
node = disk('/dev/missing-backing', kind=kind,
|
|
children=[disk('/dev/dm-0', kind='crypt', mounts=['/'])])
|
|
with self.subTest(kind=kind), self.assertRaisesRegex(i.InstallError, 'backing parent'):
|
|
topo = topology(disk(), node)
|
|
i.validate_request(self.request(), FakeProbe(topo))
|
|
|
|
def test_complete_stack_on_other_disk_preserves_eligible_target(self):
|
|
stack = disk('/dev/vdb', children=[disk('/dev/vdb1', kind='part', children=[
|
|
disk('/dev/dm-crypt', kind='crypt', children=[disk('/dev/dm-lvm', kind='lvm', mounts=['/'])])])])
|
|
topo = topology(disk(), stack)
|
|
self.assertEqual(i.validate_request(self.request(), FakeProbe(topo)), '/dev/vda')
|
|
self.assertTrue(topo.component('/dev/vdb') & topo.protected)
|
|
|
|
def test_empty_optical_drive_does_not_block_eligible_target(self):
|
|
optical = disk('/dev/sr0', kind='rom')
|
|
optical['size'] = 0
|
|
topo = topology(disk(), optical)
|
|
self.assertEqual(i.validate_request(self.request(), FakeProbe(topo)), '/dev/vda')
|
|
self.assertFalse(topo.component('/dev/sr0') & topo.protected)
|
|
|
|
def test_zero_size_disk_cannot_be_selected(self):
|
|
target = disk()
|
|
target['size'] = 0
|
|
with self.assertRaises(i.InstallError):
|
|
topo = topology(target)
|
|
i.validate_request(self.request(), FakeProbe(topo))
|
|
|
|
def test_zero_size_mounted_optical_drive_is_not_treated_as_empty(self):
|
|
optical = disk('/dev/sr0', kind='rom', mounts=['/run/initramfs/live'])
|
|
optical['size'] = 0
|
|
with self.assertRaises(i.InstallError):
|
|
topology(disk(), optical)
|
|
|
|
def test_zero_size_optical_with_separately_listed_dependent_refused(self):
|
|
optical = disk('/dev/sr0', kind='rom')
|
|
optical['size'] = 0
|
|
dependent = disk('/dev/dm-0', kind='crypt', pkname='/dev/sr0', mounts=['/'])
|
|
with self.assertRaises(i.InstallError):
|
|
topology(disk(), optical, dependent)
|
|
|
|
def test_zero_size_optical_cannot_be_protected_by_findmnt(self):
|
|
optical = disk('/dev/sr0', kind='rom')
|
|
optical['size'] = 0
|
|
topo = topology(disk(), optical)
|
|
with self.assertRaises(i.InstallError):
|
|
topo.protect('/dev/sr0')
|
|
|
|
def test_zero_size_optical_with_dependents_is_not_irrelevant(self):
|
|
optical = disk('/dev/sr0', kind='rom', children=[disk('/dev/dm-0', kind='crypt', mounts=['/'])])
|
|
optical['size'] = 0
|
|
with self.assertRaises(i.InstallError):
|
|
topology(disk(), optical)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|