Merge "Refactor use of encryption/image volume utils"

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

@@ -2356,8 +2356,7 @@ class RBDTestCase(test.TestCase):
             self.assertEqual((True, None), ret)
 
     @mock.patch('tempfile.NamedTemporaryFile')
-    @mock.patch('cinder.volume.drivers.rbd.RBDDriver.'
-                '_check_encryption_provider',
+    @mock.patch('cinder.volume.utils.check_encryption_provider',
                 return_value={'encryption_key_id': fake.ENCRYPTION_KEY_ID})
     def test_create_encrypted_volume(self,
                                      mock_check_enc_prov,
@@ -2381,8 +2380,8 @@ class RBDTestCase(test.TestCase):
                     'cipher': 'aes-xts-essiv',
                     'key_size': 256}
 
-        with mock.patch('cinder.volume.drivers.rbd.RBDDriver.'
-                        '_check_encryption_provider', return_value=enc_info), \
+        with mock.patch('cinder.volume.utils.check_encryption_provider',
+                        return_value=enc_info), \
                 mock.patch('cinder.volume.drivers.rbd.open') as mock_open, \
                 mock.patch.object(self.driver, '_execute') as mock_exec:
             self.driver._create_encrypted_volume(self.volume_c,

cinder/volume/drivers/rbd.py

@@ -23,7 +23,6 @@ import tempfile
 
 from castellan import key_manager
 from eventlet import tpool
-from os_brick import encryptors
 from os_brick.initiator import linuxrbd
 from oslo_config import cfg
 from oslo_log import log as logging
@@ -865,27 +864,6 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
 
         return volume_update
 
-    def _check_encryption_provider(self, volume, context):
-        """Check that this is a LUKS encryption provider.
-
-        :returns: encryption dict
-        """
-
-        encryption = self.db.volume_encryption_metadata_get(context, volume.id)
-        provider = encryption['provider']
-        if provider in encryptors.LEGACY_PROVIDER_CLASS_TO_FORMAT_MAP:
-            provider = encryptors.LEGACY_PROVIDER_CLASS_TO_FORMAT_MAP[provider]
-        if provider != encryptors.LUKS:
-            message = _("Provider %s not supported.") % provider
-            raise exception.VolumeDriverException(message=message)
-
-        if 'cipher' not in encryption or 'key_size' not in encryption:
-            msg = _('encryption spec must contain "cipher" and '
-                    '"key_size"')
-            raise exception.VolumeDriverException(message=msg)
-
-        return encryption
-
     def _create_encrypted_volume(self, volume, context):
         """Create an encrypted volume.
 
@@ -893,7 +871,9 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
         and then uploading it to the volume.
         """
 
-        encryption = self._check_encryption_provider(volume, context)
+        encryption = volume_utils.check_encryption_provider(self.db,
+                                                            volume,
+                                                            context)
 
         # Fetch the key associated with the volume and decode the passphrase
         keymgr = key_manager.API(CONF)
@@ -901,7 +881,7 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
         passphrase = binascii.hexlify(key.get_encoded()).decode('utf-8')
 
         # create a file
-        tmp_dir = self._image_conversion_dir()
+        tmp_dir = volume_utils.image_conversion_dir()
 
         with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp_image:
             with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp_key:
@@ -1524,16 +1504,6 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
                     return volume_update, True
         return ({}, False)
 
-    def _image_conversion_dir(self):
-        tmpdir = (CONF.image_conversion_dir or
-                  tempfile.gettempdir())
-
-        # ensure temporary directory exists
-        if not os.path.exists(tmpdir):
-            os.makedirs(tmpdir)
-
-        return tmpdir
-
     def copy_image_to_encrypted_volume(self, context, volume, image_service,
                                        image_id):
         self._copy_image_to_volume(context, volume, image_service, image_id,
@@ -1543,7 +1513,10 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
         self._copy_image_to_volume(context, volume, image_service, image_id)
 
     def _encrypt_image(self, context, volume, tmp_dir, src_image_path):
-        encryption = self._check_encryption_provider(volume, context)
+        encryption = volume_utils.check_encryption_provider(
+            self.db,
+            volume,
+            context)
 
         # Fetch the key associated with the volume and decode the passphrase
         keymgr = key_manager.API(CONF)
@@ -1554,7 +1527,7 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
         cipher_spec = image_utils.decode_cipher(encryption['cipher'],
                                                 encryption['key_size'])
 
-        tmp_dir = self._image_conversion_dir()
+        tmp_dir = volume_utils.image_conversion_dir()
 
         with tempfile.NamedTemporaryFile(prefix='luks_',
                                          dir=tmp_dir) as pass_file:
@@ -1574,7 +1547,7 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
     def _copy_image_to_volume(self, context, volume, image_service, image_id,
                               encrypted=False):
 
-        tmp_dir = self._image_conversion_dir()
+        tmp_dir = volume_utils.image_conversion_dir()
 
         with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp:
             image_utils.fetch_to_raw(context, image_service, image_id,
@@ -1609,7 +1582,7 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
                                              volume_id=volume.id)
 
     def copy_volume_to_image(self, context, volume, image_service, image_meta):
-        tmp_dir = self._image_conversion_dir()
+        tmp_dir = volume_utils.image_conversion_dir()
         tmp_file = os.path.join(tmp_dir,
                                 volume.name + '-' + image_meta['id'])
         with fileutils.remove_path_on_error(tmp_file):

cinder/volume/utils.py

@@ -1176,7 +1176,7 @@ def copy_image_to_volume(driver, context, volume, image_meta, image_location,
                'image_location': image_location})
 
 
-def _image_conversion_dir():
+def image_conversion_dir():
     tmpdir = (CONF.image_conversion_dir or
               tempfile.gettempdir())