fix(installer): identify live loop backing after dracut switch-root

tests/test_installer.py

@@ -4,6 +4,9 @@ import io
 import json
 import os
 import pty
+import stat
+import struct
+from types import SimpleNamespace
 from pathlib import Path
 import subprocess
 import tempfile
@@ -403,6 +406,86 @@ class AdditionalFailureTests(unittest.TestCase):
                 i.prompt_password()
 
 
+class RelocatedLiveLoopTests(unittest.TestCase):
+    def probe(self, *, backing_device=None, backing_rdevice=0, ioctl_error=None,
+              opened_device=None, opened_mode=stat.S_IFBLK):
+        fixtures = Path(__file__).parent / 'fixtures'
+        data = json.loads((fixtures / 'installer-dracut-lsblk.json').read_text())
+        mounts = json.loads((fixtures / 'installer-dracut-findmnt.json').read_text())
+        media = next(m for m in i.SystemProbe._mounts(json.dumps(mounts)) if m['source'] == '/dev/sr0')
+        calls = []
+        class ReadOnlyRunner:
+            def run(self, args):
+                calls.append(args)
+                if args[0] == 'lsblk':
+                    return json.dumps(data)
+                if '--target' not in args:
+                    return json.dumps(mounts)
+                if args[args.index('--target') + 1] != i.DEFAULT_PAYLOAD:
+                    raise i.InstallError('Real dracut stale backing path has no current mount')
+                return json.dumps({'filesystems': [media]})
+        probe = i.SystemProbe(ReadOnlyRunner())
+        probe.physical_disk = lambda path: path == '/dev/vda'
+        packet = struct.pack('=QQQ', os.makedev(11, 0) if backing_device is None else backing_device,
+                             3462, backing_rdevice) + bytes(208)
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(patch.object(Path, 'is_file', return_value=True))
+            stack.enter_context(patch.object(Path, 'read_text', return_value='/LiveOS/squashfs.img'))
+            opened = stack.enter_context(patch('os.open', return_value=91))
+            closed = stack.enter_context(patch('os.close'))
+            stack.enter_context(patch('os.fstat', return_value=SimpleNamespace(st_mode=opened_mode,
+                st_rdev=os.makedev(7, 0) if opened_device is None else opened_device)))
+            # Patch the syscall even before the installer imports fcntl.
+            syscall = stack.enter_context(patch('fcntl.ioctl', side_effect=ioctl_error,
+                                                return_value=packet))
+            topo = probe.snapshot(i.DEFAULT_PAYLOAD)
+            opened.assert_called_once_with('/dev/loop0', os.O_RDONLY | os.O_CLOEXEC)
+            closed.assert_called_once_with(91)
+        return probe, topo, calls, syscall
+
+    def test_actual_switched_root_topology_resolves_loop_by_kernel_identity(self):
+        probe, topo, calls, syscall = self.probe()
+        self.assertFalse(topo.component('/dev/vda') & topo.protected)
+        self.assertIn('/dev/loop0', topo.component('/dev/sr0'))
+        self.assertTrue(topo.component('/dev/sr0') & topo.protected)
+        self.assertEqual(syscall.call_args.args, (91, 0x4C05, bytes(232)))
+        self.assertFalse(any('/LiveOS/squashfs.img' in args for args in calls))
+        probe.snapshot = lambda payload: topo
+        self.assertEqual(i.validate_request(i.Request('/dev/vda', 'hydratest'), probe), '/dev/vda')
+        probe.physical_disk = lambda path: path == '/dev/sr0'
+        with self.assertRaises(i.InstallError):
+            i.validate_request(i.Request('/dev/sr0', 'hydratest'), probe)
+        # A mounted loop backed by a normal disk still forbids that whole disk.
+        topo.nodes['/dev/sr0']['type'] = 'disk'
+        topo.nodes['/dev/sr0']['size'] = 32 * 1024 ** 3
+        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
+            i.validate_request(i.Request('/dev/sr0', 'hydratest'), probe)
+
+    def test_unknown_loop_filesystem_identity_fails_closed(self):
+        for identity in (os.makedev(8, 1), os.makedev(0, 99)):
+            with self.subTest(identity=identity), self.assertRaisesRegex(i.InstallError, 'Unknown physical backing'):
+                self.probe(backing_device=identity)
+
+    def test_unreadable_or_changed_loop_device_fails_closed(self):
+        cases = [{'ioctl_error': OSError('cannot read loop status')},
+                 {'opened_device': os.makedev(7, 1)}, {'opened_mode': stat.S_IFREG}]
+        for case in cases:
+            with self.subTest(case=case), self.assertRaises(i.InstallError):
+                self.probe(**case)
+
+    def test_truncated_loop_status_fails_closed_and_closes_descriptor(self):
+        with patch('os.open', return_value=91), patch('os.close') as closed, \
+             patch('os.fstat', return_value=SimpleNamespace(st_mode=stat.S_IFBLK, st_rdev=os.makedev(7, 0))), \
+             patch('fcntl.ioctl', return_value=bytes(16)):
+            with self.assertRaisesRegex(i.InstallError, 'Malformed loop status'):
+                i.SystemProbe._loop_backing_identity('/dev/loop0', '7:0')
+            closed.assert_called_once_with(91)
+
+    def test_direct_block_backing_uses_kernel_rdevice(self):
+        _, topo, _, _ = self.probe(backing_device=os.makedev(0, 6), backing_rdevice=os.makedev(11, 0))
+        self.assertIn('/dev/loop0', topo.component('/dev/sr0'))
+
+
 class SnapshotTests(unittest.TestCase):
     def test_snapshot_connects_live_loop_to_its_physical_backing(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -416,12 +499,8 @@ class SnapshotTests(unittest.TestCase):
                     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):
+            with patch.object(i.SystemProbe, '_loop_backing_identity',
+                              return_value=media['maj:min']):
                 topo = i.SystemProbe(ReadOnlyRunner()).snapshot(str(payload))
             self.assertFalse(topo.component('/dev/vda') & topo.protected)
             self.assertIn('/dev/loop0', topo.component('/dev/vdb'))