NFS driver: Fix fail creating volume with multiple snapshots

The NFS driver uses qcow2 images with backing files to represent
volume snapshots, which is not allowed for qcow2 disk images
downloaded from glance.  The driver uses cinder.image_utils to
convert a qcow2 snapshot to a raw volume; this was not a problem
for the first snapshot, whose backing file is raw, and hence passed
the image format inspector, but the second snapshot has a qcow2
backing file, which the image_utils were rejecting as a security
risk.  Thus we now pass the qemu_img_info from the backing image as
an additional parameter to the image convert call, which indicates
that the file has already been screened and allows the conversion
to occur.

Co-authored-by: Fernando Ferraz <fesilva@redhat.com>
Co-authored-by: Brian Rosmaita <rosmaita.fossdev@gmail.com>

Closes-bug: #2074377
Change-Id: I49404e87eb0c77b4ed92918404f86c073fbfd713

cinder/image/image_utils.py

@@ -376,18 +376,20 @@ def check_qemu_img_version(minimum_version: str) -> None:
         raise exception.VolumeBackendAPIException(data=_msg)
 
 
-def _convert_image(prefix: tuple,
-                   source: str,
-                   dest: str,
-                   out_format: str,
-                   out_subformat: Optional[str] = None,
-                   src_format: Optional[str] = None,
-                   run_as_root: bool = True,
-                   cipher_spec: Optional[dict] = None,
-                   passphrase_file: Optional[str] = None,
-                   compress: bool = False,
-                   src_passphrase_file: Optional[str] = None,
-                   disable_sparse: bool = False) -> None:
+def _convert_image(
+        prefix: tuple,
+        source: str,
+        dest: str,
+        out_format: str,
+        out_subformat: Optional[str] = None,
+        src_format: Optional[str] = None,
+        run_as_root: bool = True,
+        cipher_spec: Optional[dict] = None,
+        passphrase_file: Optional[str] = None,
+        compress: bool = False,
+        src_passphrase_file: Optional[str] = None,
+        disable_sparse: bool = False,
+        src_img_info: Optional[imageutils.QemuImgInfo] = None) -> None:
     """Convert image to other format.
 
     NOTE: If the qemu-img convert command fails and this function raises an
@@ -406,6 +408,8 @@ def _convert_image(prefix: tuple,
     :param compress: compress w/ qemu-img when possible (best effort)
     :param src_passphrase_file: filename containing source volume's
                                 luks passphrase
+    :param src_img_info: a imageutils.QemuImgInfo object from this image,
+                            or None
     """
 
     # Check whether O_DIRECT is supported and set '-t none' if it is
@@ -462,17 +466,34 @@ def _convert_image(prefix: tuple,
     # some incredible event this is 0 (cirros image?) don't barf
     if duration < 1:
         duration = 1
-    try:
-        image_size = qemu_img_info(source,
-                                   run_as_root=run_as_root).virtual_size
-    except ValueError as e:
+
+    image_info = src_img_info
+    if not image_info:
+        try:
+            image_info = qemu_img_info(source, run_as_root=run_as_root)
+        except Exception:
+            # NOTE: at this point, the image conversion has already
+            # happened, and all that's left is some performance logging.
+            # So ignoring an exception from qemu_img_info here is not a
+            # security risk.  I'm afraid that if we are too strict here
+            # we will cause a regression, given that the converted image
+            # source could be cinder glance_store, Glance, or one of
+            # cinder's own backend drivers.  The nfs driver knows
+            # to pass in a src_img_info object, but others may not.
+            #
+            # We are catching the most general Exception here for a
+            # similar reason: the image conversion has already happened.
+            # If the conversion raised a ProcessExecutionError, we would
+            # never have reached this point.  But a PEE now is meaningless,
+            # so we ignore it.
+            pass
+    if not image_info or image_info.virtual_size is None:
         msg = ("The image was successfully converted, but image size "
-               "is unavailable. src %(src)s, dest %(dest)s. %(error)s")
-        LOG.info(msg, {"src": source,
-                       "dest": dest,
-                       "error": e})
+               "is unavailable. src %(src)s, dest %(dest)s")
+        LOG.info(msg, {"src": source, "dest": dest})
         return
 
