NFS encrypted volume support

Volume encryption helps provide basic data protection in
case the volume back-end is either compromised or outright
stolen. The contents of an encrypted volume can only be
read with the use of a specific key;

Volume encryption: Check volume encryption key in
'_do_create_volume' method (remotefs) and use
'_create_encrypted_volume_file' when encryption is
required.

Snapshot encryption: To create an encrypted volume from a
snapshot, we need to pass the Barbican key of the
snapshot.volume to 'convert_image' method. Because of this
I've added 'src_passphrase_file' parameter.

This patch doesn't handle encrypted volume -> unencrypted
Current error prompted: 'Invalid input received: Invalid
volume_type provided: aeac5517-6bc8-4b59-9eb2-76e84369bd0
(requested type is not compatible; recommend omitting the
type argument). (HTTP 400)'

Implements: blueprint nfs-volume-encryption
Co-Authored-By: Sofia Enriquez <lsofia.enriquez@gmail.com>

Change-Id: I896f70d204ad103e968ab242ba9045ca984827c4

cinder/image/image_utils.py

@@ -164,13 +164,63 @@ def get_qemu_img_version():
 
 
 def qemu_img_supports_force_share():
-    return get_qemu_img_version() > [2, 10, 0]
+    return get_qemu_img_version() >= [2, 10, 0]
+
+
+def _get_qemu_convert_luks_cmd(src, dest, out_format, src_format=None,
+                               out_subformat=None, cache_mode=None,
+                               prefix=None, cipher_spec=None,
+                               passphrase_file=None,
+                               src_passphrase_file=None):
+
+    cmd = ['qemu-img', 'convert']
+
+    if prefix:
+        cmd = list(prefix) + cmd
+
+    if cache_mode:
+        cmd += ('-t', cache_mode)
+
+    obj1 = ['--object',
+            'secret,id=sec1,format=raw,file=%s' % src_passphrase_file]
+    obj2 = ['--object',
+            'secret,id=sec2,format=raw,file=%s' % passphrase_file]
+
+    src_opts = 'encrypt.format=luks,encrypt.key-secret=sec1,' \
+               'file.filename=%s' % src
+
+    image_opts = ['--image-opts', src_opts]
+    output_opts = ['-O', 'luks', '-o', 'key-secret=sec2', dest]
+
+    command = cmd + obj1 + obj2 + image_opts + output_opts
+    return command
 
 
 def _get_qemu_convert_cmd(src, dest, out_format, src_format=None,
                           out_subformat=None, cache_mode=None,
                           prefix=None, cipher_spec=None,
-                          passphrase_file=None, compress=False):
+                          passphrase_file=None, compress=False,
+                          src_passphrase_file=None):
+
+    if src_passphrase_file is not None:
+        if passphrase_file is None:
+            message = _("Can't create unencrypted volume %(format)s "
+                        "from an encrypted source volume."
+                        ) % {'format': out_format}
+            LOG.error(message)
+            # TODO(enriquetaso): handle encrypted->unencrypted
+            raise exception.NotSupportedOperation(operation=message)
+        return _get_qemu_convert_luks_cmd(
+            src,
+            dest,
+            out_format,
+            src_format=src_format,
+            out_subformat=out_subformat,
+            cache_mode=cache_mode,
+            prefix=None,
+            cipher_spec=cipher_spec,
+            passphrase_file=passphrase_file,
+            src_passphrase_file=src_passphrase_file)
 
     if out_format == 'vhd':
         # qemu-img still uses the legacy vpc name
@@ -240,7 +290,8 @@ def check_qemu_img_version(minimum_version):
 def _convert_image(prefix, source, dest, out_format,
                    out_subformat=None, src_format=None,
                    run_as_root=True, cipher_spec=None,
-                   passphrase_file=None, compress=False):
+                   passphrase_file=None, compress=False,
+                   src_passphrase_file=None):
     """Convert image to other format.
 
     :param prefix: command prefix, i.e. cgexec for throttling
@@ -253,6 +304,8 @@ def _convert_image(prefix, source, dest, out_format,
     :param cipher_spec: encryption details
     :param passphrase_file: filename containing luks passphrase
     :param compress: compress w/ qemu-img when possible (best effort)
+    :param src_passphrase_file: filename containing source volume's
+                                luks passphrase
     """
 
     # Check whether O_DIRECT is supported and set '-t none' if it is
@@ -281,7 +334,8 @@ def _convert_image(prefix, source, dest, out_format,
                                 prefix=prefix,
                                 cipher_spec=cipher_spec,
                                 passphrase_file=passphrase_file,
-                                compress=compress)
+                                compress=compress,
+                                src_passphrase_file=src_passphrase_file)
 
     start_time = timeutils.utcnow()
 
@@ -333,7 +387,7 @@ def _convert_image(prefix, source, dest, out_format,
 def convert_image(source, dest, out_format, out_subformat=None,
                   src_format=None, run_as_root=True, throttle=None,
                   cipher_spec=None, passphrase_file=None,
-                  compress=False):
+                  compress=False, src_passphrase_file=None):
     if not throttle:
         throttle = throttling.Throttle.get_default()
     with throttle.subcommand(source, dest) as throttle_cmd:
@@ -345,7 +399,8 @@ def convert_image(source, dest, out_format, out_subformat=None,
                        run_as_root=run_as_root,
                        cipher_spec=cipher_spec,
                        passphrase_file=passphrase_file,
-                       compress=compress)
+                       compress=compress,
+                       src_passphrase_file=src_passphrase_file)
 
 
 def resize_image(source, size, run_as_root=False):

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

@@ -18,6 +18,7 @@ import errno
 import os
 from unittest import mock
 
+import castellan
 import ddt
 from oslo_utils import imageutils
 from oslo_utils import units
@@ -28,6 +29,7 @@ from cinder.image import image_utils
 from cinder.tests.unit import fake_constants as fake
 from cinder.tests.unit import fake_snapshot
 from cinder.tests.unit import fake_volume
+from cinder.tests.unit.keymgr import fake as fake_keymgr
 from cinder.tests.unit import test
 from cinder.volume import configuration as conf
 from cinder.volume.drivers import nfs
@@ -35,6 +37,11 @@ from cinder.volume.drivers import remotefs
 from cinder.volume import volume_utils
 
 
