Prohibit volume manage to an encrypted volume type

Managing a volume to an encrypted volume type should not be allowed.
One reason is that there is no way for an operator to specify an
encryption key ID for the volume. Another is that we already don't
allow a volume of an encrypted type to be un-managed, so this change
will be symmetric.

Also update and correct the api-ref for this call.

Co-authored-by: xuan <xuanyd@outlook.com>
Co-authored-by: Brian Rosmaita <rosmaita.fossdev@gmail.com>

Change-Id: Ic2da41f3962c1108f974aca952bce3da6d6ac277
Closes-bug: #1944577
(cherry picked from commit 8088dc9580)
change:
  cinder/tests/unit/api/contrib/test_volume_manage.py - used a
  feature of unittest.mock that was introduced in py38; rewritten to
  be compatible with py36

api-ref/source/v3/volume-manage.inc

@@ -24,6 +24,8 @@ or source-name element, if possible.
 The API chooses the size of the volume by rounding up the size of
 the existing storage volume to the next gibibyte (GiB).
 
+You cannot manage a volume to an encrypted volume type.
+
 Prior to microversion 3.16 host field was required, with the possibility of
 defining the cluster it is no longer required, but we must have either a host
 or a cluster field but we cannot have them both with values.
@@ -45,8 +47,8 @@ Request
    - volume: volume
    - description: description_vol
    - availability_zone: availability_zone
-   - bootable: bootable_required
-   - volume_type: volume_type
+   - bootable: bootable
+   - volume_type: name_volume_type_optional
    - name: volume_name_optional
    - host: host_mutex
    - cluster: cluster_mutex

cinder/api/contrib/volume_manage.py

@@ -15,6 +15,7 @@ from http import HTTPStatus
 
 from oslo_log import log as logging
 from oslo_utils import strutils
+import webob
 
 from cinder.api import api_utils
 from cinder.api import common
@@ -146,6 +147,8 @@ class VolumeManageController(wsgi.Controller):
                 'name': 'Host' if host else 'Cluster',
                 'value': host or cluster_name}
             raise exception.ServiceUnavailable(message=msg)
+        except exception.VolumeTypeDefaultMisconfiguredError as err:
+            raise webob.exc.HTTPInternalServerError(explanation=err.msg)
 
         api_utils.add_visible_admin_metadata(new_volume)
 

cinder/tests/functional/api/client.py

@@ -231,6 +231,16 @@ class TestOpenStackClient(object):
     def put_volume(self, volume_id, volume):
         return self.api_put('/volumes/%s' % volume_id, volume)['volume']
 
+    def post_manage_volume(self, host=None, ref=None):
+        if not host:
+            host = "fake-host"
+        if not ref:
+            ref = {"one": "A", "two": "B"}
+        req_body = {"volume": {}}
+        req_body['volume']['host'] = host
+        req_body['volume']['ref'] = ref
+        return self.api_post('/os-volume-manage', req_body)
+
     def get_snapshot(self, snapshot_id):
         return self.api_get('/snapshots/%s' % snapshot_id)['snapshot']
 

cinder/tests/functional/test_volumes.py

@@ -13,6 +13,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from unittest import mock
+
 from oslo_utils import uuidutils
 
 from cinder.tests.functional.api import client
@@ -142,6 +144,22 @@ class VolumesTest(functional_helpers._FunctionalTestBase):
         self.assertRaises(client.OpenStackApiException500,
                           self.api.post_volume, {'volume': {'size': 1}})
 
+    @mock.patch('cinder.api.common.get_cluster_host',
+                return_value=(None, None))
+    def test_manage_volume_default_type_set_none(self, fake_get_host):
+        """Verify missing default volume type errors out when managing."""
+
+        # configure None default type
+        self.flags(default_volume_type=None)
+
+        # manage something in the backend and verify you get a 500
+        self.assertRaises(client.OpenStackApiException500,
+                          self.api.post_manage_volume)
+
+        # make sure that we actually made it into the method we
+        # want to test and the 500 wasn't from something else
+        fake_get_host.assert_called_once()
+
     def test_create_volume_specified_type(self):
         """Verify volume_type is not default."""
 

cinder/tests/unit/api/contrib/test_volume_manage.py

@@ -65,9 +65,24 @@ def service_get(context, service_id, backend_match_level=None, host=None,
 
 # Some of the tests check that volume types are correctly validated during a
 # volume manage operation.  This data structure represents an existing volume
-# type.
-fake_vt = {'id': fake.VOLUME_TYPE_ID,
-           'name': 'good_fakevt'}
+# type.  NOTE: cinder.db.sqlalchemy.volume_type_get() returns a dict describing
+# a specific volume type; this dict always contains an 'extra_specs' key.
+fake_vt = {
+    'id': fake.VOLUME_TYPE_ID,
+    'name': 'good_fakevt',
+    'extra_specs': {},
+}
+
+fake_encrypted_vt = {
+    'id': fake.VOLUME_TYPE2_ID,
+    'name': 'fake_encrypted_vt',
+    'extra_specs': {},
+    'encryption': {
+        'cipher': 'fake_cipher',
+        'control_location': 'front-end',
+        'key_size': 256,
+        'provider': 'fake_provider'},
+}
 
 
 def vt_get_volume_type_by_name(context, name):
@@ -79,6 +94,8 @@ def vt_get_volume_type_by_name(context, name):
     """
     if name == fake_vt['name']:
         return fake_vt
+    if name == fake_encrypted_vt['name']:
+        return fake_encrypted_vt
     raise exception.VolumeTypeNotFoundByName(volume_type_name=name)
 
 
@@ -91,9 +108,37 @@ def vt_get_volume_type(context, vt_id):
     """
     if vt_id == fake_vt['id']:
         return fake_vt
+    if vt_id == fake_encrypted_vt['id']:
+        return fake_encrypted_vt
     raise exception.VolumeTypeNotFound(volume_type_id=vt_id)
 
 
+def vt_get_default_volume_type(context):
+    """Replacement for cinder.volume.volume_types.get_default_volume_type.
+
+    If you want to use a specific fake volume type defined above, set
+    the flag for default_volume_type to the name of that fake type.
+
+    If you want to raise VolumeTypeDefaultMisconfiguredError, then set
+    the flag for default_volume_type to None.
+
+    Otherwise, for *any* non-None value of default_volume_type, this
+    will return our generic fake volume type.  (NOTE: by default,
+    CONF.default_volume_type is '__DEFAULT__'.)
+
+    """
+    default_vt_name = CONF.default_volume_type
+    if not default_vt_name:
+        raise exception.VolumeTypeDefaultMisconfiguredError(
+            volume_type_name='from vt_get_default_volume_type')
+    try:
+        default_vt = vt_get_volume_type_by_name(context, default_vt_name)
+    except exception.VolumeTypeNotFoundByName:
+        default_vt = fake_vt
+
+    return default_vt
+
+
 def api_manage(*args, **kwargs):
     """Replacement for cinder.volume.api.API.manage_existing.
 
@@ -145,6 +190,8 @@ def api_get_manageable_volumes(*args, **kwargs):
 
 @ddt.ddt
 @mock.patch('cinder.db.sqlalchemy.api.service_get', service_get)
+@mock.patch('cinder.volume.volume_types.get_default_volume_type',
+            vt_get_default_volume_type)
 @mock.patch('cinder.volume.volume_types.get_volume_type_by_name',
             vt_get_volume_type_by_name)
 @mock.patch('cinder.volume.volume_types.get_volume_type',
@@ -477,3 +524,121 @@ class VolumeManageTest(test.TestCase):
         self.assertEqual(1, mock_api_manage.call_count)
         self.assertEqual('creating',
                          jsonutils.loads(res.body)['volume']['status'])
+
+    def test_negative_manage_to_encrypted_type(self):
+        """Not allowed to manage a volume to an encrypted volume type."""
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=True)
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref',
+                           'volume_type': fake_encrypted_vt['name']}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_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=ctxt))
+        self.assertEqual(HTTPStatus.BAD_REQUEST, res.status_int)
+
+    def test_negative_manage_to_encrypted_default_type(self):
+        """Fail if no vol type in request and default vol type is encrypted."""
+
+        self.flags(default_volume_type=fake_encrypted_vt['name'])
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=True)
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_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=ctxt))
+        self.assertEqual(HTTPStatus.BAD_REQUEST, res.status_int)
+
+    def test_negative_no_volume_type(self):
+        """Fail when no volume type is available for the managed volume."""
+        self.flags(default_volume_type=None)
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=True)
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_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=ctxt))
+        self.assertEqual(HTTPStatus.INTERNAL_SERVER_ERROR, res.status_int)
+
+    @mock.patch('cinder.group.API')
+    @mock.patch('cinder.flow_utils')
+    @mock.patch('cinder.volume.flows.api.manage_existing.get_flow')
+    @mock.patch('cinder.volume.api.API._get_service_by_host_cluster')
+    def test_manage_when_default_type_is_encrypted(self,
+                                                   mock_get_cluster,
+                                                   mock_get_flow,
+                                                   mock_flow_utils,
+                                                   mock_group_api):
+        """Default type doesn't matter if non-encrypted type is in request."""
+
+        # make an encrypted type the default volume type
+        self.flags(default_volume_type=fake_encrypted_vt['name'])
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=True)
+
+        # pass a non-encrypted volume type in the request
+        requested_vt = fake_vt
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref',
+                           'volume_type': requested_vt['name']}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_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=ctxt))
+
+        # request should be accepted
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+
+        # make sure the volume type passed through is the specified one
+        args, _ = mock_get_flow.call_args
+        called_with = args[2]
+        self.assertEqual(requested_vt['name'],
+                         called_with['volume_type']['name'])
+        self.assertEqual(requested_vt['id'],
+                         called_with['volume_type']['id'])
+
+    @mock.patch('cinder.group.API')
+    @mock.patch('cinder.flow_utils')
+    @mock.patch('cinder.volume.flows.api.manage_existing.get_flow')
+    @mock.patch('cinder.volume.api.API._get_service_by_host_cluster')
+    def test_manage_with_default_type(self,
+                                      mock_get_cluster,
+                                      mock_get_flow,
+                                      mock_flow_utils,
+                                      mock_group_api):
+        """A non-encrypted default volume type should cause no problems."""
+
+        # make an non-encrypted type the default volume type
+        default_vt = fake_vt
+        self.flags(default_volume_type=default_vt['name'])
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=True)
+
+        # don't pass a volume type in the request
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_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=ctxt))
+
+        # request should be accepted
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+
+        # make sure the volume type passed through is the default
+        args, _ = mock_get_flow.call_args
+        called_with = args[2]
+        self.assertEqual(default_vt['name'],
+                         called_with['volume_type']['name'])
+        self.assertEqual(default_vt['id'],
+                         called_with['volume_type']['id'])

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

@@ -46,6 +46,8 @@ def app():
 @ddt.ddt
 @mock.patch('cinder.objects.service.Service.get_by_host_and_topic',
             test_contrib.service_get)
+@mock.patch('cinder.volume.volume_types.get_default_volume_type',
+            test_contrib.vt_get_default_volume_type)
 @mock.patch('cinder.volume.volume_types.get_volume_type_by_name',
             test_contrib.vt_get_volume_type_by_name)
 @mock.patch('cinder.volume.volume_types.get_volume_type',

cinder/volume/api.py

@@ -1991,6 +1991,23 @@ class API(base.Base):
                 raise exception.InvalidVolume(
                     _("Unable to manage existing volume."
                       " The volume is already managed"))
+        if not volume_type:
+            try:
+                volume_type = volume_types.get_default_volume_type(context)
+            except exception.VolumeTypeDefaultMisconfiguredError:
+                LOG.error('Default volume type not found. This must be '
+                          'corrected immediately or all volume-create '
+                          'requests that do not specify a volume type '
+                          'will fail.')
+                raise
+        is_encrypted = False
+        if volume_type:
+            is_encrypted = volume_types.is_encrypted(context,
+                                                     volume_type['id'])
+        if is_encrypted:
+            msg = _("Managing to an encrypted volume type is not supported.")
+            LOG.error(msg)
+            raise exception.InvalidVolumeType(msg)
 
         if volume_type and 'extra_specs' not in volume_type:
             extra_specs = volume_types.get_volume_type_extra_specs(

releasenotes/notes/bug-1944577-no-manage-to-encrypted-type-b5b5d7f8360f037f.yaml

@@ -0,0 +1,8 @@
+---
+fixes:
+  - |
+    `Bug #1944577 <https://bugs.launchpad.net/cinder/+bug/1944577>`_:
+    Managing a volume to an encrypted type was never a good idea because
+    there was no way to specify an encryption key ID so that the volume
+    could be used.  Requests to manage a volume to an encrypted volume
+    type now result in an invalid request response.