Adds barbican keymgr wrapper
Adds a barbican keymgr wrapper to the key manager interface in cinder. This allows barbican to be configured as the key manager for encryption keys in cinder. The wrapper translates calls from the existing key manager interface to python-barbicanclient. There are two new configuration variables: encryption_auth_url (the authentication URL for the encryption service) and encryption_api_url (the api URL for the encryption service). Change-Id: Ifd0d9b073d7f1c3c92d318c20a080b253008b50c Implements: blueprint encryption-with-barbican DocImpact
This commit is contained in:
Brianna Poulos
parent
4b973e90e0
commit
02f2b996bf
|
|
@ -0,0 +1,281 @@
|
|||
# Copyright (c) 2014 The Johns Hopkins University/Applied Physics Laboratory
|
||||
# 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.
|
||||
|
||||
"""
|
||||
Key manager implementation for Barbican
|
||||
"""
|
||||
|
||||
import array
|
||||
import base64
|
||||
import binascii
|
||||
|
||||
from barbicanclient import client as barbican_client
|
||||
from barbicanclient.common import auth
|
||||
from keystoneclient.v2_0 import client as keystone_client
|
||||
from oslo.config import cfg
|
||||
|
||||
from cinder import exception
|
||||
from cinder.keymgr import key as keymgr_key
|
||||
from cinder.keymgr import key_mgr
|
||||
from cinder.openstack.common import excutils
|
||||
from cinder.openstack.common.gettextutils import _ # noqa
|
||||
from cinder.openstack.common import log as logging
|
||||
|
||||
CONF = cfg.CONF
|
||||
CONF.import_opt('encryption_auth_url', 'cinder.keymgr.key_mgr', group='keymgr')
|
||||
CONF.import_opt('encryption_api_url', 'cinder.keymgr.key_mgr', group='keymgr')
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BarbicanKeyManager(key_mgr.KeyManager):
|
||||
"""Key Manager Interface that wraps the Barbican client API."""
|
||||
|
||||
def _create_connection(self, ctxt):
|
||||
"""Creates a connection to the Barbican service.
|
||||
|
||||
:param ctxt: the user context for authentication
|
||||
:return: a Barbican Connection object
|
||||
:throws NotAuthorized: if the ctxt is None
|
||||
"""
|
||||
|
||||
# Confirm context is provided, if not raise not authorized
|
||||
if not ctxt:
|
||||
msg = _("User is not authorized to use key manager.")
|
||||
LOG.error(msg)
|
||||
raise exception.NotAuthorized(msg)
|
||||
|
||||
try:
|
||||
endpoint = CONF.keymgr.encryption_auth_url
|
||||
keystone = keystone_client.Client(token=ctxt.auth_token,
|
||||
endpoint=endpoint)
|
||||
keystone_auth = auth.KeystoneAuthV2(keystone=keystone)
|
||||
keystone_auth._barbican_url = CONF.keymgr.encryption_api_url
|
||||
connection = barbican_client.Client(auth_plugin=keystone_auth)
|
||||
return connection
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error creating Barbican client: %s"), (e))
|
||||
|
||||
def create_key(self, ctxt, expiration=None, name='Cinder Volume Key',
|
||||
payload_content_type='application/octet-stream', mode='CBC',
|
||||
algorithm='AES', length=256):
|
||||
"""Creates a key.
|
||||
|
||||
:param ctxt: contains information of the user and the environment
|
||||
for the request (cinder/context.py)
|
||||
:param expiration: the date the key will expire
|
||||
:param name: a friendly name for the secret
|
||||
:param payload_content_type: the format/type of the secret data
|
||||
:param mode: the algorithm mode (e.g. CBC or CTR mode)
|
||||
:param algorithm: the algorithm associated with the secret
|
||||
:param length: the bit length of the secret
|
||||
|
||||
:return: the UUID of the new key
|
||||
:throws Exception: if key creation fails
|
||||
"""
|
||||
connection = self._create_connection(ctxt)
|
||||
|
||||
try:
|
||||
order_ref = connection.orders.create(name, payload_content_type,
|
||||
algorithm, length, mode,
|
||||
expiration)
|
||||
order = connection.orders.get(order_ref)
|
||||
secret_uuid = order.secret_ref.rpartition('/')[2]
|
||||
return secret_uuid
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error creating key: %s"), (e))
|
||||
|
||||
def store_key(self, ctxt, key, expiration=None, name='Cinder Volume Key',
|
||||
payload_content_type='application/octet-stream',
|
||||
payload_content_encoding='base64', algorithm='AES',
|
||||
bit_length=256, mode='CBC', from_copy=False):
|
||||
"""Stores (i.e., registers) a key with the key manager.
|
||||
|
||||
:param ctxt: contains information of the user and the environment for
|
||||
the request (cinder/context.py)
|
||||
:param key: the unencrypted secret data. Known as "payload" to the
|
||||
barbicanclient api
|
||||
:param expiration: the expiration time of the secret in ISO 8601
|
||||
format
|
||||
:param name: a friendly name for the key
|
||||
:param payload_content_type: the format/type of the secret data
|
||||
:param payload_content_encoding: the encoding of the secret data
|
||||
:param algorithm: the algorithm associated with this secret key
|
||||
:param bit_length: the bit length of this secret key
|
||||
:param mode: the algorithm mode used with this secret key
|
||||
:param from_copy: establishes whether the function is being used
|
||||
to copy a key. In case of the latter, it does not
|
||||
try to decode the key
|
||||
|
||||
:returns: the UUID of the stored key
|
||||
:throws Exception: if key storage fails
|
||||
"""
|
||||
connection = self._create_connection(ctxt)
|
||||
|
||||
try:
|
||||
if key.get_algorithm():
|
||||
algorithm = key.get_algorithm()
|
||||
if payload_content_type == 'text/plain':
|
||||
payload_content_encoding = None
|
||||
encoded_key = key.get_encoded()
|
||||
elif (payload_content_type == 'application/octet-stream' and
|
||||
not from_copy):
|
||||
key_list = key.get_encoded()
|
||||
string_key = ''.join(map(lambda byte: "%02x" % byte, key_list))
|
||||
encoded_key = base64.b64encode(binascii.unhexlify(string_key))
|
||||
else:
|
||||
encoded_key = key.get_encoded()
|
||||
secret_ref = connection.secrets.store(name, encoded_key,
|
||||
payload_content_type,
|
||||
payload_content_encoding,
|
||||
algorithm, bit_length, mode,
|
||||
expiration)
|
||||
secret_uuid = secret_ref.rpartition('/')[2]
|
||||
return secret_uuid
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error storing key: %s"), (e))
|
||||
|
||||
def copy_key(self, ctxt, key_id):
|
||||
"""Copies (i.e., clones) a key stored by barbican.
|
||||
|
||||
:param ctxt: contains information of the user and the environment for
|
||||
the request (cinder/context.py)
|
||||
:param key_id: the UUID of the key to copy
|
||||
:return: the UUID of the key copy
|
||||
:throws Exception: if key copying fails
|
||||
"""
|
||||
connection = self._create_connection(ctxt)
|
||||
|
||||
try:
|
||||
secret_ref = self._create_secret_ref(key_id, connection)
|
||||
meta = self._get_secret_metadata(ctxt, secret_ref)
|
||||
con_type = meta.content_types['default']
|
||||
secret_data = self._get_secret_data(ctxt, secret_ref,
|
||||
payload_content_type=con_type)
|
||||
key = keymgr_key.SymmetricKey(meta.algorithm, secret_data)
|
||||
copy_uuid = self.store_key(ctxt, key, meta.expiration,
|
||||
meta.name, con_type,
|
||||
'base64',
|
||||
meta.algorithm, meta.bit_length,
|
||||
meta.mode, True)
|
||||
return copy_uuid
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error copying key: %s"), (e))
|
||||
|
||||
def _create_secret_ref(self, key_id, connection):
|
||||
"""Creates the URL required for accessing a secret.
|
||||
|
||||
:param key_id: the UUID of the key to copy
|
||||
:param connection: barbican key manager object
|
||||
|
||||
:return: the URL of the requested secret
|
||||
"""
|
||||
return connection.base_url + "/secrets/" + key_id
|
||||
|
||||
def _get_secret_data(self, ctxt, secret_ref,
|
||||
payload_content_type='application/octet-stream'):
|
||||
"""Retrieves the secret data given a secret_ref and content_type.
|
||||
|
||||
:param ctxt: contains information of the user and the environment for
|
||||
the request (cinder/context.py)
|
||||
:param secret_ref: URL to access the secret
|
||||
:param payload_content_type: the format/type of the secret data
|
||||
|
||||
:returns: the secret data
|
||||
:throws Exception: if data cannot be retrieved
|
||||
"""
|
||||
connection = self._create_connection(ctxt)
|
||||
|
||||
try:
|
||||
generated_data = connection.secrets.decrypt(secret_ref,
|
||||
payload_content_type)
|
||||
if payload_content_type == 'application/octet-stream':
|
||||
secret_data = base64.b64encode(generated_data)
|
||||
else:
|
||||
secret_data = generated_data
|
||||
return secret_data
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error getting secret data: %s"), (e))
|
||||
|
||||
def _get_secret_metadata(self, ctxt, secret_ref):
|
||||
"""Creates the URL required for accessing a secret's metadata.
|
||||
|
||||
:param ctxt: contains information of the user and the environment for
|
||||
the request (cinder/context.py)
|
||||
:param secret_ref: URL to access the secret
|
||||
|
||||
:return: the secret's metadata
|
||||
:throws Exception: if there is an error retrieving the data
|
||||
"""
|
||||
|
||||
connection = self._create_connection(ctxt)
|
||||
|
||||
try:
|
||||
return connection.secrets.get(secret_ref)
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error getting secret metadata: %s"), (e))
|
||||
|
||||
def get_key(self, ctxt, key_id,
|
||||
payload_content_type='application/octet-stream'):
|
||||
"""Retrieves the specified key.
|
||||
|
||||
:param ctxt: contains information of the user and the environment for
|
||||
the request (cinder/context.py)
|
||||
:param key_id: the UUID of the key to retrieve
|
||||
:param payload_content_type: The format/type of the secret data
|
||||
|
||||
:return: SymmetricKey representation of the key
|
||||
:throws Exception: if key retrieval fails
|
||||
"""
|
||||
connection = self._create_connection(ctxt)
|
||||
|
||||
try:
|
||||
secret_ref = self._create_secret_ref(key_id, connection)
|
||||
secret_data = self._get_secret_data(ctxt, secret_ref,
|
||||
payload_content_type)
|
||||
if payload_content_type == 'application/octet-stream':
|
||||
# convert decoded string to list of unsigned ints for each byte
|
||||
secret = array.array('B',
|
||||
base64.b64decode(secret_data)).tolist()
|
||||
else:
|
||||
secret = secret_data
|
||||
meta = self._get_secret_metadata(ctxt, secret_ref)
|
||||
key = keymgr_key.SymmetricKey(meta.algorithm, secret)
|
||||
return key
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error getting key: %s"), (e))
|
||||
|
||||
def delete_key(self, ctxt, key_id):
|
||||
"""Deletes the specified key.
|
||||
|
||||
:param ctxt: contains information of the user and the environment for
|
||||
the request (cinder/context.py)
|
||||
:param key_id: the UUID of the key to delete
|
||||
:throws Exception: if key deletion fails
|
||||
"""
|
||||
connection = self._create_connection(ctxt)
|
||||
|
||||
try:
|
||||
secret_ref = self._create_secret_ref(key_id, connection)
|
||||
connection.secrets.delete(secret_ref)
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(_("Error deleting key: %s"), (e))
|
||||
|
|
@ -19,8 +19,21 @@ Key manager API
|
|||
|
||||
import abc
|
||||
|
||||
from oslo.config import cfg
|
||||
import six
|
||||
|
||||
encryption_opts = [
|
||||
cfg.StrOpt('encryption_auth_url',
|
||||
default='http://localhost:5000/v2.0',
|
||||
help='Authentication url for encryption service.'),
|
||||
cfg.StrOpt('encryption_api_url',
|
||||
default='http://localhost:9311/v1',
|
||||
help='Url for encryption service.'),
|
||||
]
|
||||
|
||||
CONF = cfg.CONF
|
||||
CONF.register_opts(encryption_opts, 'keymgr')
|
||||
|
||||
|
||||
@six.add_metaclass(abc.ABCMeta)
|
||||
class KeyManager(object):
|
||||
|
|
|
|||
|
|
@ -0,0 +1,265 @@
|
|||
# Copyright (c) 2014 The Johns Hopkins University/Applied Physics Laboratory
|
||||
# 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.
|
||||
|
||||
"""
|
||||
Test cases for the barbican key manager.
|
||||
"""
|
||||
|
||||
import array
|
||||
import base64
|
||||
import binascii
|
||||
|
||||
from barbicanclient import client as barbican_client
|
||||
from barbicanclient.common import auth
|
||||
from keystoneclient.v2_0 import client as keystone_client
|
||||
import mock
|
||||
from oslo.config import cfg
|
||||
|
||||
from cinder import exception
|
||||
from cinder.keymgr import barbican
|
||||
from cinder.keymgr import key as keymgr_key
|
||||
from cinder.tests.keymgr import test_key_mgr
|
||||
|
||||
CONF = cfg.CONF
|
||||
CONF.import_opt('encryption_auth_url', 'cinder.keymgr.key_mgr', group='keymgr')
|
||||
CONF.import_opt('encryption_api_url', 'cinder.keymgr.key_mgr', group='keymgr')
|
||||
|
||||
|
||||
class BarbicanKeyManagerTestCase(test_key_mgr.KeyManagerTestCase):
|
||||
|
||||
def _create_key_manager(self):
|
||||
return barbican.BarbicanKeyManager()
|
||||
|
||||
def setUp(self):
|
||||
super(BarbicanKeyManagerTestCase, self).setUp()
|
||||
|
||||
# Create fake auth_token
|
||||
self.ctxt = mock.Mock()
|
||||
self.ctxt.auth_token = "fake_token"
|
||||
|
||||
# Create mock keystone auth
|
||||
self._build_mock_auth()
|
||||
|
||||
# Create mock barbican client
|
||||
self._build_mock_barbican()
|
||||
|
||||
# Create mock keystone client
|
||||
self._build_mock_keystone()
|
||||
|
||||
# Create a key_id, secret_ref, pre_hex, and hex to use
|
||||
self.key_id = "d152fa13-2b41-42ca-a934-6c21566c0f40"
|
||||
self.secret_ref = self.key_mgr._create_secret_ref(self.key_id,
|
||||
self.mock_barbican)
|
||||
self.pre_hex = "AIDxQp2++uAbKaTVDMXFYIu8PIugJGqkK0JLqkU0rhY="
|
||||
self.hex = ("0080f1429dbefae01b29a4d50cc5c5608bbc3c8ba0246aa42b424baa4"
|
||||
"534ae16")
|
||||
self.addCleanup(self._restore)
|
||||
|
||||
def _restore(self):
|
||||
auth.KeystoneAuthV2 = self.original_auth
|
||||
barbican_client.Client = self.original_barbican
|
||||
keystone_client.Client = self.original_keystone
|
||||
if hasattr(self, 'original_key'):
|
||||
keymgr_key.SymmetricKey = self.original_key
|
||||
if hasattr(self, 'original_base64'):
|
||||
base64.b64encode = self.original_base64
|
||||
|
||||
def _build_mock_auth(self):
|
||||
self.mock_auth = mock.Mock()
|
||||
|
||||
def fake_keystone_auth(keystone):
|
||||
return self.mock_auth
|
||||
self.original_auth = auth.KeystoneAuthV2
|
||||
auth.KeystoneAuthV2 = fake_keystone_auth
|
||||
|
||||
def _build_mock_barbican(self):
|
||||
self.mock_barbican = mock.MagicMock(name='mock_barbican')
|
||||
self.mock_barbican.base_url = "http://localhost:9311/v1/None"
|
||||
|
||||
# Set commonly used methods
|
||||
self.get = self.mock_barbican.secrets.get
|
||||
self.decrypt = self.mock_barbican.secrets.decrypt
|
||||
self.delete = self.mock_barbican.secrets.delete
|
||||
self.store = self.mock_barbican.secrets.store
|
||||
|
||||
def fake_barbican_client(auth_plugin):
|
||||
return self.mock_barbican
|
||||
self.original_barbican = barbican_client.Client
|
||||
barbican_client.Client = fake_barbican_client
|
||||
|
||||
def _build_mock_keystone(self):
|
||||
self.mock_keystone = mock.Mock()
|
||||
|
||||
def fake_keystone_client(token, endpoint):
|
||||
self.barbican_auth_endpoint = endpoint
|
||||
return self.mock_keystone
|
||||
self.original_keystone = keystone_client.Client
|
||||
keystone_client.Client = fake_keystone_client
|
||||
|
||||
def _build_mock_symKey(self):
|
||||
self.mock_symKey = mock.Mock()
|
||||
|
||||
def fake_sym_key(alg, key):
|
||||
self.mock_symKey.get_encoded.return_value = key
|
||||
self.mock_symKey.get_algorithm.return_value = alg
|
||||
return self.mock_symKey
|
||||
self.original_key = keymgr_key.SymmetricKey
|
||||
keymgr_key.SymmetricKey = fake_sym_key
|
||||
|
||||
def _build_mock_base64(self):
|
||||
|
||||
def fake_base64_b64encode(string):
|
||||
return self.pre_hex
|
||||
|
||||
self.original_base64 = base64.b64encode
|
||||
base64.b64encode = fake_base64_b64encode
|
||||
|
||||
def test_conf_urls(self):
|
||||
# Create a Key
|
||||
self.key_mgr.create_key(self.ctxt)
|
||||
|
||||
# Confirm proper URL's were used
|
||||
self.assertEqual(self.barbican_auth_endpoint,
|
||||
CONF.keymgr.encryption_auth_url)
|
||||
self.assertEqual(self.mock_auth._barbican_url,
|
||||
CONF.keymgr.encryption_api_url)
|
||||
|
||||
def test_copy_key(self):
|
||||
# Create metadata for original secret
|
||||
original_secret_metadata = mock.Mock()
|
||||
original_secret_metadata.algorithm = 'fake_algorithm'
|
||||
original_secret_metadata.bit_length = 'fake_bit_length'
|
||||
original_secret_metadata.name = 'original_name'
|
||||
original_secret_metadata.expiration = 'fake_expiration'
|
||||
original_secret_metadata.mode = 'fake_mode'
|
||||
content_types = {'default': 'fake_type'}
|
||||
original_secret_metadata.content_types = content_types
|
||||
self.get.return_value = original_secret_metadata
|
||||
|
||||
# Create data for original secret
|
||||
original_secret_data = mock.Mock()
|
||||
self.decrypt.return_value = original_secret_data
|
||||
|
||||
# Create the mock key
|
||||
self._build_mock_symKey()
|
||||
|
||||
# Copy the original
|
||||
self.key_mgr.copy_key(self.ctxt, self.key_id)
|
||||
|
||||
# Assert proper methods were called
|
||||
self.get.assert_called_once_with(self.secret_ref)
|
||||
self.decrypt.assert_called_once_with(self.secret_ref,
|
||||
content_types['default'])
|
||||
self.store.assert_called_once_with(original_secret_metadata.name,
|
||||
self.mock_symKey.get_encoded(),
|
||||
content_types['default'],
|
||||
'base64',
|
||||
original_secret_metadata.algorithm,
|
||||
original_secret_metadata.bit_length,
|
||||
original_secret_metadata.mode,
|
||||
original_secret_metadata.expiration)
|
||||
|
||||
def test_copy_null_context(self):
|
||||
self.assertRaises(exception.NotAuthorized,
|
||||
self.key_mgr.copy_key, None, None)
|
||||
|
||||
def test_create_key(self):
|
||||
# Create order_ref_url and assign return value
|
||||
order_ref_url = ("http://localhost:9311/v1/None/orders/"
|
||||
"4fe939b7-72bc-49aa-bd1e-e979589858af")
|
||||
self.mock_barbican.orders.create.return_value = order_ref_url
|
||||
|
||||
# Create order and assign return value
|
||||
order = mock.Mock()
|
||||
order.secret_ref = self.secret_ref
|
||||
self.mock_barbican.orders.get.return_value = order
|
||||
|
||||
# Create the key, get the UUID
|
||||
returned_uuid = self.key_mgr.create_key(self.ctxt)
|
||||
|
||||
self.mock_barbican.orders.get.assert_called_once_with(order_ref_url)
|
||||
self.assertEqual(returned_uuid, self.key_id)
|
||||
|
||||
def test_create_null_context(self):
|
||||
self.assertRaises(exception.NotAuthorized,
|
||||
self.key_mgr.create_key, None)
|
||||
|
||||
def test_delete_null_context(self):
|
||||
self.assertRaises(exception.NotAuthorized,
|
||||
self.key_mgr.delete_key, None, None)
|
||||
|
||||
def test_delete_key(self):
|
||||
self.key_mgr.delete_key(self.ctxt, self.key_id)
|
||||
self.delete.assert_called_once_with(self.secret_ref)
|
||||
|
||||
def test_delete_unknown_key(self):
|
||||
self.assertRaises(TypeError, self.key_mgr.delete_key, self.ctxt, None)
|
||||
|
||||
def test_get_key(self):
|
||||
self._build_mock_base64()
|
||||
content_type = 'application/octet-stream'
|
||||
|
||||
key = self.key_mgr.get_key(self.ctxt, self.key_id, content_type)
|
||||
|
||||
self.decrypt.assert_called_once_with(self.secret_ref,
|
||||
content_type)
|
||||
encoded = array.array('B', binascii.unhexlify(self.hex)).tolist()
|
||||
self.assertEqual(key.get_encoded(), encoded)
|
||||
|
||||
def test_get_null_context(self):
|
||||
self.assertRaises(exception.NotAuthorized,
|
||||
self.key_mgr.get_key, None, None)
|
||||
|
||||
def test_get_unknown_key(self):
|
||||
self.assertRaises(TypeError, self.key_mgr.get_key, self.ctxt, None)
|
||||
|
||||
def test_store_key_base64(self):
|
||||
# Create Key to store
|
||||
secret_key = array.array('B', [0x01, 0x02, 0xA0, 0xB3]).tolist()
|
||||
_key = keymgr_key.SymmetricKey('AES', secret_key)
|
||||
|
||||
# Define the return value
|
||||
self.store.return_value = self.secret_ref
|
||||
|
||||
# Store the Key
|
||||
returned_uuid = self.key_mgr.store_key(self.ctxt, _key, bit_length=32)
|
||||
|
||||
self.store.assert_called_once_with('Cinder Volume Key',
|
||||
'AQKgsw==',
|
||||
'application/octet-stream',
|
||||
'base64',
|
||||
'AES', 32, 'CBC',
|
||||
None)
|
||||
self.assertEqual(returned_uuid, self.key_id)
|
||||
|
||||
def test_store_key_plaintext(self):
|
||||
# Create the plaintext key
|
||||
secret_key_text = "This is a test text key."
|
||||
_key = keymgr_key.SymmetricKey('AES', secret_key_text)
|
||||
|
||||
# Store the Key
|
||||
self.key_mgr.store_key(self.ctxt, _key,
|
||||
payload_content_type='text/plain',
|
||||
payload_content_encoding=None)
|
||||
self.store.assert_called_once_with('Cinder Volume Key',
|
||||
secret_key_text,
|
||||
'text/plain',
|
||||
None,
|
||||
'AES', 256, 'CBC',
|
||||
None)
|
||||
|
||||
def test_store_null_context(self):
|
||||
self.assertRaises(exception.NotAuthorized,
|
||||
self.key_mgr.store_key, None, None)
|
||||
|
|
@ -2182,6 +2182,17 @@
|
|||
#fixed_key=<None>
|
||||
|
||||
|
||||
#
|
||||
# Options defined in cinder.keymgr.key_mgr
|
||||
#
|
||||
|
||||
# Authentication url for encryption service. (string value)
|
||||
#encryption_auth_url=http://localhost:5000/v2.0
|
||||
|
||||
# Url for encryption service. (string value)
|
||||
#encryption_api_url=http://localhost:9311/v1
|
||||
|
||||
|
||||
[keystone_authtoken]
|
||||
|
||||
#
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ paramiko>=1.13.0
|
|||
Paste
|
||||
PasteDeploy>=1.5.0
|
||||
pycrypto>=2.6
|
||||
python-barbicanclient>=2.1.0
|
||||
python-glanceclient>=0.13.1
|
||||
python-novaclient>=2.17.0
|
||||
python-swiftclient>=2.0.2
|
||||
|
|
|
|||
Loading…
Reference in New Issue