+class KeyObject(object):
+    def get_encoded(arg):
+        return "asdf".encode('utf-8')
+
+
 class RemoteFsDriverTestCase(test.TestCase):
     TEST_FILE_NAME = 'test.txt'
     TEST_EXPORT = 'nas-host1:/export'
@@ -374,6 +381,57 @@ Format specific information:
     corrupt: false
     """
 
+QEMU_IMG_INFO_OUT5 = """image: volume-%(volid)s.%(snapid)s
+file format: qcow2
+virtual size: %(size_gb)sG (%(size_b)s bytes)
+disk size: 196K
+encrypted: yes
+cluster_size: 65536
+backing file: volume-%(volid)s
+backing file format: raw
+Format specific information:
+    compat: 1.1
+    lazy refcounts: false
+    refcount bits: 16
+    encrypt:
+        ivgen alg: plain64
+        hash alg: sha256
+        cipher alg: aes-256
+        uuid: 386f8626-33f0-4683-a517-78ddfe385e33
+        format: luks
+        cipher mode: xts
+        slots:
+            [0]:
+                active: true
+                iters: 1892498
+                key offset: 4096
+                stripes: 4000
+            [1]:
+                active: false
+                key offset: 262144
+            [2]:
+                active: false
+                key offset: 520192
+            [3]:
+                active: false
+                key offset: 778240
+            [4]:
+                active: false
+                key offset: 1036288
+            [5]:
+                active: false
+                key offset: 1294336
+            [6]:
+                active: false
+                key offset: 1552384
+            [7]:
+                active: false
+                key offset: 1810432
+        payload offset: 2068480
+        master key iters: 459347
+    corrupt: false
+"""
+
 
 @ddt.ddt
 class NfsDriverTestCase(test.TestCase):
@@ -691,6 +749,17 @@ class NfsDriverTestCase(test.TestCase):
                                            provider_location=loc,
                                            size=size)
 
+    def _simple_encrypted_volume(self, size=10):
+        loc = self.TEST_NFS_EXPORT1
+        info_dic = {'name': u'volume-0000000a',
+                    'id': '55555555-222f-4b32-b585-9991b3bf0a99',
+                    'size': size,
+                    'encryption_key_id': fake.ENCRYPTION_KEY_ID}
+
+        return fake_volume.fake_volume_obj(self.context,
+                                           provider_location=loc,
+                                           **info_dic)
+
     def test_get_provisioned_capacity(self):
         self._set_driver()
         drv = self._driver
@@ -1124,14 +1193,15 @@ class NfsDriverTestCase(test.TestCase):
         """Case where the mount works the first time."""
 
         self._set_driver()
-        self.mock_object(self._driver._remotefsclient, 'mount')
+        self.mock_object(self._driver._remotefsclient, 'mount', autospec=True)
         drv = self._driver
         drv.configuration.nfs_mount_attempts = 3
         drv.shares = {self.TEST_NFS_EXPORT1: ''}
 
         drv._ensure_share_mounted(self.TEST_NFS_EXPORT1)
 
-        drv._remotefsclient.mount.called_once()
+        drv._remotefsclient.mount.assert_called_once_with(
+            self.TEST_NFS_EXPORT1, [])
 
     @mock.patch('time.sleep')
     def test_ensure_share_mounted_exception(self, _mock_sleep):
@@ -1169,16 +1239,48 @@ class NfsDriverTestCase(test.TestCase):
         self.assertEqual(min_num_attempts,
                          drv._remotefsclient.mount.call_count)
 
-    @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3],
+    @mock.patch('tempfile.NamedTemporaryFile')
-              [NFS_CONFIG2, QEMU_IMG_INFO_OUT4],
+    @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3, False],
-              [NFS_CONFIG3, QEMU_IMG_INFO_OUT3],
+              [NFS_CONFIG2, QEMU_IMG_INFO_OUT4, False],
-              [NFS_CONFIG4, QEMU_IMG_INFO_OUT4])
+              [NFS_CONFIG3, QEMU_IMG_INFO_OUT3, False],
+              [NFS_CONFIG4, QEMU_IMG_INFO_OUT4, False],
+              [NFS_CONFIG4, QEMU_IMG_INFO_OUT5, True])
     @ddt.unpack
-    def test_copy_volume_from_snapshot(self, nfs_conf, qemu_img_info):
+    def test_copy_volume_from_snapshot(self, nfs_conf, qemu_img_info,
+                                       encryption, mock_temp_file):
+
+        class DictObj(object):
+            # convert a dict to object w/ attributes
+            def __init__(self, d):
+                self.__dict__ = d
+
         self._set_driver(extra_confs=nfs_conf)
         drv = self._driver
-        dest_volume = self._simple_volume()
+
-        src_volume = self._simple_volume()
+        src_encryption_key_id = None
+        dest_encryption_key_id = None
+
+        if encryption:
+            mock_temp_file.return_value.__enter__.side_effect = [
+                DictObj({'name': '/tmp/imgfile'}),
+                DictObj({'name': '/tmp/passfile'})]
+
+            dest_volume = self._simple_encrypted_volume()
+            src_volume = self._simple_encrypted_volume()
+
+            key_mgr = fake_keymgr.fake_api()
+            self.mock_object(castellan.key_manager, 'API',
+                             return_value=key_mgr)
+            key_id = key_mgr.store(self.context, KeyObject())
+
+            src_volume.encryption_key_id = key_id
+            dest_volume.encryption_key_id = key_id
+
+            src_encryption_key_id = src_volume.encryption_key_id
+            dest_encryption_key_id = dest_volume.encryption_key_id
+        else:
+            dest_volume = self._simple_volume()
+            src_volume = self._simple_volume()
 
         fake_snap = fake_snapshot.fake_snapshot_obj(self.context)
         fake_snap.volume = src_volume
@@ -1209,16 +1311,25 @@ class NfsDriverTestCase(test.TestCase):
 
         mock_permission = self.mock_object(drv, '_set_rw_permissions_for_all')
 
-        drv._copy_volume_from_snapshot(fake_snap, dest_volume, size)
+        drv._copy_volume_from_snapshot(fake_snap, dest_volume, size,
+                                       src_encryption_key_id,
+                                       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)
         used_qcow = nfs_conf['nfs_qcow2_volumes']
-        mock_convert_image.assert_called_once_with(
+        if encryption:
-            src_vol_path, dest_vol_path, 'qcow2' if used_qcow else 'raw',
+            mock_convert_image.assert_called_once_with(
-            run_as_root=True)
+                src_vol_path, dest_vol_path, 'luks',
+                passphrase_file='/tmp/passfile',
+                run_as_root=True,
+                src_passphrase_file='/tmp/imgfile')
+        else:
+            mock_convert_image.assert_called_once_with(
+                src_vol_path, dest_vol_path, 'qcow2' if used_qcow else 'raw',
+                run_as_root=True)
         mock_permission.assert_called_once_with(dest_vol_path)
 
     @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3],

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

@@ -19,20 +19,28 @@ import re
 import sys
 from unittest import mock
 
+import castellan
 import ddt
 
 from cinder import context
 from cinder import exception
 from cinder.image import image_utils
 from cinder.objects import fields
+from cinder.tests.unit import fake_constants as fake
 from cinder.tests.unit import fake_snapshot
 from cinder.tests.unit import fake_volume
+from cinder.tests.unit.keymgr import fake as fake_keymgr
 from cinder.tests.unit import test
 from cinder import utils
 from cinder.volume.drivers import remotefs
 from cinder.volume import volume_utils
 
 
+class KeyObject(object):
+    def get_encoded(arg):
+        return "asdf".encode('utf-8')
+
+
 @ddt.ddt
 class RemoteFsSnapDriverTestCase(test.TestCase):
 
@@ -56,6 +64,20 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
                                     self._fake_snapshot.id)
         self._fake_snapshot.volume = self._fake_volume
 
+        # Encrypted volume and snapshot
+        self.volume_c = fake_volume.fake_volume_obj(
+            self.context,
+            **{'name': u'volume-0000000a',
+               'id': '55555555-222f-4b32-b585-9991b3bf0a99',
+               'size': 12,
+               'encryption_key_id': fake.ENCRYPTION_KEY_ID})
+        self._fake_snap_c = fake_snapshot.fake_snapshot_obj(self.context)
+        self._fake_snap_c.volume = self.volume_c
+        self.volume_c_path = os.path.join(self._FAKE_MNT_POINT,
+                                          self._fake_snap_c.name)
+        self._fake_snap_c_path = (self.volume_c_path + '.' +
+                                  self._fake_snap_c.id)
+
     @ddt.data({'current_state': 'in-use',
                'acceptable_states': ['available', 'in-use']},
               {'current_state': 'in-use',
@@ -75,19 +97,43 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
     def _test_delete_snapshot(self, volume_in_use=False,
                               stale_snapshot=False,
                               is_active_image=True,
-                              is_tmp_snap=False):
+                              is_tmp_snap=False,
+                              encryption=False):
         # If the snapshot is not the active image, it is guaranteed that
         # another snapshot exists having it as backing file.
 
-        fake_snapshot_name = os.path.basename(self._fake_snapshot_path)
+        fake_upper_snap_id = 'fake_upper_snap_id'
-        fake_info = {'active': fake_snapshot_name,
+        if encryption:
-                     self._fake_snapshot.id: fake_snapshot_name}
+            fake_snapshot_name = os.path.basename(self._fake_snap_c_path)
+            fake_info = {'active': fake_snapshot_name,
+                         self._fake_snap_c.id: fake_snapshot_name}
+            expected_info = fake_info
+
+            fake_upper_snap_path = (
+                self.volume_c_path + '-snapshot' + fake_upper_snap_id)
+
+            snapshot = self._fake_snap_c
+            snapshot_path = self._fake_snap_c_path
+            volume_name = self.volume_c.name
+        else:
+            fake_snapshot_name = os.path.basename(self._fake_snapshot_path)
+            fake_info = {'active': fake_snapshot_name,
+                         self._fake_snapshot.id: fake_snapshot_name}
+            expected_info = fake_info
+
+            fake_upper_snap_path = (
+                self._fake_volume_path + '-snapshot' + fake_upper_snap_id)
+
+            snapshot = self._fake_snapshot
+            snapshot_path = self._fake_snapshot_path
+            volume_name = self._fake_volume.name
+
         fake_snap_img_info = mock.Mock()
         fake_base_img_info = mock.Mock()
         if stale_snapshot:
             fake_snap_img_info.backing_file = None
         else:
-            fake_snap_img_info.backing_file = self._fake_volume.name
+            fake_snap_img_info.backing_file = volume_name
         fake_snap_img_info.file_format = 'qcow2'
         fake_base_img_info.backing_file = None
         fake_base_img_info.file_format = 'raw'
@@ -107,61 +153,53 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         self._driver._delete_stale_snapshot = mock.Mock()
         self._driver._delete_snapshot_online = mock.Mock()
 
-        expected_info = {
-            'active': fake_snapshot_name,
-            self._fake_snapshot.id: fake_snapshot_name
-        }
-
         exp_acceptable_states = ['available', 'in-use', 'backing-up',
                                  'deleting', 'downloading']
 
         if volume_in_use:
-            self._fake_snapshot.volume.status = 'backing-up'
+            snapshot.volume.status = 'backing-up'
-            self._fake_snapshot.volume.attach_status = 'attached'
+            snapshot.volume.attach_status = 'attached'
 
             self._driver._read_info_file.return_value = fake_info
 
-            self._driver._delete_snapshot(self._fake_snapshot)
+            self._driver._delete_snapshot(snapshot)
             self._driver._validate_state.assert_called_once_with(
-                self._fake_snapshot.volume.status,
+                snapshot.volume.status,
                 exp_acceptable_states)
             if stale_snapshot:
                 self._driver._delete_stale_snapshot.assert_called_once_with(
-                    self._fake_snapshot)
+                    snapshot)
             else:
                 expected_online_delete_info = {
                     'active_file': fake_snapshot_name,
                     'snapshot_file': fake_snapshot_name,
-                    'base_file': self._fake_volume.name,
+                    'base_file': volume_name,
                     'base_id': None,
                     'new_base_file': None
                 }
                 self._driver._delete_snapshot_online.assert_called_once_with(
-                    self.context, self._fake_snapshot,
+                    self.context, snapshot,
                     expected_online_delete_info)
 
         elif is_active_image:
             self._driver._read_info_file.return_value = fake_info
 
-            self._driver._delete_snapshot(self._fake_snapshot)
+            self._driver._delete_snapshot(snapshot)
 
             self._driver._img_commit.assert_called_once_with(
-                self._fake_snapshot_path)
+                snapshot_path)
-            self.assertNotIn(self._fake_snapshot.id, fake_info)
+            self.assertNotIn(snapshot.id, fake_info)
             self._driver._write_info_file.assert_called_once_with(
                 mock.sentinel.fake_info_path, fake_info)
         else:
-            fake_upper_snap_id = 'fake_upper_snap_id'
-            fake_upper_snap_path = (
-                self._fake_volume_path + '-snapshot' + fake_upper_snap_id)
             fake_upper_snap_name = os.path.basename(fake_upper_snap_path)
 
             fake_backing_chain = [
                 {'filename': fake_upper_snap_name,
                  'backing-filename': fake_snapshot_name},
                 {'filename': fake_snapshot_name,
-                 'backing-filename': self._fake_volume.name},
+                 'backing-filename': volume_name},
-                {'filename': self._fake_volume.name,
+                {'filename': volume_name,
                  'backing-filename': None}]
 
             fake_info[fake_upper_snap_id] = fake_upper_snap_name
@@ -169,42 +207,62 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
             fake_info['active'] = fake_upper_snap_name
 
             expected_info = copy.deepcopy(fake_info)
-            del expected_info[self._fake_snapshot.id]
+            del expected_info[snapshot.id]
 
             self._driver._read_info_file.return_value = fake_info
             self._driver._get_backing_chain_for_path = mock.Mock(
                 return_value=fake_backing_chain)
 
-            self._driver._delete_snapshot(self._fake_snapshot)
+            self._driver._delete_snapshot(snapshot)
 
             self._driver._img_commit.assert_called_once_with(
-                self._fake_snapshot_path)
+                snapshot_path)
             self._driver._rebase_img.assert_called_once_with(
-                fake_upper_snap_path, self._fake_volume.name,
+                fake_upper_snap_path, volume_name,
                 fake_base_img_info.file_format)
             self._driver._write_info_file.assert_called_once_with(
                 mock.sentinel.fake_info_path, expected_info)
 
-    def test_delete_snapshot_when_active_file(self):
+    @ddt.data({'encryption': True}, {'encryption': False})
-        self._test_delete_snapshot()
+    def test_delete_snapshot_when_active_file(self, encryption):
+        self._test_delete_snapshot(encryption=encryption)
 
-    def test_delete_snapshot_in_use(self):
+    @ddt.data({'encryption': True}, {'encryption': False})
-        self._test_delete_snapshot(volume_in_use=True)
+    def test_delete_snapshot_in_use(self, encryption):
-
-    def test_delete_snapshot_in_use_stale_snapshot(self):
         self._test_delete_snapshot(volume_in_use=True,
-                                   stale_snapshot=True)
+                                   encryption=encryption)
 
-    def test_delete_snapshot_with_one_upper_file(self):
+    @ddt.data({'encryption': True}, {'encryption': False})
-        self._test_delete_snapshot(is_active_image=False)
+    def test_delete_snapshot_in_use_stale_snapshot(self,
+                                                   encryption):
+        self._test_delete_snapshot(volume_in_use=True,
+                                   stale_snapshot=True,
+                                   encryption=encryption)
+
+    @ddt.data({'encryption': True}, {'encryption': False})
+    def test_delete_snapshot_with_one_upper_file(self,
+                                                 encryption):
+        self._test_delete_snapshot(is_active_image=False,
+                                   encryption=encryption)
+
+    @ddt.data({'encryption': True}, {'encryption': False})
+    def test_delete_stale_snapshot(self, encryption):
+        if encryption:
+            fake_snapshot_name = os.path.basename(self._fake_snap_c_path)
+            volume_name = self.volume_c.name
+            snapshot = self._fake_snap_c
+            snapshot_path = self._fake_snap_c_path
+        else:
+            fake_snapshot_name = os.path.basename(self._fake_snapshot_path)
+            volume_name = self._fake_volume.name
+            snapshot = self._fake_snapshot
+            snapshot_path = self._fake_snapshot_path
 
-    def test_delete_stale_snapshot(self):
-        fake_snapshot_name = os.path.basename(self._fake_snapshot_path)
         fake_snap_info = {
-            'active': self._fake_volume.name,
+            'active': volume_name,
-            self._fake_snapshot.id: fake_snapshot_name
+            snapshot.id: fake_snapshot_name
         }
-        expected_info = {'active': self._fake_volume.name}
+        expected_info = {'active': volume_name}
 
         self._driver._local_path_volume_info = mock.Mock(
             return_value=mock.sentinel.fake_info_path)
@@ -214,9 +272,9 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
             return_value=self._FAKE_MNT_POINT)
         self._driver._write_info_file = mock.Mock()
 
-        self._driver._delete_stale_snapshot(self._fake_snapshot)
+        self._driver._delete_stale_snapshot(snapshot)
 
-        self._driver._delete.assert_called_once_with(self._fake_snapshot_path)
+        self._driver._delete.assert_called_once_with(snapshot_path)
         self._driver._write_info_file.assert_called_once_with(
             mock.sentinel.fake_info_path, expected_info)
 
@@ -256,9 +314,20 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
                  mock.call(*command3, run_as_root=True)]
         self._driver._execute.assert_has_calls(calls)
 
-    def _test_create_snapshot(self, volume_in_use=False, tmp_snap=False):
+    def _test_create_snapshot(self, volume_in_use=False, tmp_snap=False,
+                              encryption=False):
         fake_snapshot_info = {}
-        fake_snapshot_file_name = os.path.basename(self._fake_snapshot_path)
+        if encryption:
+            fake_snapshot_file_name = os.path.basename(self._fake_snap_c_path)
+            volume_name = self.volume_c.name
+            snapshot = self._fake_snap_c
+            snapshot_path = self._fake_snap_c_path
+        else:
+            fake_snapshot_file_name = os.path.basename(
+                self._fake_snapshot_path)
+            volume_name = self._fake_volume.name
+            snapshot = self._fake_snapshot
+            snapshot_path = self._fake_snapshot_path
 
         self._driver._local_path_volume_info = mock.Mock(
             return_value=mock.sentinel.fake_info_path)
@@ -268,14 +337,14 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         self._driver._create_snapshot_online = mock.Mock()
         self._driver._write_info_file = mock.Mock()
         self._driver.get_active_image_from_info = mock.Mock(
-            return_value=self._fake_volume.name)
+            return_value=volume_name)
         self._driver._get_new_snap_path = mock.Mock(
-            return_value=self._fake_snapshot_path)
+            return_value=snapshot_path)
         self._driver._validate_state = mock.Mock()
 
         expected_snapshot_info = {
             'active': fake_snapshot_file_name,
-            self._fake_snapshot.id: fake_snapshot_file_name
+            snapshot.id: fake_snapshot_file_name
         }
         exp_acceptable_states = ['available', 'in-use', 'backing-up']
         if tmp_snap:
@@ -285,31 +354,34 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
             self._fake_snapshot.display_name = display_name
 
         if volume_in_use:
-            self._fake_snapshot.volume.status = 'backing-up'
+            snapshot.volume.status = 'backing-up'
-            self._fake_snapshot.volume.attach_status = 'attached'
+            snapshot.volume.attach_status = 'attached'
             expected_method_called = '_create_snapshot_online'
         else:
-            self._fake_snapshot.volume.status = 'available'
+            snapshot.volume.status = 'available'
             expected_method_called = '_do_create_snapshot'
 
-        self._driver._create_snapshot(self._fake_snapshot)
+        self._driver._create_snapshot(snapshot)
 
         self._driver._validate_state.assert_called_once_with(
-            self._fake_snapshot.volume.status,
+            snapshot.volume.status,
             exp_acceptable_states)
         fake_method = getattr(self._driver, expected_method_called)
         fake_method.assert_called_with(
-            self._fake_snapshot, self._fake_volume.name,
+            snapshot, volume_name,
-            self._fake_snapshot_path)
+            snapshot_path)
         self._driver._write_info_file.assert_called_with(
             mock.sentinel.fake_info_path,
             expected_snapshot_info)
 
-    def test_create_snapshot_volume_available(self):
+    @ddt.data({'encryption': True}, {'encryption': False})
-        self._test_create_snapshot()
+    def test_create_snapshot_volume_available(self, encryption):
+        self._test_create_snapshot(encryption=encryption)
 
-    def test_create_snapshot_volume_in_use(self):
+    @ddt.data({'encryption': True}, {'encryption': False})
-        self._test_create_snapshot(volume_in_use=True)
+    def test_create_snapshot_volume_in_use(self, encryption):
+        self._test_create_snapshot(volume_in_use=True,
+                                   encryption=encryption)
 
     def test_create_snapshot_invalid_volume(self):
         self._fake_snapshot.volume.status = 'error'
@@ -624,14 +696,15 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         src_vref = fake_volume.fake_volume_obj(
             self.context,
             id=src_vref_id,
-            name='volume-%s' % src_vref_id)
+            name='volume-%s' % src_vref_id,
+            obj_context=self.context)
         src_vref.context = self.context
 
         mock_snapshots_exist.return_value = snapshots_exist
         drv._always_use_temp_snap_when_cloning = force_temp_snap
 
         vol_attrs = ['provider_location', 'size', 'id', 'name', 'status',
-                     'volume_type', 'metadata']
+                     'volume_type', 'metadata', 'obj_context']
         Volume = collections.namedtuple('Volume', vol_attrs)
 
         volume_ref = Volume(id=volume.id,
@@ -640,7 +713,8 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
                             provider_location=volume.provider_location,
                             status=volume.status,
                             size=volume.size,
-                            volume_type=volume.volume_type,)
+                            volume_type=volume.volume_type,
+                            obj_context=self.context,)
 
         snap_args_creation = {
             'volume_id': src_vref.id,
@@ -679,7 +753,8 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
             mock_create_snapshot.assert_called_once_with(
                 mock_obj_snap.return_value)
             mock_copy_volume_from_snapshot.assert_called_once_with(
-                mock_obj_snap.return_value, volume_ref, volume['size'])
+                mock_obj_snap.return_value, volume_ref, volume['size'],
+                src_encryption_key_id=None, new_encryption_key_id=None)
             mock_delete_snapshot.called_once_with(snap_args_deletion)
         else:
             self.assertFalse(mock_create_snapshot.called)
@@ -693,6 +768,47 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
                 [mock.call(src_vref), mock.call(volume_ref)])
             mock_extend_volume.assert_called_once_with(volume_ref, volume.size)
 
+    @mock.patch('tempfile.NamedTemporaryFile')
+    @mock.patch('cinder.volume.volume_utils.check_encryption_provider',
+                return_value={'encryption_key_id': fake.ENCRYPTION_KEY_ID})
+    def test_create_encrypted_volume(self,
+                                     mock_check_enc_prov,
+                                     mock_temp_file):
+        class DictObj(object):
+            # convert a dict to object w/ attributes
+            def __init__(self, d):
+                self.__dict__ = d
+
+        drv = self._driver
+
+        mock_temp_file.return_value.__enter__.side_effect = [
+            DictObj({'name': '/imgfile'}),
+            DictObj({'name': '/passfile'})]
+
+        key_mgr = fake_keymgr.fake_api()
+
+        self.mock_object(castellan.key_manager, 'API', return_value=key_mgr)
+        key_id = key_mgr.store(self.context, KeyObject())
+        self.volume_c.encryption_key_id = key_id
+
+        enc_info = {'encryption_key_id': key_id,
+                    'cipher': 'aes-xts-essiv',
+                    'key_size': 256}
+
+        remotefs_path = 'cinder.volume.drivers.remotefs.open'
+        with mock.patch('cinder.volume.volume_utils.check_encryption_provider',
+                        return_value=enc_info), \
+                mock.patch(remotefs_path) as mock_open, \
+                mock.patch.object(drv, '_execute') as mock_exec:
+
+            drv._create_encrypted_volume_file("/passfile",
+                                              self.volume_c.size,
+                                              enc_info,
+                                              self.context)
+
+            mock_open.assert_called_with('/imgfile', 'w')
+            mock_exec.assert_called()
+
     @mock.patch('shutil.copyfile')
     @mock.patch.object(remotefs.RemoteFSSnapDriver, '_set_rw_permissions')
     def test_copy_volume_image(self, mock_set_perm, mock_copyfile):

cinder/volume/drivers/nfs.py

@@ -14,10 +14,13 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import binascii
 import errno
 import os
+import tempfile
 import time
 
+from castellan import key_manager
 from os_brick.remotefs import remotefs as remotefs_brick
 from oslo_concurrency import processutils as putils
 from oslo_config import cfg
@@ -122,6 +125,8 @@ class NfsDriver(remotefs.RemoteFSSnapDriverDistributed):
                 self.configuration.max_over_subscription_ratio,
                 supports_auto=supports_auto_mosr))
 
+        self._supports_encryption = True
+
     @staticmethod
     def get_driver_options():
         return nfs_opts + remotefs.nas_opts
@@ -577,7 +582,9 @@ class NfsDriver(remotefs.RemoteFSSnapDriverDistributed):
         self._check_snapshot_support()
         return self._delete_snapshot(snapshot)
 
-    def _copy_volume_from_snapshot(self, snapshot, volume, volume_size):
+    def _copy_volume_from_snapshot(self, snapshot, volume, volume_size,
+                                   src_encryption_key_id=None,
+                                   new_encryption_key_id=None):
         """Copy data from snapshot to destination volume.
 
         This is done with a qemu-img convert to raw/qcow2 from the snapshot
