Zed: Backport fixes for OSSA-2024-002

bindep.txt

@@ -54,3 +54,5 @@ pcre-devel [platform:rpm test]
 # runtime and in unit tests. Net result is the same that lsscsi will be
 # installed for any nova installation.
 lsscsi
+# NOTE(priteau): Install qemu-img for format inspector tests
+qemu-utils [platform:dpkg test]

nova/tests/unit/console/test_websocketproxy.py

@@ -635,7 +635,9 @@ def test_reject_open_redirect(self, url='//example.com/%2F..'):
         # now the same url but with extra leading '/' characters removed.
         if expected_cpython in errmsg:
             location = result[3].decode()
-            location = location.removeprefix('Location: ').rstrip('\r\n')
+            if location.startswith('Location: '):
+                location = location[len('Location: '):]
+            location = location.rstrip('\r\n')
             self.assertTrue(
                 location.startswith('/example.com/%2F..'),
                 msg='Redirect location is not the expected sanitized URL',

nova/tests/unit/virt/libvirt/test_utils.py

@@ -405,12 +405,12 @@ def test_fetch_initrd_image(self, mock_images):
             _context, image_id, target, trusted_certs)
 
     @mock.patch.object(images, 'IMAGE_API')
-    @mock.patch.object(format_inspector, 'get_inspector')
+    @mock.patch.object(format_inspector, 'detect_file_format')
     @mock.patch.object(compute_utils, 'disk_ops_semaphore')
     @mock.patch('nova.privsep.utils.supports_direct_io', return_value=True)
     @mock.patch('nova.privsep.qemu.unprivileged_convert_image')
     def test_fetch_raw_image(self, mock_convert_image, mock_direct_io,
-                             mock_disk_op_sema, mock_gi, mock_glance):
+                             mock_disk_op_sema, mock_detect, mock_glance):
 
         def fake_rename(old, new):
             self.executes.append(('mv', old, new))
@@ -450,7 +450,7 @@ class FakeImgInfo(object):
         self.stub_out('oslo_utils.fileutils.delete_if_exists',
                       fake_rm_on_error)
 
-        mock_inspector = mock_gi.return_value.from_file.return_value
+        mock_inspector = mock_detect.return_value
 
         # Since the remove param of fileutils.remove_path_on_error()
         # is initialized at load time, we must provide a wrapper
@@ -464,6 +464,7 @@ class FakeImgInfo(object):
 
         # Make sure qcow2 gets converted to raw
         mock_inspector.safety_check.return_value = True
+        mock_inspector.__str__.return_value = 'qcow2'
         mock_glance.get.return_value = {'disk_format': 'qcow2'}
         target = 't.qcow2'
         self.executes = []
@@ -477,12 +478,13 @@ class FakeImgInfo(object):
             CONF.instances_path, False)
         mock_convert_image.reset_mock()
         mock_inspector.safety_check.assert_called_once_with()
-        mock_gi.assert_called_once_with('qcow2')
+        mock_detect.assert_called_once_with('t.qcow2.part')
 
         # Make sure raw does not get converted
-        mock_gi.reset_mock()
+        mock_detect.reset_mock()
         mock_inspector.safety_check.reset_mock()
         mock_inspector.safety_check.return_value = True
+        mock_inspector.__str__.return_value = 'raw'
         mock_glance.get.return_value = {'disk_format': 'raw'}
         target = 't.raw'
         self.executes = []
@@ -491,12 +493,13 @@ class FakeImgInfo(object):
         self.assertEqual(self.executes, expected_commands)
         mock_convert_image.assert_not_called()
         mock_inspector.safety_check.assert_called_once_with()
-        mock_gi.assert_called_once_with('raw')
+        mock_detect.assert_called_once_with('t.raw.part')
 
         # Make sure safety check failure prevents us from proceeding
-        mock_gi.reset_mock()
+        mock_detect.reset_mock()
         mock_inspector.safety_check.reset_mock()
         mock_inspector.safety_check.return_value = False
+        mock_inspector.__str__.return_value = 'qcow2'
         mock_glance.get.return_value = {'disk_format': 'qcow2'}
         target = 'backing.qcow2'
         self.executes = []
@@ -506,10 +509,10 @@ class FakeImgInfo(object):
         self.assertEqual(self.executes, expected_commands)
         mock_convert_image.assert_not_called()
         mock_inspector.safety_check.assert_called_once_with()
-        mock_gi.assert_called_once_with('qcow2')
+        mock_detect.assert_called_once_with('backing.qcow2.part')
 
         # Make sure a format mismatch prevents us from proceeding
