Merge "Enforce image safety during image_conversion" into stable/ussuri

glance/async_/flows/plugins/image_conversion.py

@@ -105,6 +105,29 @@ class _ConvertImage(task.Task):
         image = self.image_repo.get(self.image_id)
         image.virtual_size = virtual_size
 
+        if 'backing-filename' in metadata:
+            LOG.warning('Refusing to process QCOW image with a backing file')
+            raise RuntimeError(
+                'QCOW images with backing files are not allowed')
+
+        if metadata.get('format') == 'vmdk':
+            create_type = metadata.get(
+                'format-specific', {}).get(
+                    'data', {}).get('create-type')
+            allowed = CONF.image_format.vmdk_allowed_types
+            if not create_type:
+                raise RuntimeError(_('Unable to determine VMDK create-type'))
+            if not len(allowed):
+                LOG.warning(_('Refusing to process VMDK file as '
+                              'vmdk_allowed_types is empty'))
+                raise RuntimeError(_('Image is a VMDK, but no VMDK createType '
+                                     'is specified'))
+            if create_type not in allowed:
+                LOG.warning(_('Refusing to process VMDK file with create-type '
+                              'of %r which is not in allowed set of: %s'),
+                            create_type, ','.join(allowed))
+                raise RuntimeError(_('Invalid VMDK create-type specified'))
+
         if source_format == target_format:
             LOG.debug("Source is already in target format, "
                       "not doing conversion for %s", self.image_id)

glance/common/config.py

@@ -98,6 +98,18 @@ image_format_opts = [
                        "image attribute"),
                 deprecated_opts=[cfg.DeprecatedOpt('disk_formats',
                                                    group='DEFAULT')]),
+    cfg.ListOpt('vmdk_allowed_types',
+                default=['streamOptimized', 'monolithicSparse'],
+                help=_("A list of strings describing allowed VMDK "
+                       "'create-type' subformats that will be allowed. "
+                       "This is recommended to only include "
+                       "single-file-with-sparse-header variants to avoid "
+                       "potential host file exposure due to processing named "
+                       "extents. If this list is empty, then no VDMK image "
+                       "types allowed. Note that this is currently only "
+                       "checked during image conversion (if enabled), and "
+                       "limits the types of VMDK images we will convert "
+                       "from.")),
 ]
 task_opts = [
     cfg.IntOpt('task_time_to_live',

glance/tests/unit/async_/flows/plugins/test_image_conversion.py

@@ -105,6 +105,68 @@ class TestConvertImageTask(test_utils.BaseTestCase):
                 self.assertIn('-f', exc_mock.call_args[0])
                 self.assertEqual("qcow2", image.disk_format)
 
+    def _setup_image_convert_info_fail(self):
+        image_convert = image_conversion._ConvertImage(self.context,
+                                                       self.task.task_id,
+                                                       self.task_type,
+                                                       self.img_repo,
+                                                       self.image_id)
+
+        self.task_repo.get.return_value = self.task
+        image = mock.MagicMock(image_id=self.image_id, virtual_size=None,
+                               extra_properties={
+                                   'os_glance_import_task': self.task.task_id},
+                               disk_format='qcow2')
+        self.img_repo.get.return_value = image
+        return image_convert
+
+    def test_image_convert_invalid_qcow(self):
+        data = {'format': 'qcow2',
+                'backing-filename': '/etc/hosts'}
+
+        convert = self._setup_image_convert_info_fail()
+        with mock.patch.object(processutils, 'execute') as exc_mock:
+            exc_mock.return_value = json.dumps(data), ''
+            e = self.assertRaises(RuntimeError,
+                                  convert.execute, 'file:///test/path.qcow')
+            self.assertEqual('QCOW images with backing files are not allowed',
+                             str(e))
+
+    def _test_image_convert_invalid_vmdk(self):
+        data = {'format': 'vmdk',
+                'format-specific': {
+                    'data': {
+                        'create-type': 'monolithicFlat',
+                    }}}
+
+        convert = self._setup_image_convert_info_fail()
+        with mock.patch.object(processutils, 'execute') as exc_mock:
+            exc_mock.return_value = json.dumps(data), ''
+            convert.execute('file:///test/path.vmdk')
+
+    def test_image_convert_invalid_vmdk(self):
+        e = self.assertRaises(RuntimeError,
+                              self._test_image_convert_invalid_vmdk)
+        self.assertEqual('Invalid VMDK create-type specified', str(e))
+
+    def test_image_convert_valid_vmdk_no_types(self):
+        with mock.patch.object(CONF.image_format, 'vmdk_allowed_types',
+                               new=[]):
+            # We make it past the VMDK check and fail because our file
+            # does not exist
+            e = self.assertRaises(RuntimeError,
+                                  self._test_image_convert_invalid_vmdk)
+            self.assertEqual('Image is a VMDK, but no VMDK createType is '
+                             'specified', str(e))
+
+    def test_image_convert_valid_vmdk(self):
+        with mock.patch.object(CONF.image_format, 'vmdk_allowed_types',
+                               new=['monolithicSparse', 'monolithicFlat']):
+            # We make it past the VMDK check and fail because our file
+            # does not exist
+            self.assertRaises(FileNotFoundError,
+                              self._test_image_convert_invalid_vmdk)
+
     @mock.patch.object(os, 'remove')
     def test_image_convert_revert_success(self, mock_os_remove):
         mock_os_remove.return_value = None