@@ -610,9 +617,46 @@ class NfsDriver(remotefs.RemoteFSSnapDriverDistributed):
         else:
             out_format = 'raw'
 
-        image_utils.convert_image(path_to_snap_img,
+        if new_encryption_key_id is not None:
-                                  path_to_new_vol,
+            if src_encryption_key_id is None:
-                                  out_format,
+                message = _("Can't create an encrypted volume %(format)s "
-                                  run_as_root=self._execute_as_root)
+                            "from an unencrypted source."
+                            ) % {'format': out_format}
+                LOG.error(message)
+                # TODO(enriquetaso): handle unencrypted snap->encrypted vol
+                raise exception.NfsException(message)
+            keymgr = key_manager.API(CONF)
+            new_key = keymgr.get(volume.obj_context, new_encryption_key_id)
+            new_passphrase = \
+                binascii.hexlify(new_key.get_encoded()).decode('utf-8')
+
+            # volume.obj_context is the owner of this request
+            src_key = keymgr.get(volume.obj_context, src_encryption_key_id)
+            src_passphrase = \
+                binascii.hexlify(src_key.get_encoded()).decode('utf-8')
+
+            tmp_dir = volume_utils.image_conversion_dir()
+            with tempfile.NamedTemporaryFile(prefix='luks_',
+                                             dir=tmp_dir) as src_pass_file:
+                with open(src_pass_file.name, 'w') as f:
+                    f.write(src_passphrase)
+
+                with tempfile.NamedTemporaryFile(prefix='luks_',
+                                                 dir=tmp_dir) as new_pass_file:
+                    with open(new_pass_file.name, 'w') as f:
+                        f.write(new_passphrase)
+
+                    image_utils.convert_image(
+                        path_to_snap_img,
+                        path_to_new_vol,
+                        'luks',
+                        passphrase_file=new_pass_file.name,
+                        src_passphrase_file=src_pass_file.name,
+                        run_as_root=self._execute_as_root)
+        else:
+            image_utils.convert_image(path_to_snap_img,
+                                      path_to_new_vol,
+                                      out_format,
+                                      run_as_root=self._execute_as_root)
 
         self._set_rw_permissions_for_all(path_to_new_vol)

