Reject incomplete backing ancestry and allow empty optical drives

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]:
+            if node['size'] == 0 and result.edges[path]:
-                raise InstallError('Partition has no known parent')
+                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):

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()