openstack
/
cinder-tempest-plugin
Code Issues Proposed changes

Test the ability to transfer encrypted volumes 

Add tempest tests that transfer an encrypted volume to a user in
another project, and one that verifies the ability to cancel (delete)
a transfer. The purpose is to test the ability to transfer the
barbican secret (i.e. the encryption key) to another user.

The tests create an encrypted volume from an image, and launch an
instance to write a timestamp. After the volume is transferred (or
the transfer is deleted), another instance is booted and the timestamp
read back. The tests also verify the process results in the volume
having a new encrytion_key_id.

Implements: blueprint transfer-encrypted-volume
Depends-On: I459f06504e90025c9c0b539981d3d56a2a9394c7
Depends-On: I11072d6d8a185037c7f4cdd52c45933b0cccaf05
Depends-On: I4ad7fe336c5193604d95fca5a72695d82adaa9f0
Depends-On: Ia3f414c4b9b0829f60841a6dd63c97a893fdde4d
Depends-On: I216f78e8a300ab3f79bbcbb38110adf2bbec2196
Change-Id: I30d02c81dd1cf8509585fe3b7a0f3256f17939b5
This commit is contained in:
Alan Bishop 2022-08-09 14:00:57 -07:00
parent 8f39caf91a
commit 40104c78f8
3 changed files with 202 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
cinder_tempest_plugin/config.py
View File

@ -23,3 +23,15 @@ cinder_option = [
default=False,
help='Enable to run Cinder volume revert tests'),
]
# The barbican service is discovered by config_tempest [1], and will appear
# in the [service_available] group in tempest.conf. However, the 'barbican'
# option isn't registered by tempest itself, and so we do it here. This adds
# the ability to test CONF.service_available.barbican.
#
# [1] I96800a95f844ce7675d266e456e01620e63e347a
service_available_option = [
cfg.BoolOpt('barbican',
default=False,
help="Whether or not barbican is expected to be available"),
]

5
cinder_tempest_plugin/plugin.py
View File

@ -46,6 +46,9 @@ class CinderTempestPlugin(plugins.TempestPlugin):
config.register_opt_group(conf, config.volume_feature_group,
project_config.cinder_option)
config.register_opt_group(conf, config.service_available_group,
project_config.service_available_option)
def get_opt_lists(self):
"""Get a list of options for sample config generation.
@ -54,4 +57,6 @@ class CinderTempestPlugin(plugins.TempestPlugin):
"""
return [
(config.volume_feature_group.name, project_config.cinder_option),
(config.service_available_group.name,
project_config.service_available_option),
]

185
cinder_tempest_plugin/scenario/test_encrypted_volume_transfer.py Normal file
View File