-        mock_gi.reset_mock()
+        mock_detect.reset_mock()
         mock_inspector.safety_check.reset_mock()
         mock_inspector.safety_check.side_effect = (
             format_inspector.ImageFormatError)
@@ -522,7 +525,7 @@ class FakeImgInfo(object):
         self.assertEqual(self.executes, expected_commands)
         mock_convert_image.assert_not_called()
         mock_inspector.safety_check.assert_called_once_with()
-        mock_gi.assert_called_once_with('qcow2')
+        mock_detect.assert_called_once_with('backing.qcow2.part')
 
         del self.executes
 

nova/tests/unit/virt/test_images.py

@@ -21,7 +21,6 @@
 
 from nova.compute import utils as compute_utils
 from nova import exception
-from nova.image import format_inspector
 from nova import test
 from nova.virt import images
 
@@ -101,15 +100,16 @@ def test_qemu_img_info_with_disk_not_found(self, exists, mocked_execute):
         mocked_execute.assert_called_once()
 
     @mock.patch.object(images, 'IMAGE_API')
-    @mock.patch('nova.image.format_inspector.get_inspector')
+    @mock.patch('nova.image.format_inspector.detect_file_format')
     @mock.patch.object(images, 'convert_image',
                        side_effect=exception.ImageUnacceptable)
     @mock.patch.object(images, 'qemu_img_info')
     @mock.patch.object(images, 'fetch')
     def test_fetch_to_raw_errors(self, convert_image, qemu_img_info, fetch,
-                                 get_inspector, glance):
-        inspector = get_inspector.return_value.from_file.return_value
+                                 mock_detect, glance):
+        inspector = mock_detect.return_value
         inspector.safety_check.return_value = True
+        inspector.__str__.return_value = 'qcow2'
         glance.get.return_value = {'disk_format': 'qcow2'}
         qemu_img_info.backing_file = None
         qemu_img_info.file_format = 'qcow2'
@@ -120,16 +120,17 @@ def test_fetch_to_raw_errors(self, convert_image, qemu_img_info, fetch,
                                None, 'href123', '/no/path')
 
     @mock.patch.object(images, 'IMAGE_API')
-    @mock.patch('nova.image.format_inspector.get_inspector')
+    @mock.patch('nova.image.format_inspector.detect_file_format')
     @mock.patch.object(images, 'convert_image',
                        side_effect=exception.ImageUnacceptable)
     @mock.patch.object(images, 'qemu_img_info')
     @mock.patch.object(images, 'fetch')
     def test_fetch_to_raw_data_file(self, convert_image, qemu_img_info_fn,
-                                    fetch, mock_gi, mock_glance):
+                                    fetch, mock_detect, mock_glance):
         mock_glance.get.return_value = {'disk_format': 'qcow2'}
-        inspector = mock_gi.return_value.from_file.return_value
+        inspector = mock_detect.return_value
         inspector.safety_check.return_value = True
+        inspector.__str__.return_value = 'qcow2'
         # NOTE(danms): the above test needs the following line as well, as it
         # is broken without it.
         qemu_img_info = qemu_img_info_fn.return_value
@@ -142,16 +143,17 @@ def test_fetch_to_raw_data_file(self, convert_image, qemu_img_info_fn,
                                images.fetch_to_raw,
                                None, 'href123', '/no/path')
 
-    @mock.patch('nova.image.format_inspector.get_inspector')
+    @mock.patch('nova.image.format_inspector.detect_file_format')
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch('os.rename')
     @mock.patch.object(images, 'qemu_img_info')
     @mock.patch.object(images, 'fetch')
     def test_fetch_to_raw_from_raw(self, fetch, qemu_img_info_fn, mock_rename,
-                                   mock_glance, mock_gi):
+                                   mock_glance, mock_detect):
         # Make sure we support a case where we fetch an already-raw image and
         # qemu-img returns None for "format_specific".
         mock_glance.get.return_value = {'disk_format': 'raw'}
+        mock_detect.return_value.__str__.return_value = 'raw'
         qemu_img_info = qemu_img_info_fn.return_value
         qemu_img_info.file_format = 'raw'
         qemu_img_info.backing_file = None
@@ -215,14 +217,15 @@ def test_convert_image_vmdk_allowed_list_checking(self):
                                                  format='json'))
 
     @mock.patch.object(images, 'IMAGE_API')
-    @mock.patch('nova.image.format_inspector.get_inspector')
+    @mock.patch('nova.image.format_inspector.detect_file_format')
     @mock.patch.object(images, 'fetch')
     @mock.patch('nova.privsep.qemu.unprivileged_qemu_img_info')
