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: Yadiel Xuan(轩艳东) <xuanyandong@inspur.com>
Co-authored-by: Brian Rosmaita <rosmaita.fossdev@gmail.com>

Change-Id: Ic2da41f3962c1108f974aca952bce3da6d6ac277
Closes-bug: #1944577
changes/58/768458/20
Yadiel Xuan(轩艳东) 2 years ago committed by Brian Rosmaita
parent 8537730c8c
commit 8088dc9580

@ -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

@ -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)

@ -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']

@ -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."""

@ -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,119 @@ 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
called_with = mock_get_flow.call_args.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
called_with = mock_get_flow.call_args.args[2]
self.assertEqual(default_vt['name'],
called_with['volume_type']['name'])
self.assertEqual(default_vt['id'],
called_with['volume_type']['id'])

@ -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',

@ -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(

@ -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.
Loading…
Cancel
Save