+    image_size = image_info.virtual_size
     fsz_mb = image_size / units.Mi
     mbps = (fsz_mb / duration)
     msg = ("Image conversion details: src %(src)s, size %(sz).2f MB, "
@@ -539,7 +560,8 @@ def convert_image(source: str,
                        passphrase_file=passphrase_file,
                        compress=compress,
                        src_passphrase_file=src_passphrase_file,
-                       disable_sparse=disable_sparse)
+                       disable_sparse=disable_sparse,
+                       src_img_info=data)
 
 
 def resize_image(source: str,

cinder/tests/unit/test_image_utils.py

@@ -496,6 +496,80 @@ class TestConvertImage(test.TestCase):
         mock_log.assert_called_with('Insufficient free space on fakedir for'
                                     ' image conversion.')
 
+    @mock.patch('cinder.image.image_utils.qemu_img_info')
+    @mock.patch('cinder.utils.execute')
+    @mock.patch('cinder.image.image_utils._get_qemu_convert_cmd')
+    @mock.patch('cinder.utils.is_blk_device', return_value=False)
+    @mock.patch.object(image_utils.LOG, 'info')
+    @mock.patch.object(image_utils.LOG, 'debug')
+    def test__convert_image_no_virt_size(self,
+                                         mock_debug_log,
+                                         mock_info_log,
+                                         mock_isblk,
+                                         mock_cmd,
+                                         mock_execute,
+                                         mock_info):
+        """Make sure we don't try to do math with a None value"""
+        prefix = ('cgexec', '-g', 'blkio:cg')
+        source = '/source'
+        dest = '/dest'
+        out_format = 'unspecified'
+
+        # 1. no qemu_img_info passed in and qemu_img_info() raises exc
+        mock_info.side_effect = processutils.ProcessExecutionError
+        image_utils._convert_image(prefix, source, dest, out_format)
+        mock_debug_log.assert_not_called()
+        log_msg = mock_info_log.call_args.args[0]
+        self.assertIn("image size is unavailable", log_msg)
+
+        mock_info.reset_mock(side_effect=True)
+        mock_info_log.reset_mock()
+
+        # 2. no qemu_img_info passed in, returned obj has no virtual_size
+        mock_info.return_value = imageutils.QemuImgInfo()
+        image_utils._convert_image(prefix, source, dest, out_format)
+        mock_debug_log.assert_not_called()
+        log_msg = mock_info_log.call_args.args[0]
+        self.assertIn("image size is unavailable", log_msg)
+
+        mock_info.reset_mock(return_value=True)
+        mock_info_log.reset_mock()
+
+        # 3. no qemu_img_info passed in, returned obj has virtual_size
+        mock_info.return_value = imageutils.QemuImgInfo(
+            '{"virtual-size": 1073741824}', format='json')
+        image_utils._convert_image(prefix, source, dest, out_format)
+        log_msg = mock_debug_log.call_args.args[0]
+        self.assertIn("Image conversion details", log_msg)
+        log_msg = mock_info_log.call_args.args[0]
+        self.assertIn("Converted", log_msg)
+
+        mock_info.reset_mock()
+        mock_debug_log.reset_mock()
+        mock_info_log.reset_mock()
+
+        # 4. qemu_img_info passed in but without virtual_size
+        src_img_info = imageutils.QemuImgInfo()
+        image_utils._convert_image(prefix, source, dest, out_format,
+                                   src_img_info=src_img_info)
+        mock_info.assert_not_called()
+        mock_debug_log.assert_not_called()
+        log_msg = mock_info_log.call_args.args[0]
+        self.assertIn("image size is unavailable", log_msg)
+
+        mock_info_log.reset_mock()
+
+        # 5. qemu_img_info passed in with virtual_size
+        src_img_info = imageutils.QemuImgInfo('{"virtual-size": 1073741824}',
+                                              format='json')
+        image_utils._convert_image(prefix, source, dest, out_format,
+                                   src_img_info=src_img_info)
+        mock_info.assert_not_called()
+        log_msg = mock_debug_log.call_args.args[0]
+        self.assertIn("Image conversion details", log_msg)
+        log_msg = mock_info_log.call_args.args[0]
+        self.assertIn("Converted", log_msg)
+
 
 @ddt.ddt
 class TestResizeImage(test.TestCase):

cinder/tests/unit/volume/drivers/test_nfs.py

@@ -1327,17 +1327,22 @@ class NfsDriverTestCase(test.TestCase):
             dest_volume = self._simple_volume()
             src_volume = self._simple_volume()
 
+        # snapshot img_info
         fake_snap = fake_snapshot.fake_snapshot_obj(self.context)
         fake_snap.volume = src_volume
-
         img_out = qemu_img_info % {'volid': src_volume.id,
                                    'snapid': fake_snap.id,
                                    'size_gb': src_volume.size,
                                    'size_b': src_volume.size * units.Gi}
-
         img_info = imageutils.QemuImgInfo(img_out, format='json')
+
+        # backing file img_info
+        img_out = QEMU_IMG_INFO_OUT1 % {'volid': src_volume.id,
+                                        'size_b': src_volume.size * units.Gi}
+        bk_img_info = imageutils.QemuImgInfo(img_out, format='json')
+
         mock_img_info = self.mock_object(image_utils, 'qemu_img_info')
-        mock_img_info.return_value = img_info
+        mock_img_info.side_effect = [img_info, bk_img_info]
         mock_convert_image = self.mock_object(image_utils, 'convert_image')
 
         vol_dir = os.path.join(self.TEST_MNT_POINT_BASE,
@@ -1361,21 +1366,27 @@ class NfsDriverTestCase(test.TestCase):
                                        dest_encryption_key_id)
 
         mock_read_info_file.assert_called_once_with(info_path)
-        mock_img_info.assert_called_once_with(snap_path,
-                                              force_share=True,
-                                              run_as_root=True,
-                                              allow_qcow2_backing_file=True)
+        snap_info_call = mock.call(snap_path,
+                                   force_share=True, run_as_root=True,
+                                   allow_qcow2_backing_file=True)
+        src_info_call = mock.call(src_vol_path,
+                                  force_share=True, run_as_root=True,
+                                  allow_qcow2_backing_file=True)
+        self.assertEqual(2, mock_img_info.call_count)
+        mock_img_info.assert_has_calls([snap_info_call, src_info_call])
         used_qcow = nfs_conf['nfs_qcow2_volumes']
         if encryption:
             mock_convert_image.assert_called_once_with(
                 src_vol_path, dest_vol_path, 'luks',
                 passphrase_file='/tmp/passfile',
                 run_as_root=True,
-                src_passphrase_file='/tmp/imgfile')
+                src_passphrase_file='/tmp/imgfile',
+                data=bk_img_info)
         else:
             mock_convert_image.assert_called_once_with(
                 src_vol_path, dest_vol_path, 'qcow2' if used_qcow else 'raw',
-                run_as_root=True)
+                run_as_root=True,
+                data=bk_img_info)
         mock_permission.assert_called_once_with(dest_vol_path)
 
     @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3, 'available'],