@ -0,0 +1,185 @@
# 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 tempest.common import utils
from tempest.common import waiters
from tempest import config
from tempest.lib import decorators
from tempest.scenario import manager
CONF = config.CONF
class TransferEncryptedVolumeTest(manager.EncryptionScenarioTest):
volume_min_microversion = '3.70'
volume_max_microversion = 'latest'
credentials = ['primary', 'alt', 'admin']
@classmethod
def setup_clients(cls):
super(TransferEncryptedVolumeTest, cls).setup_clients()
# We need the "mv355" volume transfers client
cls.client = cls.os_primary.volume_transfers_mv355_client_latest
cls.alt_client = cls.os_alt.volume_transfers_mv355_client_latest
cls.alt_volumes_client = cls.os_alt.volumes_client_latest
@classmethod
def skip_checks(cls):
super(TransferEncryptedVolumeTest, cls).skip_checks()
if not CONF.service_available.barbican:
raise cls.skipException('Barbican is required')
def setUp(self):
super(TransferEncryptedVolumeTest, self).setUp()
self.keypair = self.create_keypair()
self.security_group = self.create_security_group()
def _create_encrypted_volume_from_image(self):
volume_type = self.create_volume_type()
self.create_encryption_type(type_id=volume_type['id'],
provider='luks',
key_size=256,
cipher='aes-xts-plain64',
control_location='front-end')
return self.create_volume_from_image(volume_type=volume_type['id'])
def _create_or_get_timestamp(self, volume, timestamp_fn):
server = self.boot_instance_from_resource(
source_id=volume['id'],
source_type='volume',
keypair=self.keypair,
security_group=self.security_group)
server_ip = self.get_server_ip(server)
timestamp = timestamp_fn(server_ip,
private_key=self.keypair['private_key'],
server=server)
self.servers_client.delete_server(server['id'])
waiters.wait_for_server_termination(self.servers_client, server['id'])
return timestamp
def _create_transfer(self, volume, transfer_client, volumes_client):
body = transfer_client.create_volume_transfer(volume_id=volume['id'])
transfer = body['transfer']
waiters.wait_for_volume_resource_status(volumes_client,
volume['id'],
'awaiting-transfer')
return transfer
def _accept_transfer(self, transfer, transfer_client, volumes_client):
_ = transfer_client.accept_volume_transfer(
transfer['id'], auth_key=transfer['auth_key'])
waiters.wait_for_volume_resource_status(volumes_client,
transfer['volume_id'],
'available')
def _delete_transfer(self, transfer, transfer_client, volumes_client):
_ = transfer_client.delete_volume_transfer(transfer['id'])
waiters.wait_for_volume_resource_status(volumes_client,
transfer['volume_id'],
'available')
@decorators.idempotent_id('a694dc4d-d11b-45cb-b268-62e76cc1b4f4')
@utils.services('compute', 'volume', 'image', 'network')
def test_create_accept_volume_transfer(self):
"""Verify the ability to transfer an encrypted volume:
* Create an encrypted volume from image
* Boot an instance from the volume and write a timestamp
* Transfer the volume to another project, then transfer it back
again to the original project (see comments in the code for why
this is done).
* Boot annother instance from the volume and read the timestamp
* Verify the timestamps match, and the volume has a new
encryption_key_id.
"""
# Create a bootable encrypted volume.
volume = self._create_encrypted_volume_from_image()
# Create an instance from the volume and write a timestamp.
timestamp_1 = self._create_or_get_timestamp(volume,
self.create_timestamp)
# Transfer the volume to another project.
transfer = self._create_transfer(volume,
self.client,
self.volumes_client)
self._accept_transfer(transfer,
self.alt_client,
self.alt_volumes_client)
# Transfer the volume back to the original project. This is done
# only because it's awkward in tempest to boot an instance and
# access it (to read the timestamp) in another project without
# setting up another security group and group rules.
transfer = self._create_transfer(volume,
self.alt_client,
self.alt_volumes_client)
self._accept_transfer(transfer, self.client, self.volumes_client)
# Create another instance from the volume and read the timestamp.
timestamp_2 = self._create_or_get_timestamp(volume,
self.get_timestamp)
self.assertEqual(timestamp_1, timestamp_2)
# Verify the volume has a new encryption_key_id.
encryption_key_id_1 = volume['encryption_key_id']
volume = self.volumes_client.show_volume(volume['id'])['volume']
encryption_key_id_2 = volume['encryption_key_id']
self.assertNotEqual(encryption_key_id_1, encryption_key_id_2)
@decorators.idempotent_id('00c04d27-b3c6-454c-a0b4-223a195c4a89')
@utils.services('compute', 'volume', 'image', 'network')
def test_create_delete_volume_transfer(self):
"""Verify the ability to cancel an encrypted volume transfer:
* Create an encrypted volume from image
* Boot an instance from the volume and write a timestamp
* Create and delete a volume transfer
* Boot annother instance from the volume and read the timestamp
* Verify the timestamps match, and the volume has a new
encryption_key_id.
"""
# Create a bootable encrypted volume.
volume = self._create_encrypted_volume_from_image()
# Create an instance from the volume and write a timestamp.
timestamp_1 = self._create_or_get_timestamp(volume,
self.create_timestamp)
# Create and then delete a transfer of the volume
transfer = self._create_transfer(volume,
self.client,
self.volumes_client)
self._delete_transfer(transfer, self.client, self.volumes_client)
# Create another instance from the volume and read the timestamp.
timestamp_2 = self._create_or_get_timestamp(volume,
self.get_timestamp)
self.assertEqual(timestamp_1, timestamp_2)
# Verify the volume has a new encryption_key_id.
encryption_key_id_1 = volume['encryption_key_id']
volume = self.volumes_client.show_volume(volume['id'])['volume']
encryption_key_id_2 = volume['encryption_key_id']
self.assertNotEqual(encryption_key_id_1, encryption_key_id_2)