cinder/volume/drivers/remotefs.py

@@ -14,6 +14,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import binascii
 import collections
 import errno
 import hashlib
@@ -24,10 +25,13 @@ import os
 import re
 import shutil
 import string
+import tempfile
 import time
 
+from castellan import key_manager
 from oslo_config import cfg
 from oslo_log import log as logging
+from oslo_serialization import jsonutils
 from oslo_utils import units
 import six
 
@@ -295,8 +299,20 @@ class RemoteFSDriver(driver.BaseVD):
         volume_path = self.local_path(volume)
         volume_size = volume.size
 
-        if getattr(self.configuration,
+        encrypted = volume.encryption_key_id is not None
-                   self.driver_prefix + '_qcow2_volumes', False):
+
+        if encrypted:
+            encryption = volume_utils.check_encryption_provider(
+                self.db,
+                volume,
+                volume.obj_context)
+
+            self._create_encrypted_volume_file(volume_path,
+                                               volume_size,
+                                               encryption,
+                                               volume.obj_context)
+        elif getattr(self.configuration,
+                     self.driver_prefix + '_qcow2_volumes', False):
             # QCOW2 volumes are inherently sparse, so this setting
             # will override the _sparsed_volumes setting.
             self._create_qcow2_file(volume_path, volume_size)
