Add support for transferring encrypted volumes

A new microversion 3.70 adds the ability to transfer a volume's
encryption key when transferring a volume to another project.

When the volume transfer is initiated, the volume's encryption
secret is essentially transferred to the cinder service.
- The cinder service creates a new encryption_key_id that contains
  a copy of the volume's encryption secret.
- The volume (and its snapshots) is updated with the new
  encryption_key_id (the one owned by the cinder service).
- The volume's original encryption_key_id (owned by the volume's
  owner) is deleted.

When the transfer is accepted, the secret is transferred to the
user accepting the transfer.
- A new encryption_key_id is generated on behalf of the new user
  that contains a copy of the volume's encryption secret.
- The volume (and its snapshots) is updated with the new
  encryption_key_id (the one owned by the user).
- The intermediate encryption_key_id owned by the cinder service
  is deleted.

When a transfer is cancelled (deleted), the same process is used
to transfer ownship back to the user that cancelled the transfer.

Implements: blueprint transfer-encrypted-volume
Change-Id: I459f06504e90025c9c0b539981d3d56a2a9394c7

api-ref/source/v3/samples/versions/version-show-response.json

@@ -21,8 +21,8 @@
             ],
             "min_version": "3.0",
             "status": "CURRENT",
-            "updated": "2022-03-30T00:00:00Z",
-            "version": "3.69"
+            "updated": "2022-08-31T00:00:00Z",
+            "version": "3.70"
         }
     ]
 }

api-ref/source/v3/samples/versions/versions-response.json

@@ -21,8 +21,8 @@
             ],
             "min_version": "3.0",
             "status": "CURRENT",
-            "updated": "2022-03-30T00:00:00Z",
-            "version": "3.69"
+            "updated": "2022-08-31T00:00:00Z",
+            "version": "3.70"
         }
     ]
 }

cinder/api/microversions.py

@@ -177,6 +177,8 @@
 
 SHARED_TARGETS_TRISTATE = '3.69'
 
+TRANSFER_ENCRYPTED_VOLUME = '3.70'
+
 
 def get_mv_header(version):
     """Gets a formatted HTTP microversion header.

cinder/api/openstack/api_version_request.py

@@ -155,14 +155,15 @@
     * 3.67 - API URLs no longer need to include a project_id parameter.
     * 3.68 - Support re-image volume
     * 3.69 - Allow null value for shared_targets
+    * 3.70 - Support encrypted volume transfers
 """
 
 # The minimum and maximum versions of the API supported
 # The default api version request is defined to be the
 # minimum version of the API supported.
 _MIN_API_VERSION = "3.0"
-_MAX_API_VERSION = "3.69"
-UPDATED = "2022-04-20T00:00:00Z"
+_MAX_API_VERSION = "3.70"
+UPDATED = "2022-08-31T00:00:00Z"
 
 
 # NOTE(cyeoh): min and max versions declared as functions so we can

cinder/api/openstack/rest_api_version_history.rst

@@ -528,3 +528,8 @@ following meanings:
   manual scans.
 - ``false``: Never do locking.
 - ``null``: Forced locking regardless of the iSCSI initiator.
+
+3.70
+----
+
+Add the ability to transfer encrypted volumes and their snapshots.

cinder/api/v3/volume_transfer.py

@@ -96,11 +96,15 @@ def create(self, req, body):
         no_snapshots = strutils.bool_from_string(transfer.get('no_snapshots',
                                                               False))
 
+        req_version = req.api_version_request
+        allow_encrypted = req_version.matches(mv.TRANSFER_ENCRYPTED_VOLUME)
+
         LOG.info("Creating transfer of volume %s", volume_id)
 
         try:
-            new_transfer = self.transfer_api.create(context, volume_id, name,
-                                                    no_snapshots=no_snapshots)
+            new_transfer = self.transfer_api.create(
+                context, volume_id, name,
+                no_snapshots=no_snapshots, allow_encrypted=allow_encrypted)
         # Not found exception will be handled at the wsgi level
         except exception.Invalid as error:
             raise exc.HTTPBadRequest(explanation=error.msg)

cinder/keymgr/transfer.py

