Prevent creating encrypted volume with image

When creating an encrypted volume with a unencrypted image,
cinder just copies the unencrypted data to the volume,
which keeps unencrypted data in the volume.
But when booting/attaching the volume to Nova, it thinks
encrypted data on the volume. As a result, Nova reads
the data incorrectly.

The fix is to prevent such action. Later, we can
implement to write encrypted data in encrypted volumes
when creating from unencrypted image.

The test case file had conflicts due to tests moving from
cinder/tests to cinder/tests/unit.  I have resolved the
issues for this cherry pick.

Conflicts: cinder/tests/test_create_volume_flow.py

(cherry picked from commit: 19d5610f93)
Change-Id: I0af8318222cee023451c38812bf83cdb4086faae
Closes-bug: #1482464
This commit is contained in:
lisali 2015-08-07 14:35:27 +08:00 committed by Jay S. Bryant
parent f303c9bd42
commit d836a8f6b0
2 changed files with 106 additions and 0 deletions

View File

@ -23,6 +23,8 @@ from cinder import exception
from cinder import test
from cinder.tests import fake_snapshot
from cinder.tests import fake_volume
from cinder.tests.image import fake as fake_image
from cinder.tests.keymgr import mock_key_mgr
from cinder.volume.flows.api import create_volume
from cinder.volume.flows.manager import create_volume as create_volume_manager
@ -123,6 +125,103 @@ class CreateVolumeFlowTestCase(test.TestCase):
task._cast_create_volume(self.ctxt, spec, props)
@mock.patch('cinder.volume.volume_types.is_encrypted')
@mock.patch('cinder.volume.flows.api.create_volume.'
'ExtractVolumeRequestTask.'
'_get_volume_type_id')
def test_extract_volume_request_from_image_encrypted(
self,
fake_get_volume_type_id,
fake_is_encrypted):
fake_image_service = fake_image.FakeImageService()
image_id = 1
image_meta = {}
image_meta['id'] = image_id
image_meta['status'] = 'active'
image_meta['size'] = 1
fake_image_service.create(self.ctxt, image_meta)
fake_key_manager = mock_key_mgr.MockKeyManager()
task = create_volume.ExtractVolumeRequestTask(
fake_image_service,
{'nova'},
rebind={'size': 1,
'availability_zone': 'nova',
'volume_type': 'my_type'})
fake_is_encrypted.return_value = True
self.assertRaises(exception.InvalidInput,
task.execute,
self.ctxt,
size=1,
snapshot=None,
image_id=image_id,
source_volume=None,
availability_zone='nova',
volume_type=None,
metadata=None,
key_manager=fake_key_manager,
source_replica=None,
consistencygroup=None,
cgsnapshot=None)
@mock.patch('cinder.volume.volume_types.is_encrypted')
@mock.patch('cinder.volume.volume_types.get_volume_type_qos_specs')
@mock.patch('cinder.volume.flows.api.create_volume.'
'ExtractVolumeRequestTask.'
'_get_volume_type_id')
def test_extract_volume_request_from_image(
self,
fake_get_type_id,
fake_get_qos,
fake_is_encrypted):
fake_image_service = fake_image.FakeImageService()
image_id = 2
image_meta = {}
image_meta['id'] = image_id
image_meta['status'] = 'active'
image_meta['size'] = 1
fake_image_service.create(self.ctxt, image_meta)
fake_key_manager = mock_key_mgr.MockKeyManager()
volume_type = 'type1'
task = create_volume.ExtractVolumeRequestTask(
fake_image_service,
{'nova'},
rebind={'size': 1,
'availability_zone': 'nova',
'volume_type': 'my_type'})
fake_is_encrypted.return_value = False
fake_get_type_id.return_value = 1
fake_get_qos.return_value = {'qos_specs': None}
result = task.execute(self.ctxt,
size=1,
snapshot=None,
image_id=image_id,
source_volume=None,
availability_zone='nova',
volume_type=volume_type,
metadata=None,
key_manager=fake_key_manager,
source_replica=None,
consistencygroup=None,
cgsnapshot=None)
expected_result = {'size': 1,
'snapshot_id': None,
'source_volid': None,
'availability_zone': 'nova',
'volume_type': volume_type,
'volume_type_id': 1,
'encryption_key_id': None,
'qos_specs': None,
'source_replicaid': None,
'consistencygroup_id': None,
'cgsnapshot_id': None, }
self.assertEqual(expected_result, result)
class CreateVolumeFlowManagerTestCase(test.TestCase):

View File

@ -438,6 +438,13 @@ class ExtractVolumeRequestTask(flow_utils.CinderTask):
volume_type_id = self._get_volume_type_id(volume_type,
source_volume, snapshot)
if image_id and volume_types.is_encrypted(context, volume_type_id):
msg = _('Create encrypted volumes with type %(type)s '
'from image %(image)s is not supported.')
msg = msg % {'type': volume_type_id,
'image': image_id, }
raise exception.InvalidInput(reason=msg)
encryption_key_id = self._get_encryption_key_id(key_manager,
context,
volume_type_id,