@@ -401,6 +417,47 @@ class RemoteFSDriver(driver.BaseVD):
                       path, str(size_gb * units.Gi),
                       run_as_root=self._execute_as_root)
 
+    def _create_encrypted_volume_file(self,
+                                      path,
+                                      size_gb,
+                                      encryption,
+                                      context):
+        """Create an encrypted volume.
+
+        This works by creating an encrypted image locally,
+        and then uploading it to the volume.
+        """
+
+        cipher_spec = image_utils.decode_cipher(encryption['cipher'],
+                                                encryption['key_size'])
+
+        # TODO(enriquetaso): share this code w/ the RBD driver
+        # Fetch the key associated with the volume and decode the passphrase
+        keymgr = key_manager.API(CONF)
+        key = keymgr.get(context, encryption['encryption_key_id'])
+        passphrase = binascii.hexlify(key.get_encoded()).decode('utf-8')
+
+        # create a file
+        tmp_dir = volume_utils.image_conversion_dir()
+
+        with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp_key:
+            # TODO(enriquetaso): encrypt w/ aes256 cipher text
+            # (qemu-img feature) ?
+            with open(tmp_key.name, 'w') as f:
+                f.write(passphrase)
+
+            self._execute(
+                'qemu-img', 'create', '-f', 'qcow2',
+                '-o',
+                'encrypt.format=luks,'
+                'encrypt.key-secret=sec1,'
+                'encrypt.cipher-alg=%(cipher_alg)s,'
+                'encrypt.cipher-mode=%(cipher_mode)s,'
+                'encrypt.ivgen-alg=%(ivgen_alg)s' % cipher_spec,
+                '--object', 'secret,id=sec1,format=raw,file=' + tmp_key.name,
+                path, str(size_gb * units.Gi),
+                run_as_root=self._execute_as_root)
+
     def _set_rw_permissions(self, path):
         """Sets access permissions for given NFS path.
 
@@ -820,22 +877,53 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
     def _qemu_img_info(self, path, volume_name):
         raise NotImplementedError()
 
-    def _img_commit(self, path):
+    def _img_commit(self, path, passphrase_file=None, backing_file=None):
         # TODO(eharney): this is not using the correct permissions for
         # NFS snapshots
         #  It needs to run as root for volumes attached to instances, but
         #  does not when in secure mode.
-        self._execute('qemu-img', 'commit', '-d', path,
+        cmd = ['qemu-img', 'commit']
-                      run_as_root=self._execute_as_root)
+        if passphrase_file:
+            obj = ['--object',
+                   'secret,id=s0,format=raw,file=%s' % passphrase_file]
+            image_opts = ['--image-opts']
+
+            src_opts = \
+                "file.filename=%(filename)s,encrypt.format=luks," \
+                "encrypt.key-secret=s0,backing.file.filename=%(backing)s," \
+                "backing.encrypt.key-secret=s0" % {
+                    'filename': path,
+                    'backing': backing_file,
+                }
+
+            path_no_to_delete = ['-d', src_opts]
+            cmd += obj + image_opts + path_no_to_delete
+        else:
+            cmd += ['-d', path]
+
+        self._execute(*cmd, run_as_root=self._execute_as_root)
         self._delete(path)
 
-    def _rebase_img(self, image, backing_file, volume_format):
+    def _rebase_img(self, image, backing_file, volume_format,
+                    passphrase_file=None):
         # qemu-img create must run as root, because it reads from the
         # backing file, which will be owned by qemu:qemu if attached to an
         # instance.
         # TODO(erlon): Sanity check this.
-        self._execute('qemu-img', 'rebase', '-u', '-b', backing_file, image,
+        command = ['qemu-img', 'rebase', '-u']
-                      '-F', volume_format, run_as_root=self._execute_as_root)
+        # if encrypted
+        if passphrase_file:
+            objectdef = "secret,id=s0,file=%s" % passphrase_file
+            filename = "encrypt.key-secret=s0,"\
+                "file.filename=%(filename)s" % {'filename': image}
+
+            command += ['--object', objectdef, '-b', backing_file,
+                        '-F', volume_format, '--image-opts', filename]
+        # not encrypted
+        else:
+            command += ['-b', backing_file, image, '-F', volume_format]
+
+        self._execute(*command, run_as_root=self._execute_as_root)
 
     def _read_info_file(self, info_path, empty_if_missing=False):
         """Return dict of snapshot information.