@@ -0,0 +1,107 @@
+# Copyright 2022 Red Hat, Inc.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from castellan.common.credentials import keystone_password
+from castellan.common import exception as castellan_exception
+from castellan import key_manager as castellan_key_manager
+from oslo_config import cfg
+from oslo_log import log as logging
+from oslo_utils import excutils
+
+from cinder import context
+from cinder import objects
+
+LOG = logging.getLogger(__name__)
+
+CONF = cfg.CONF
+
+
+class KeyTransfer(object):
+    def __init__(self, conf: cfg.ConfigOpts):
+        self.conf = conf
+        self._service_context = keystone_password.KeystonePassword(
+            password=conf.keystone_authtoken.password,
+            auth_url=conf.keystone_authtoken.auth_url,
+            username=conf.keystone_authtoken.username,
+            user_domain_name=conf.keystone_authtoken.user_domain_name,
+            project_name=conf.keystone_authtoken.project_name,
+            project_domain_name=conf.keystone_authtoken.project_domain_name)
+
+    @property
+    def service_context(self):
+        """Returns the cinder service's context."""
+        return self._service_context
+
+    def transfer_key(self,
+                     volume: objects.volume.Volume,
+                     src_context: context.RequestContext,
+                     dst_context: context.RequestContext) -> None:
+        """Transfer the key from the src_context to the dst_context."""
+        key_manager = castellan_key_manager.API(self.conf)
+
+        old_encryption_key_id = volume.encryption_key_id
+        secret = key_manager.get(src_context, old_encryption_key_id)
+        try:
+            new_encryption_key_id = key_manager.store(dst_context, secret)
+        except castellan_exception.KeyManagerError:
+            with excutils.save_and_reraise_exception():
+                LOG.error("Failed to transfer the encryption key. This is "
+                          "likely because the cinder service lacks the "
+                          "privilege to create secrets.")
+
+        volume.encryption_key_id = new_encryption_key_id
+        volume.save()
+
+        snapshots = objects.snapshot.SnapshotList.get_all_for_volume(
+            context.get_admin_context(),
+            volume.id)
+        for snapshot in snapshots:
+            snapshot.encryption_key_id = new_encryption_key_id
+            snapshot.save()
+
+        key_manager.delete(src_context, old_encryption_key_id)
+
+
+def transfer_create(context: context.RequestContext,
+                    volume: objects.volume.Volume,
+                    conf: cfg.ConfigOpts = CONF) -> None:
+    """Transfer the key from the owner to the cinder service."""
+    LOG.info("Initiating transfer of encryption key for volume %s", volume.id)
+    key_transfer = KeyTransfer(conf)
+    key_transfer.transfer_key(volume,
+                              src_context=context,
+                              dst_context=key_transfer.service_context)
+
+
+def transfer_accept(context: context.RequestContext,
+                    volume: objects.volume.Volume,
+                    conf: cfg.ConfigOpts = CONF) -> None:
+    """Transfer the key from the cinder service to the recipient."""
+    LOG.info("Accepting transfer of encryption key for volume %s", volume.id)
+    key_transfer = KeyTransfer(conf)
+    key_transfer.transfer_key(volume,
+                              src_context=key_transfer.service_context,
+                              dst_context=context)
+
+
+def transfer_delete(context: context.RequestContext,
+                    volume: objects.volume.Volume,
+                    conf: cfg.ConfigOpts = CONF) -> None:
+    """Transfer the key from the cinder service back to the owner."""
+    LOG.info("Cancelling transfer of encryption key for volume %s", volume.id)
+    key_transfer = KeyTransfer(conf)
+    key_transfer.transfer_key(volume,
+                              src_context=key_transfer.service_context,
+                              dst_context=context)

cinder/tests/unit/api/v3/test_volume_transfer.py

@@ -27,12 +27,14 @@
 from cinder.api.v3 import volume_transfer as volume_transfer_v3
 from cinder import context
 from cinder import db
+from cinder import exception
 from cinder.objects import fields
 from cinder import quota
 from cinder.tests.unit.api import fakes
 from cinder.tests.unit.api.v2 import fakes as v2_fakes
 from cinder.tests.unit import fake_constants as fake
 from cinder.tests.unit import test
+from cinder.tests.unit import utils as test_utils
 import cinder.transfer
 
 
@@ -358,3 +360,145 @@ class VolumeTransferAPITestCase357(VolumeTransferAPITestCase):
     microversion = mv.TRANSFER_WITH_HISTORY
     DETAIL_LEN = 9
     expect_transfer_history = True
+
+
+@ddt.ddt
+class VolumeTransferEncryptedAPITestCase(test.TestCase):
+    # NOTE:
+    # - The TRANSFER_ENCRYPTED_VOLUME microversion is only relevant when
+    #   creating a volume transfer. The microversion specified when accepting
+    #   or deleting a transfer is not relevant.
+    # - The tests take advantage of the fact that a project_id is no longer
+    #   required in API URLs.
+
+    def setUp(self):
+        super(VolumeTransferEncryptedAPITestCase, self).setUp()
+        self.volume_transfer_api = cinder.transfer.API()
+        self.controller = volume_transfer_v3.VolumeTransferController()
+        self.user_ctxt = context.RequestContext(
+            fake.USER_ID, fake.PROJECT_ID, auth_token=True)
+        self.admin_ctxt = context.get_admin_context()
+
+    def _create_volume(self, encryption_key_id):
+        vol_type = test_utils.create_volume_type(self.admin_ctxt,
+                                                 name='fake_vol_type',
+                                                 testcase_instance=self)
+        volume = test_utils.create_volume(self.user_ctxt,
+                                          volume_type_id=vol_type.id,
+                                          testcase_instance=self,
+                                          encryption_key_id=encryption_key_id)
+        return volume
+
+    @mock.patch('cinder.keymgr.transfer.transfer_create')
+    def _create_transfer(self, volume_id, mock_key_transfer_create):
+        transfer = self.volume_transfer_api.create(self.admin_ctxt,
+                                                   volume_id,
+                                                   display_name='test',
+                                                   allow_encrypted=True)
+        return transfer
+
+    @ddt.data(None, fake.ENCRYPTION_KEY_ID)
+    @mock.patch('cinder.keymgr.transfer.transfer_create')
+    def test_create_transfer(self,
+                             encryption_key_id,
+                             mock_key_transfer_create):
+        volume = self._create_volume(encryption_key_id)
+        body = {"transfer": {"name": "transfer1",
+                             "volume_id": volume.id}}
+
+        req = webob.Request.blank('/v3/volume-transfers')
+        req.method = 'POST'
+        req.headers = mv.get_mv_header(mv.TRANSFER_ENCRYPTED_VOLUME)
+        req.headers['Content-Type'] = 'application/json'
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(fakes.wsgi_app(
+            fake_auth_context=self.user_ctxt))
+
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+
+        call_count = 0 if encryption_key_id is None else 1
+        self.assertEqual(mock_key_transfer_create.call_count, call_count)
+
+    def test_create_transfer_encrypted_volume_not_supported(self):
+        volume = self._create_volume(fake.ENCRYPTION_KEY_ID)
+        body = {"transfer": {"name": "transfer1",
+                             "volume_id": volume.id}}
+
+        req = webob.Request.blank('/v3/volume-transfers')
+        req.method = 'POST'
+        req.headers = mv.get_mv_header(
+            mv.get_prior_version(mv.TRANSFER_ENCRYPTED_VOLUME))
+        req.headers['Content-Type'] = 'application/json'
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(fakes.wsgi_app(
+            fake_auth_context=self.user_ctxt))
+
+        res_dict = jsonutils.loads(res.body)
+
+        self.assertEqual(HTTPStatus.BAD_REQUEST, res.status_int)
+        self.assertEqual(('Invalid volume: '
+                          'transferring encrypted volume is not supported'),
+                         res_dict['badRequest']['message'])
+
+    @mock.patch('cinder.keymgr.transfer.transfer_create',
+                side_effect=exception.KeyManagerError('whoops!'))
+    def test_create_transfer_key_transfer_failed(self,
+                                                 mock_key_transfer_create):
+        volume = self._create_volume(fake.ENCRYPTION_KEY_ID)
+        body = {"transfer": {"name": "transfer1",
+                             "volume_id": volume.id}}
+
+        req = webob.Request.blank('/v3/volume-transfers')
+        req.method = 'POST'
+        req.headers = mv.get_mv_header(mv.TRANSFER_ENCRYPTED_VOLUME)
+        req.headers['Content-Type'] = 'application/json'
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(fakes.wsgi_app(
+            fake_auth_context=self.user_ctxt))
+
+        self.assertEqual(HTTPStatus.INTERNAL_SERVER_ERROR, res.status_int)
+
+    @ddt.data(None, fake.ENCRYPTION_KEY_ID)
+    @mock.patch('cinder.keymgr.transfer.transfer_accept')
+    @mock.patch('cinder.volume.api.API.accept_transfer')
+    def test_accept_transfer(self,
+                             encryption_key_id,
+                             mock_volume_accept_transfer,
+                             mock_key_transfer_accept):
+        volume = self._create_volume(encryption_key_id)
+        transfer = self._create_transfer(volume.id)
+
+        body = {"accept": {"auth_key": transfer['auth_key']}}
+
+        req = webob.Request.blank('/v3/volume-transfers/%s/accept' % (
+                                  transfer['id']))
+        req.method = 'POST'
+        req.headers['Content-Type'] = 'application/json'
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(fakes.wsgi_app(
+            fake_auth_context=self.user_ctxt))
+
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+
+        call_count = 0 if encryption_key_id is None else 1
+        self.assertEqual(mock_key_transfer_accept.call_count, call_count)
+
+    @ddt.data(None, fake.ENCRYPTION_KEY_ID)
+    @mock.patch('cinder.keymgr.transfer.transfer_delete')
+    def test_delete_transfer(self,
+                             encryption_key_id,
+                             mock_key_transfer_delete):
+        volume = self._create_volume(encryption_key_id)
+        transfer = self._create_transfer(volume.id)
+
+        req = webob.Request.blank('/v3/volume-transfers/%s' % (
+            transfer['id']))
+        req.method = 'DELETE'
+        req.headers['Content-Type'] = 'application/json'
+        res = req.get_response(fakes.wsgi_app(
+            fake_auth_context=self.user_ctxt))
+
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+
+        call_count = 0 if encryption_key_id is None else 1
+        self.assertEqual(mock_key_transfer_delete.call_count, call_count)