From 4ce155ea6efb04fd034f56a51e3e6775eb004098 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 13 Oct 2026 04:22:14 +0300 Subject: [PATCH] Reject incomplete backing ancestry and allow empty optical drives --- hydra/installer.py | 18 +++++++++-- tests/test_installer.py | 66 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 3 deletions(-) diff --git a/hydra/installer.py b/hydra/installer.py index e439e85..d08a6ed 100644 --- a/hydra/installer.py +++ b/hydra/installer.py @@ -97,8 +97,13 @@ class Topology: size, identity = node.get('size'), node.get('maj:min') if not isinstance(mounts, list) or any(m is not None and not isinstance(m, str) for m in mounts): raise InstallError('Malformed lsblk mountpoints') - if isinstance(size, bool) or not isinstance(size, int) or size <= 0: + if isinstance(size, bool) or not isinstance(size, int) or size < 0: raise InstallError('Invalid lsblk size') + empty_optical = (kind == 'rom' and not any(mounts) + and not node.get('children') and node.get('pkname') is None + and ancestor is None) + if size == 0 and not empty_optical: + raise InstallError('Zero-size device is not an unrelated empty optical drive') if not isinstance(identity, str) or not re.fullmatch(r'\d+:\d+', identity): raise InstallError('Invalid lsblk device identity') if path in result.nodes and any(result.nodes[path][key] != node[key] for key in ('type', 'size', 'maj:min', 'mountpoints')): @@ -149,8 +154,13 @@ class Topology: for path in directed: check_cycle(path) for path, node in result.nodes.items(): - if node['type'] == 'part' and not result.edges[path]: - raise InstallError('Partition has no known parent') + if node['size'] == 0 and result.edges[path]: + raise InstallError('Empty optical drive has block-device dependencies') + # An undirected child edge does not establish backing ancestry. + # Each layer must identify its own parent, even when another + # apparently unrelated disk is the requested installation target. + if node['type'] not in {'disk', 'rom', 'loop', 'zram'} and not directed[path]: + raise InstallError(f"Device has no known backing parent: {path} ({node['type']})") return result def link(self, child, parent): @@ -162,6 +172,8 @@ class Topology: def protect(self, path): if path not in self.nodes: raise InstallError(f'Cannot resolve mounted/live source: {path}') + if self.nodes[path]['size'] == 0: + raise InstallError(f'Empty optical drive unexpectedly backs mounted/live storage: {path}') self.protected.add(path) def component(self, path): diff --git a/tests/test_installer.py b/tests/test_installer.py index 6f6b969..5085b07 100644 --- a/tests/test_installer.py +++ b/tests/test_installer.py @@ -489,5 +489,71 @@ class TtyTests(unittest.TestCase): 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()