@@ -1041,7 +1129,7 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
 
         # Create fake volume and snapshot objects
         vol_attrs = ['provider_location', 'size', 'id', 'name', 'status',
-                     'volume_type', 'metadata']
+                     'volume_type', 'metadata', 'obj_context']
         Volume = collections.namedtuple('Volume', vol_attrs)
         volume_info = Volume(provider_location=src_vref.provider_location,
                              size=src_vref.size,
@@ -1049,7 +1137,8 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
                              name=volume_name,
                              status=src_vref.status,
                              volume_type=src_vref.volume_type,
-                             metadata=src_vref.metadata)
+                             metadata=src_vref.metadata,
+                             obj_context=volume.obj_context)
 
         if (self._always_use_temp_snap_when_cloning or
                 self._snapshots_exist(src_vref)):
@@ -1071,9 +1160,13 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
 
             self._create_snapshot(temp_snapshot)
             try:
-                self._copy_volume_from_snapshot(temp_snapshot,
+                self._copy_volume_from_snapshot(
-                                                volume_info,
+                    temp_snapshot,
-                                                volume.size)
+                    volume_info,
+                    volume.size,
+                    src_encryption_key_id=src_vref.encryption_key_id,
+                    new_encryption_key_id=volume.encryption_key_id)
+
                 # remove temp snapshot after the cloning is done
                 temp_snapshot.status = fields.SnapshotStatus.DELETING
                 temp_snapshot.context = context.elevated()
@@ -1134,6 +1227,7 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
         self._validate_state(volume_status, acceptable_states)
 
         vol_path = self._local_volume_dir(snapshot.volume)
+        volume_path = os.path.join(vol_path, snapshot.volume.name)
 
         # Determine the true snapshot file for this snapshot
         # based on the .info file
@@ -1206,14 +1300,33 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
                                                 snapshot,
                                                 online_delete_info)
 
+        encrypted = snapshot.encryption_key_id is not None
+
+        if encrypted:
+            keymgr = key_manager.API(CONF)
+            encryption_key = snapshot.encryption_key_id
+            new_key = keymgr.get(snapshot.obj_context, encryption_key)
+            src_passphrase = \
+                binascii.hexlify(new_key.get_encoded()).decode('utf-8')
+
+            tmp_dir = volume_utils.image_conversion_dir()
+
         if utils.paths_normcase_equal(snapshot_file, active_file):
             # There is no top file
             #      T0       |        T1         |
             #     base      |   snapshot_file   | None
             # (guaranteed to|  (being deleted,  |
             #    exist)     |  committed down)  |
-
+            if encrypted:
-            self._img_commit(snapshot_path)
+                with tempfile.NamedTemporaryFile(prefix='luks_',
+                                                 dir=tmp_dir) as src_file:
+                    with open(src_file.name, 'w') as f:
+                        f.write(src_passphrase)
+                    self._img_commit(snapshot_path,
+                                     passphrase_file=src_file.name,
+                                     backing_file=volume_path)
+            else:
+                self._img_commit(snapshot_path)
             # Active file has changed
             snap_info['active'] = base_file
         else:
@@ -1241,11 +1354,25 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
                     higher_file
                 raise exception.RemoteFSException(msg)
 
-            self._img_commit(snapshot_path)
+            if encrypted:
+                with tempfile.NamedTemporaryFile(prefix='luks_',
+                                                 dir=tmp_dir) as src_file:
+                    with open(src_file.name, 'w') as f:
+                        f.write(src_passphrase)
+                    self._img_commit(snapshot_path,
+                                     passphrase_file=src_file.name,
+                                     backing_file=volume_path)
 
-            higher_file_path = os.path.join(vol_path, higher_file)
+                    higher_file_path = os.path.join(vol_path, higher_file)
-            base_file_fmt = base_file_img_info.file_format
+                    base_file_fmt = base_file_img_info.file_format
-            self._rebase_img(higher_file_path, base_file, base_file_fmt)
+                    self._rebase_img(higher_file_path, volume_path,
+                                     base_file_fmt, src_file.name)
+            else:
+                self._img_commit(snapshot_path)
+
+                higher_file_path = os.path.join(vol_path, higher_file)
+                base_file_fmt = base_file_img_info.file_format
+                self._rebase_img(higher_file_path, base_file, base_file_fmt)
 
         # Remove snapshot_file from info
         del(snap_info[snapshot.id])
@@ -1274,11 +1401,15 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
 
         self._copy_volume_from_snapshot(snapshot,
                                         volume,
-                                        volume.size)
+                                        volume.size,
+                                        snapshot.volume.encryption_key_id,
+                                        volume.encryption_key_id)
 
         return {'provider_location': volume.provider_location}
 
-    def _copy_volume_from_snapshot(self, snapshot, volume, volume_size):
+    def _copy_volume_from_snapshot(self, snapshot, volume, volume_size,
+                                   src_encryption_key_id=None,
+                                   new_encryption_key_id=None):
         raise NotImplementedError()
 
     def _do_create_snapshot(self, snapshot, backing_filename,
@@ -1294,24 +1425,86 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
             self._local_volume_dir(snapshot.volume),
             backing_filename)
 
+        volume_path = os.path.join(
+            self._local_volume_dir(snapshot.volume),
+            snapshot.volume.name)
+
         info = self._qemu_img_info(backing_path_full_path,
                                    snapshot.volume.name)
         backing_fmt = info.file_format
+        obj_context = snapshot.volume.obj_context
 
-        command = ['qemu-img', 'create', '-f', 'qcow2', '-o',
+        # create new qcow2 file
-                   'backing_file=%s,backing_fmt=%s' %
+        if snapshot.volume.encryption_key_id is None:
-                   (backing_path_full_path, backing_fmt),
+            command = ['qemu-img', 'create', '-f', 'qcow2', '-o',
-                   new_snap_path,
+                       'backing_file=%s,backing_fmt=%s' %
-                   "%dG" % snapshot.volume.size]
+                       (backing_path_full_path, backing_fmt),
-        self._execute(*command, run_as_root=self._execute_as_root)
+                       new_snap_path,
+                       "%dG" % snapshot.volume.size]
 
-        command = ['qemu-img', 'rebase', '-u',
+            self._execute(*command, run_as_root=self._execute_as_root)
-                   '-b', backing_filename,
-                   '-F', backing_fmt,
-                   new_snap_path]
 
-        # qemu-img rebase must run as root for the same reasons as above
+            command = ['qemu-img', 'rebase', '-u',
-        self._execute(*command, run_as_root=self._execute_as_root)
+                       '-b', backing_filename,
+                       '-F', backing_fmt,
+                       new_snap_path]
+
+            # qemu-img rebase must run as root for the same reasons as above
+            self._execute(*command, run_as_root=self._execute_as_root)
+
+        else:
+            # encrypted
+            keymgr = key_manager.API(CONF)
+            # Get key for the source volume using the context of this request.
+            key = keymgr.get(obj_context,
+                             snapshot.volume.encryption_key_id)
+            passphrase = binascii.hexlify(key.get_encoded()).decode('utf-8')
+
+            tmp_dir = volume_utils.image_conversion_dir()
+            with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp_key:
+                with open(tmp_key.name, 'w') as f:
+                    f.write(passphrase)
+
+                file_json_dict = {"driver": "qcow2",
+                                  "encrypt.key-secret": "s0",
+                                  "backing.encrypt.key-secret": "s0",
+                                  "backing.file.filename": volume_path,
+                                  "file": {"driver": "file",
+                                           "filename": backing_path_full_path,
+                                           }}
+                file_json = jsonutils.dumps(file_json_dict)
+
+                encryption = volume_utils.check_encryption_provider(
+                    db=db,
+                    volume=snapshot.volume,
+                    context=obj_context)
+
+                cipher_spec = image_utils.decode_cipher(encryption['cipher'],
+                                                        encryption['key_size'])
+
+                command = ('qemu-img', 'create', '-f' 'qcow2',
+                           '-o', 'encrypt.format=luks,encrypt.key-secret=s1,'
+                           'encrypt.cipher-alg=%(cipher_alg)s,'
+                           'encrypt.cipher-mode=%(cipher_mode)s,'
+                           'encrypt.ivgen-alg=%(ivgen_alg)s' % cipher_spec,
+                           '-b', 'json:' + file_json,
+                           '--object', 'secret,id=s0,file=' + tmp_key.name,
+                           '--object', 'secret,id=s1,file=' + tmp_key.name,
+                           new_snap_path)
+                self._execute(*command, run_as_root=self._execute_as_root)
+
+                command_path = 'encrypt.key-secret=s0,file.filename='
+                command = ['qemu-img', 'rebase',
+                           '--object', 'secret,id=s0,file=' + tmp_key.name,
+                           '--image-opts',
+                           command_path + new_snap_path,
+                           '-u',
+                           '-b', backing_filename,
+                           '-F', backing_fmt]
+
+                # qemu-img rebase must run as root for the same reasons as
+                # above
+                self._execute(*command, run_as_root=self._execute_as_root)
 
         self._set_rw_permissions(new_snap_path)
 

cinder/volume/volume_utils.py

@@ -1194,6 +1194,11 @@ def check_encryption_provider(db, volume, context):
     """
 
     encryption = db.volume_encryption_metadata_get(context, volume.id)
+
+    if 'provider' not in encryption:
+        message = _("Invalid encryption spec.")
+        raise exception.VolumeDriverException(message=message)
+
     provider = encryption['provider']
     if provider in encryptors.LEGACY_PROVIDER_CLASS_TO_FORMAT_MAP:
         provider = encryptors.LEGACY_PROVIDER_CLASS_TO_FORMAT_MAP[provider]

releasenotes/notes/bp-nfs-volume-encryption-3d8362843caeb39c.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    The NFS driver now supports the creation of encrypted volumes.