fix(installer): resolve nested loop ancestry to a fixed point

tests/test_installer.py

@@ -486,6 +486,83 @@ class RelocatedLiveLoopTests(unittest.TestCase):
         self.assertIn('/dev/loop0', topo.component('/dev/sr0'))
 
 
+class NestedLoopAncestryTests(unittest.TestCase):
+    def probe(self, backings=None, *, indirect=False, payload_loop=False, shared=False):
+        loop0 = disk('/dev/loop0', kind='loop', mounts=['/mnt/also'] if shared else [])
+        loop0['maj:min'] = '7:0'
+        loop1 = disk('/dev/loop1', kind='loop', mounts=[] if payload_loop else ['/mnt/valuable'])
+        loop1['maj:min'] = '7:1'
+        target, unrelated = disk('/dev/vda'), disk('/dev/vdb')
+        target['maj:min'], unrelated['maj:min'] = '253:0', '253:1'
+        media = disk('/dev/sr0', kind='rom', mounts=['/run/initramfs/live'])
+        media['maj:min'] = '11:0'
+        nodes = [loop0, loop1, target, unrelated, media]
+        if indirect:
+            partition = disk('/dev/loop0p1', kind='part', pkname='/dev/loop0')
+            partition['maj:min'] = '259:0'
+            nodes.insert(1, partition)
+        identities = backings or {'/dev/loop0': '253:0',
+                                 '/dev/loop1': '253:0' if shared else '259:0' if indirect else '7:0'}
+        mount = {'source': '/dev/loop1' if payload_loop else '/dev/sr0',
+                 'target': '/payload' if payload_loop else '/run/initramfs/live',
+                 'fstype': 'ext4' if payload_loop else 'iso9660',
+                 'maj:min': '7:1' if payload_loop else '11:0', 'options': 'ro'}
+        class ReadOnlyRunner:
+            def run(self, args):
+                if args[0] == 'lsblk':
+                    return json.dumps({'blockdevices': nodes})
+                full_mount = {'source': '/dev/sr0', 'target': '/run/initramfs/live',
+                              'fstype': 'iso9660', 'maj:min': '11:0', 'options': 'ro'}
+                return json.dumps({'filesystems': [mount] if '--target' in args else [full_mount]})
+        probe = i.SystemProbe(ReadOnlyRunner())
+        probe.physical_disk = lambda path: path in {'/dev/vda', '/dev/vdb'}
+        with patch.object(Path, 'is_file', return_value=True), \
+             patch.object(i.SystemProbe, '_loop_backing_identity',
+                          side_effect=lambda path, expected: identities[path]):
+            topo = probe.snapshot(i.DEFAULT_PAYLOAD)
+        probe.snapshot = lambda payload: topo
+        return probe, topo
+
+    def test_reverse_order_nested_loop_refuses_physical_backing(self):
+        probe, _ = self.probe()
+        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
+            i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
+
+    def test_reverse_order_nested_loop_allows_unrelated_disk(self):
+        probe, topo = self.probe()
+        self.assertIn('/dev/vda', topo.component('/dev/loop1'))
+        self.assertEqual(i.validate_request(i.Request('/dev/vdb', 'hydratest'), probe), '/dev/vdb')
+
+    def test_unknown_nested_backing_refuses_unrelated_installation(self):
+        with self.assertRaisesRegex(i.InstallError, 'Unknown physical backing'):
+            self.probe({'/dev/loop1': '7:0', '/dev/loop0': '0:99'})
+
+    def test_nested_backing_cycles_fail_closed(self):
+        for backings in ({'/dev/loop1': '7:0', '/dev/loop0': '7:1'},
+                         {'/dev/loop1': '7:1', '/dev/loop0': '253:0'}):
+            with self.subTest(backings=backings), self.assertRaisesRegex(i.InstallError, '[Cc]yclic'):
+                self.probe(backings)
+
+    def test_nested_backing_through_loop_partition_refuses_disk(self):
+        probe, _ = self.probe(indirect=True)
+        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
+            i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
+
+    def test_cycle_through_loop_partition_fails_closed(self):
+        with self.assertRaisesRegex(i.InstallError, '[Cc]yclic'):
+            self.probe({'/dev/loop1': '259:0', '/dev/loop0': '7:1'}, indirect=True)
+
+    def test_unmounted_payload_loop_ancestors_refuse_disk(self):
+        probe, _ = self.probe(payload_loop=True)
+        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
+            i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
+
+    def test_two_loops_on_same_backing_are_not_a_directed_cycle(self):
+        probe, topo = self.probe(shared=True)
+        self.assertIn('/dev/loop0', topo.component('/dev/loop1'))
+        self.assertEqual(i.validate_request(i.Request('/dev/vdb', 'hydratest'), probe), '/dev/vdb')
+
+
 class SnapshotTests(unittest.TestCase):
     def test_snapshot_connects_live_loop_to_its_physical_backing(self):
         with tempfile.TemporaryDirectory() as tmp: