Merge "Prohibit volume manage to an encrypted volume type" into stable/xena

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

@@ -1794,6 +1794,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.