@@ -1443,7 +1454,7 @@ class NfsDriverTestCase(test.TestCase):
         used_qcow = nfs_conf['nfs_qcow2_volumes']
         mock_convert_image.assert_called_once_with(
             src_volume_path, new_volume_path, 'qcow2' if used_qcow else 'raw',
-            run_as_root=True)
+            run_as_root=True, data=img_info)
         mock_ensure.assert_called_once()
         mock_find_share.assert_called_once_with(new_volume)
 

cinder/volume/drivers/nfs.py

@@ -642,6 +642,8 @@ class NfsDriver(remotefs.RemoteFSSnapDriverDistributed):
         # when this snapshot was created.
         img_info = self._qemu_img_info(forward_path, snapshot.volume.name)
         path_to_snap_img = os.path.join(vol_path, img_info.backing_file)
+        snap_backing_file_img_info = self._qemu_img_info(path_to_snap_img,
+                                                         snapshot.volume.name)
 
         path_to_new_vol = self._local_path_volume(volume)
 
@@ -687,11 +689,13 @@ class NfsDriver(remotefs.RemoteFSSnapDriverDistributed):
                         'luks',
                         passphrase_file=new_pass_file.name,
                         src_passphrase_file=src_pass_file.name,
-                        run_as_root=self._execute_as_root)
+                        run_as_root=self._execute_as_root,
+                        data=snap_backing_file_img_info)
         else:
             image_utils.convert_image(path_to_snap_img,
                                       path_to_new_vol,
                                       out_format,
-                                      run_as_root=self._execute_as_root)
+                                      run_as_root=self._execute_as_root,
+                                      data=snap_backing_file_img_info)
 
         self._set_rw_permissions_for_all(path_to_new_vol)

releasenotes/notes/fix-nfs-vol-from-snapshot-654a07d25a33bf7d.yaml

@@ -0,0 +1,8 @@
+---
+fixes:
+  - |
+    NFS driver `bug #2074377
+    <https://bugs.launchpad.net/cinder/+bug/2074377>`_: Fixed regression
+    caused by change I65857288b797 (the mitigation for CVE-2024-32498)
+    that was preventing the creation of a new volume from the second and
+    subsequent snapshots of an existing volume.