-    def test_fetch_checks_vmdk_rules(self, mock_info, mock_fetch, mock_gi,
+    def test_fetch_checks_vmdk_rules(self, mock_info, mock_fetch, mock_detect,
                                      mock_glance):
         mock_glance.get.return_value = {'disk_format': 'vmdk'}
-        inspector = mock_gi.return_value.from_file.return_value
+        inspector = mock_detect.return_value
         inspector.safety_check.return_value = True
+        inspector.__str__.return_value = 'vmdk'
         info = {'format': 'vmdk',
                 'format-specific': {
                     'type': 'vmdk',
@@ -235,22 +238,54 @@ def test_fetch_checks_vmdk_rules(self, mock_info, mock_fetch, mock_gi,
                                   images.fetch_to_raw, None, 'foo', 'anypath')
             self.assertIn('Invalid VMDK create-type specified', str(e))
 
+    @mock.patch('os.rename')
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch('nova.image.format_inspector.get_inspector')
+    @mock.patch('nova.image.format_inspector.detect_file_format')
+    @mock.patch.object(images, 'fetch')
+    @mock.patch('nova.privsep.qemu.unprivileged_qemu_img_info')
+    def test_fetch_iso_is_raw(
+        self, mock_info, mock_fetch, mock_detect_file_format, mock_gi,
+        mock_glance, mock_rename):
+        mock_glance.get.return_value = {'disk_format': 'iso'}
+        inspector = mock_gi.return_value.from_file.return_value
+        inspector.safety_check.return_value = True
+        inspector.__str__.return_value = 'iso'
+        mock_detect_file_format.return_value = inspector
+        # qemu-img does not have a parser for iso so it is treated as raw
+        info = {
+            "virtual-size": 356352,
+            "filename": "foo.iso",
+            "format": "raw",
+            "actual-size": 356352,
+            "dirty-flag": False
+        }
+        mock_info.return_value = jsonutils.dumps(info)
+        with mock.patch('os.path.exists', return_value=True):
+            images.fetch_to_raw(None, 'foo', 'anypath')
+        # Make sure we called info with -f raw for an iso, since qemu-img does
+        # not support iso
+        mock_info.assert_called_once_with('anypath.part', format=None)
+        # Make sure that since we considered this to be a raw file, we did the
+        # just-rename-don't-convert path
+        mock_rename.assert_called_once_with('anypath.part', 'anypath')
+
+    @mock.patch.object(images, 'IMAGE_API')
+    @mock.patch('nova.image.format_inspector.detect_file_format')
     @mock.patch.object(images, 'qemu_img_info')
     @mock.patch.object(images, 'fetch')
-    def test_fetch_to_raw_inspector(self, fetch, qemu_img_info, mock_gi,
+    def test_fetch_to_raw_inspector(self, fetch, qemu_img_info, mock_detect,
                                     mock_glance):
         # Image claims to be qcow2, is qcow2, but fails safety check, so we
         # abort before qemu-img-info
         mock_glance.get.return_value = {'disk_format': 'qcow2'}
-        inspector = mock_gi.return_value.from_file.return_value
+        inspector = mock_detect.return_value
         inspector.safety_check.return_value = False
+        inspector.__str__.return_value = 'qcow2'
         self.assertRaises(exception.ImageUnacceptable,
                           images.fetch_to_raw, None, 'href123', '/no.path')
         qemu_img_info.assert_not_called()
-        mock_gi.assert_called_once_with('qcow2')
-        mock_gi.return_value.from_file.assert_called_once_with('/no.path.part')
+        mock_detect.assert_called_once_with('/no.path.part')
         inspector.safety_check.assert_called_once_with()
         mock_glance.get.assert_called_once_with(None, 'href123')
 
@@ -264,18 +299,17 @@ def test_fetch_to_raw_inspector(self, fetch, qemu_img_info, mock_gi,
         # Image claims to be qcow2 in glance, but the image is something else,
         # so we abort before qemu-img-info
         qemu_img_info.reset_mock()
-        mock_gi.reset_mock()
+        mock_detect.reset_mock()
         inspector.safety_check.reset_mock()
-        mock_gi.return_value.from_file.side_effect = (
-            format_inspector.ImageFormatError)
+        mock_detect.return_value.__str__.return_value = 'vmdk'
         self.assertRaises(exception.ImageUnacceptable,
                           images.fetch_to_raw, None, 'href123', '/no.path')
-        mock_gi.assert_called_once_with('qcow2')
-        inspector.safety_check.assert_not_called()
+        mock_detect.assert_called_once_with('/no.path.part')
+        inspector.safety_check.assert_called_once_with()
         qemu_img_info.assert_not_called()
 
     @mock.patch.object(images, 'IMAGE_API')
-    @mock.patch('nova.image.format_inspector.get_inspector')
+    @mock.patch('nova.image.format_inspector.detect_file_format')
     @mock.patch.object(images, 'qemu_img_info')
     @mock.patch.object(images, 'fetch')
     def test_fetch_to_raw_inspector_disabled(self, fetch, qemu_img_info,
@@ -288,36 +322,41 @@ def test_fetch_to_raw_inspector_disabled(self, fetch, qemu_img_info,
         # If deep inspection is disabled, we should never call the inspector
         mock_gi.assert_not_called()
         # ... and we let qemu-img detect the format itself.
-        qemu_img_info.assert_called_once_with('/no.path.part',
-                                              format=None)
+        qemu_img_info.assert_called_once_with('/no.path.part')
         mock_glance.get.assert_not_called()
 
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch.object(images, 'qemu_img_info')
-    def test_fetch_inspect_ami(self, imginfo, glance):
+    @mock.patch('nova.image.format_inspector.detect_file_format')
+    def test_fetch_inspect_ami(self, detect, imginfo, glance):
         glance.get.return_value = {'disk_format': 'ami'}
+        detect.return_value.__str__.return_value = 'raw'
         self.assertRaises(exception.ImageUnacceptable,
                           images.fetch_to_raw, None, 'href123', '/no.path')
         # Make sure 'ami was translated into 'raw' before we call qemu-img
-        imginfo.assert_called_once_with('/no.path.part', format='raw')
+        imginfo.assert_called_once_with('/no.path.part')
 
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch.object(images, 'qemu_img_info')
-    def test_fetch_inspect_aki(self, imginfo, glance):
+    @mock.patch('nova.image.format_inspector.detect_file_format')
+    def test_fetch_inspect_aki(self, detect, imginfo, glance):
         glance.get.return_value = {'disk_format': 'aki'}
+        detect.return_value.__str__.return_value = 'raw'
         self.assertRaises(exception.ImageUnacceptable,
                           images.fetch_to_raw, None, 'href123', '/no.path')
         # Make sure 'aki was translated into 'raw' before we call qemu-img
-        imginfo.assert_called_once_with('/no.path.part', format='raw')
+        imginfo.assert_called_once_with('/no.path.part')
 
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch.object(images, 'qemu_img_info')
-    def test_fetch_inspect_ari(self, imginfo, glance):
+    @mock.patch('nova.image.format_inspector.detect_file_format')
+    def test_fetch_inspect_ari(self, detect, imginfo, glance):
         glance.get.return_value = {'disk_format': 'ari'}
+        detect.return_value.__str__.return_value = 'raw'
         self.assertRaises(exception.ImageUnacceptable,
                           images.fetch_to_raw, None, 'href123', '/no.path')
         # Make sure 'aki was translated into 'raw' before we call qemu-img
-        imginfo.assert_called_once_with('/no.path.part', format='raw')
+        imginfo.assert_called_once_with('/no.path.part')
 
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch.object(images, 'qemu_img_info')
@@ -330,13 +369,16 @@ def test_fetch_inspect_unknown_format(self, imginfo, glance):
 
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch.object(images, 'qemu_img_info')
-    @mock.patch('nova.image.format_inspector.get_inspector')
-    def test_fetch_inspect_disagrees_qemu(self, mock_gi, imginfo, glance):
+    @mock.patch('nova.image.format_inspector.detect_file_format')
+    def test_fetch_inspect_disagrees_qemu(self, mock_detect, imginfo, glance):
         glance.get.return_value = {'disk_format': 'qcow2'}
+        mock_detect.return_value.__str__.return_value = 'qcow2'
         # Glance and inspector think it is a qcow2 file, but qemu-img does not
-        # agree. It was forced to interpret as a qcow2, but returned no
-        # format information as a result.
+        # agree.
         imginfo.return_value.data_file = None
-        self.assertRaises(exception.ImageUnacceptable,
-                          images.fetch_to_raw, None, 'href123', '/no.path')
-        imginfo.assert_called_once_with('/no.path.part', format='qcow2')
+        imginfo.return_value.file_format = 'vmdk'
+        ex = self.assertRaises(exception.ImageUnacceptable,
+                               images.fetch_to_raw,
+                               None, 'href123', '/no.path')
+        self.assertIn('content does not match disk_format', str(ex))
+        imginfo.assert_called_once_with('/no.path.part')