Add Barbican key manager
Adds the first usable key manager plugin to Castellan. While there is an
implementation of a mock key manager in the test directories, it is used
only for testing.
This code is based on the barbican key manager code in Nova written by
Brianna Poulos. See: https://review.openstack.org/#/c/104001/
The Barbican API version info will be read from a config option until
the Barbican Version API is fixed. See fix-version-api blueprint.
Implements: blueprint add-barbican-key-manager
Co-authored-by: Brianna Poulos <brianna.poulos@jhuapl.edu>
Change-Id: Ia27cd831f42c6b027778240b3396b1c4149dc689
Kaitlin Farr
8 years ago
53 | 53 | |
54 | 54 | class Forbidden(CastellanException): |
55 | 55 | message = u._("You are not authorized to complete this action.") |
56 | ||
57 | ||
58 | class KeyManagerError(CastellanException): | |
59 | message = u._("Key manager error: %(reason)s") |
18 | 18 | |
19 | 19 | key_manager_opts = [ |
20 | 20 | cfg.StrOpt('api_class', |
21 | default='castellan.key_manager.barbican_key_manager' | |
22 | '.BarbicanKeyManager', | |
21 | 23 | help='The full class name of the key manager API class'), |
22 | 24 | ] |
23 | 25 |
0 | # Copyright (c) The Johns Hopkins University/Applied Physics Laboratory | |
1 | # All Rights Reserved. | |
2 | # | |
3 | # Licensed under the Apache License, Version 2.0 (the "License"); you may | |
4 | # not use this file except in compliance with the License. You may obtain | |
5 | # a copy of the License at | |
6 | # | |
7 | # http://www.apache.org/licenses/LICENSE-2.0 | |
8 | # | |
9 | # Unless required by applicable law or agreed to in writing, software | |
10 | # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT | |
11 | # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the | |
12 | # License for the specific language governing permissions and limitations | |
13 | # under the License. | |
14 | ||
15 | """ | |
16 | Key manager implementation for Barbican | |
17 | """ | |
18 | from barbicanclient import client as barbican_client | |
19 | from barbicanclient import exceptions as barbican_exceptions | |
20 | from keystoneclient.auth import token_endpoint | |
21 | from keystoneclient import session | |
22 | from oslo_config import cfg | |
23 | from oslo_log import log as logging | |
24 | from oslo_utils import excutils | |
25 | ||
26 | from castellan.common import exception | |
27 | from castellan.key_manager import key_manager | |
28 | from castellan.key_manager import symmetric_key as key_manager_key | |
29 | from castellan.openstack.common import _i18n as u | |
30 | ||
31 | from six.moves import urllib | |
32 | ||
33 | barbican_opts = [ | |
34 | cfg.StrOpt('barbican_endpoint', | |
35 | default='http://localhost:9311/', | |
36 | help='Use this endpoint to connect to Barbican'), | |
37 | cfg.StrOpt('api_version', | |
38 | default='v1', | |
39 | help='Version of the Barbican API'), | |
40 | ] | |
41 | ||
42 | CONF = cfg.CONF | |
43 | BARBICAN_OPT_GROUP = 'barbican' | |
44 | ||
45 | CONF.register_opts(barbican_opts, group=BARBICAN_OPT_GROUP) | |
46 | ||
47 | session.Session.register_conf_options(CONF, BARBICAN_OPT_GROUP) | |
48 | ||
49 | LOG = logging.getLogger(__name__) | |
50 | ||
51 | ||
52 | class BarbicanKeyManager(key_manager.KeyManager): | |
53 | """Key Manager Interface that wraps the Barbican client API.""" | |
54 | ||
55 | def __init__(self): | |
56 | self._barbican_client = None | |
57 | self._base_url = None | |
58 | ||
59 | def _get_barbican_client(self, context): | |
60 | """Creates a client to connect to the Barbican service. | |
61 | ||
62 | :param context: the user context for authentication | |
63 | :return: a Barbican Client object | |
64 | :raises Forbidden: if the context is None | |
65 | """ | |
66 | ||
67 | # Confirm context is provided, if not raise forbidden | |
68 | if not context: | |
69 | msg = u._("User is not authorized to use key manager.") | |
70 | LOG.error(msg) | |
71 | raise exception.Forbidden(msg) | |
72 | ||
73 | if self._barbican_client and self._current_context == context: | |
74 | return self._barbican_client | |
75 | ||
76 | try: | |
77 | self._current_context = context | |
78 | sess = self._get_keystone_session(context) | |
79 | ||
80 | self._barbican_client = barbican_client.Client( | |
81 | session=sess, | |
82 | endpoint=self._barbican_endpoint) | |
83 | ||
84 | except Exception as e: | |
85 | with excutils.save_and_reraise_exception(): | |
86 | LOG.error(u._LE("Error creating Barbican client: %s"), e) | |
87 | ||
88 | self._base_url = self._create_base_url() | |
89 | ||
90 | return self._barbican_client | |
91 | ||
92 | def _get_keystone_session(self, context): | |
93 | sess = session.Session.load_from_conf_options( | |
94 | CONF, BARBICAN_OPT_GROUP) | |
95 | ||
96 | self._barbican_endpoint = CONF.barbican.barbican_endpoint | |
97 | ||
98 | auth = token_endpoint.Token(self._barbican_endpoint, | |
99 | context.auth_token) | |
100 | sess.auth = auth | |
101 | return sess | |
102 | ||
103 | def _create_base_url(self): | |
104 | base_url = urllib.parse.urljoin(self._barbican_endpoint, | |
105 | CONF.barbican.api_version) | |
106 | return base_url | |
107 | ||
108 | def create_key(self, context, algorithm, length, expiration=None): | |
109 | """Creates a key. | |
110 | ||
111 | :param context: contains information of the user and the environment | |
112 | for the request (castellan/context.py) | |
113 | :param algorithm: the algorithm associated with the secret | |
114 | :param length: the bit length of the secret | |
115 | :param expiration: the date the key will expire | |
116 | :return: the UUID of the new key | |
117 | :raises HTTPAuthError: if key creation fails with 401 | |
118 | :raises HTTPClientError: if key creation failes with 4xx | |
119 | :raises HTTPServerError: if key creation fails with 5xx | |
120 | """ | |
121 | barbican_client = self._get_barbican_client(context) | |
122 | ||
123 | try: | |
124 | key_order = barbican_client.orders.create_key( | |
125 | algorithm=algorithm, | |
126 | bit_length=length, | |
127 | expiration=expiration) | |
128 | order_ref = key_order.submit() | |
129 | order = barbican_client.orders.get(order_ref) | |
130 | return self._retrieve_secret_uuid(order.secret_ref) | |
131 | except (barbican_exceptions.HTTPAuthError, | |
132 | barbican_exceptions.HTTPClientError, | |
133 | barbican_exceptions.HTTPServerError) as e: | |
134 | with excutils.save_and_reraise_exception(): | |
135 | LOG.error(u._LE("Error creating key: %s"), e) | |
136 | ||
137 | def store_key(self, context, key, expiration=None): | |
138 | """Stores (i.e., registers) a key with the key manager. | |
139 | ||
140 | :param context: contains information of the user and the environment | |
141 | for the request (castellan/context.py) | |
142 | :param key: the unencrypted secret data. Known as "payload" to the | |
143 | barbicanclient api | |
144 | :param expiration: the expiration time of the secret in ISO 8601 | |
145 | format | |
146 | :returns: the UUID of the stored key | |
147 | :raises HTTPAuthError: if key creation fails with 401 | |
148 | :raises HTTPClientError: if key creation failes with 4xx | |
149 | :raises HTTPServerError: if key creation fails with 5xx | |
150 | """ | |
151 | barbican_client = self._get_barbican_client(context) | |
152 | ||
153 | try: | |
154 | if key.get_algorithm(): | |
155 | algorithm = key.get_algorithm() | |
156 | encoded_key = key.get_encoded() | |
157 | # TODO(kfarr) add support for objects other than symmetric keys | |
158 | secret = barbican_client.secrets.create(payload=encoded_key, | |
159 | algorithm=algorithm, | |
160 | expiration=expiration) | |
161 | secret_ref = secret.store() | |
162 | return self._retrieve_secret_uuid(secret_ref) | |
163 | except (barbican_exceptions.HTTPAuthError, | |
164 | barbican_exceptions.HTTPClientError, | |
165 | barbican_exceptions.HTTPServerError) as e: | |
166 | with excutils.save_and_reraise_exception(): | |
167 | LOG.error(u._LE("Error storing key: %s"), e) | |
168 | ||
169 | def copy_key(self, context, key_id): | |
170 | """Copies (i.e., clones) a key stored by barbican. | |
171 | ||
172 | :param context: contains information of the user and the environment | |
173 | for the request (castellan/context.py) | |
174 | :param key_id: the UUID of the key to copy | |
175 | :return: the UUID of the key copy | |
176 | :raises HTTPAuthError: if key creation fails with 401 | |
177 | :raises HTTPClientError: if key creation failes with 4xx | |
178 | :raises HTTPServerError: if key creation fails with 5xx | |
179 | """ | |
180 | ||
181 | try: | |
182 | secret = self._get_secret(context, key_id) | |
183 | secret_data = self._get_secret_data(secret) | |
184 | # TODO(kfarr) modify to support other types of keys | |
185 | key = key_manager_key.SymmetricKey(secret.algorithm, secret_data) | |
186 | copy_uuid = self.store_key(context, key, secret.expiration) | |
187 | return copy_uuid | |
188 | except (barbican_exceptions.HTTPAuthError, | |
189 | barbican_exceptions.HTTPClientError, | |
190 | barbican_exceptions.HTTPServerError) as e: | |
191 | with excutils.save_and_reraise_exception(): | |
192 | LOG.error(u._LE("Error copying key: %s"), e) | |
193 | ||
194 | def _create_secret_ref(self, key_id): | |
195 | """Creates the URL required for accessing a secret. | |
196 | ||
197 | :param key_id: the UUID of the key to copy | |
198 | :return: the URL of the requested secret | |
199 | """ | |
200 | if not key_id: | |
201 | msg = "Key ID is None" | |
202 | raise exception.KeyManagerError(msg) | |
203 | base_url = self._base_url | |
204 | if base_url[-1] != '/': | |
205 | base_url += '/' | |
206 | return urllib.parse.urljoin(base_url, "secrets/" + key_id) | |
207 | ||
208 | def _retrieve_secret_uuid(self, secret_ref): | |
209 | """Retrieves the UUID of the secret from the secret_ref. | |
210 | ||
211 | :param secret_ref: the href of the secret | |
212 | :return: the UUID of the secret | |
213 | """ | |
214 | ||
215 | # The secret_ref is assumed to be of a form similar to | |
216 | # http://host:9311/v1/secrets/d152fa13-2b41-42ca-a934-6c21566c0f40 | |
217 | # with the UUID at the end. This command retrieves everything | |
218 | # after the last '/', which is the UUID. | |
219 | return secret_ref.rpartition('/')[2] | |
220 | ||
221 | def _get_secret_data(self, secret): | |
222 | """Retrieves the secret data given a secret and content_type. | |
223 | ||
224 | :param secret: the secret from barbican with the payload of data | |
225 | :returns: the secret data | |
226 | """ | |
227 | # TODO(kfarr) support other types of keys | |
228 | return secret.payload | |
229 | ||
230 | def _get_secret(self, context, key_id): | |
231 | """Returns the metadata of the secret. | |
232 | ||
233 | :param context: contains information of the user and the environment | |
234 | for the request (castellan/context.py) | |
235 | :param key_id: UUID of the secret | |
236 | :return: the secret's metadata | |
237 | :raises HTTPAuthError: if key creation fails with 401 | |
238 | :raises HTTPClientError: if key creation failes with 4xx | |
239 | :raises HTTPServerError: if key creation fails with 5xx | |
240 | """ | |
241 | ||
242 | barbican_client = self._get_barbican_client(context) | |
243 | ||
244 | try: | |
245 | secret_ref = self._create_secret_ref(key_id) | |
246 | return barbican_client.secrets.get(secret_ref) | |
247 | except (barbican_exceptions.HTTPAuthError, | |
248 | barbican_exceptions.HTTPClientError, | |
249 | barbican_exceptions.HTTPServerError) as e: | |
250 | with excutils.save_and_reraise_exception(): | |
251 | LOG.error(u._LE("Error getting secret metadata: %s"), e) | |
252 | ||
253 | def get_key(self, context, key_id): | |
254 | """Retrieves the specified key. | |
255 | ||
256 | :param context: contains information of the user and the environment | |
257 | for the request (castellan/context.py) | |
258 | :param key_id: the UUID of the key to retrieve | |
259 | :return: SymmetricKey representation of the key | |
260 | :raises HTTPAuthError: if key creation fails with 401 | |
261 | :raises HTTPClientError: if key creation failes with 4xx | |
262 | :raises HTTPServerError: if key creation fails with 5xx | |
263 | """ | |
264 | try: | |
265 | secret = self._get_secret(context, key_id) | |
266 | secret_data = self._get_secret_data(secret) | |
267 | # TODO(kfarr) add support for other objects | |
268 | key = key_manager_key.SymmetricKey(secret.algorithm, secret_data) | |
269 | return key | |
270 | except (barbican_exceptions.HTTPAuthError, | |
271 | barbican_exceptions.HTTPClientError, | |
272 | barbican_exceptions.HTTPServerError) as e: | |
273 | with excutils.save_and_reraise_exception(): | |
274 | LOG.error(u._LE("Error getting key: %s"), e) | |
275 | ||
276 | def delete_key(self, context, key_id): | |
277 | """Deletes the specified key. | |
278 | ||
279 | :param context: contains information of the user and the environment | |
280 | for the request (castellan/context.py) | |
281 | :param key_id: the UUID of the key to delete | |
282 | :raises HTTPAuthError: if key creation fails with 401 | |
283 | :raises HTTPClientError: if key creation failes with 4xx | |
284 | :raises HTTPServerError: if key creation fails with 5xx | |
285 | """ | |
286 | barbican_client = self._get_barbican_client(context) | |
287 | ||
288 | try: | |
289 | secret_ref = self._create_secret_ref(key_id) | |
290 | barbican_client.secrets.delete(secret_ref) | |
291 | except (barbican_exceptions.HTTPAuthError, | |
292 | barbican_exceptions.HTTPClientError, | |
293 | barbican_exceptions.HTTPServerError) as e: | |
294 | with excutils.save_and_reraise_exception(): | |
295 | LOG.error(u._LE("Error deleting key: %s"), e) |
30 | 30 | """ |
31 | 31 | |
32 | 32 | @abc.abstractmethod |
33 | def create_key(self, context, algorithm='AES', length=256, | |
34 | expiration=None, **kwargs): | |
33 | def create_key(self, context, algorithm, length, | |
34 | expiration=None): | |
35 | 35 | """Creates a key. |
36 | 36 | |
37 | 37 | This method creates a key and returns the key's UUID. If the specified |
41 | 41 | pass |
42 | 42 | |
43 | 43 | @abc.abstractmethod |
44 | def store_key(self, context, key, expiration=None, **kwargs): | |
44 | def store_key(self, context, key, expiration=None): | |
45 | 45 | """Stores (i.e., registers) a key with the key manager. |
46 | 46 | |
47 | 47 | This method stores the specified key and returns its UUID that |
52 | 52 | pass |
53 | 53 | |
54 | 54 | @abc.abstractmethod |
55 | def copy_key(self, context, key_id, **kwargs): | |
55 | def copy_key(self, context, key_id): | |
56 | 56 | """Copies (i.e., clones) a key stored by the key manager. |
57 | 57 | |
58 | 58 | This method copies the specified key and returns the copy's UUID. If |
67 | 67 | pass |
68 | 68 | |
69 | 69 | @abc.abstractmethod |
70 | def get_key(self, context, key_id, **kwargs): | |
70 | def get_key(self, context, key_id): | |
71 | 71 | """Retrieves the specified key. |
72 | 72 | |
73 | 73 | Implementations should verify that the caller has permissions to |
83 | 83 | pass |
84 | 84 | |
85 | 85 | @abc.abstractmethod |
86 | def delete_key(self, context, key_id, **kwargs): | |
86 | def delete_key(self, context, key_id): | |
87 | 87 | """Deletes the specified key. |
88 | 88 | |
89 | 89 | Implementations should verify that the caller has permission to delete |
0 | # Copyright (c) The Johns Hopkins University/Applied Physics Laboratory | |
1 | # All Rights Reserved. | |
2 | # | |
3 | # Licensed under the Apache License, Version 2.0 (the "License"); you may | |
4 | # not use this file except in compliance with the License. You may obtain | |
5 | # a copy of the License at | |
6 | # | |
7 | # http://www.apache.org/licenses/LICENSE-2.0 | |
8 | # | |
9 | # Unless required by applicable law or agreed to in writing, software | |
10 | # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT | |
11 | # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the | |
12 | # License for the specific language governing permissions and limitations | |
13 | # under the License. | |
14 | ||
15 | """ | |
16 | Test cases for the barbican key manager. | |
17 | """ | |
18 | ||
19 | import array | |
20 | ||
21 | import mock | |
22 | ||
23 | from castellan.common import exception | |
24 | from castellan.key_manager import barbican_key_manager | |
25 | from castellan.key_manager import symmetric_key as key_manager_key | |
26 | from castellan.tests.key_manager import test_key_manager | |
27 | ||
28 | ||
29 | class BarbicanKeyManagerTestCase(test_key_manager.KeyManagerTestCase): | |
30 | ||
31 | def _create_key_manager(self): | |
32 | return barbican_key_manager.BarbicanKeyManager() | |
33 | ||
34 | def setUp(self): | |
35 | super(BarbicanKeyManagerTestCase, self).setUp() | |
36 | ||
37 | # Create fake auth_token | |
38 | self.ctxt = mock.Mock() | |
39 | self.ctxt.auth_token = "fake_token" | |
40 | ||
41 | # Create mock barbican client | |
42 | self._build_mock_barbican() | |
43 | ||
44 | # Create a key_id, secret_ref, pre_hex, and hex to use | |
45 | self.key_id = "d152fa13-2b41-42ca-a934-6c21566c0f40" | |
46 | self.secret_ref = ("http://host:9311/v1/secrets/" + self.key_id) | |
47 | self.pre_hex = "AIDxQp2++uAbKaTVDMXFYIu8PIugJGqkK0JLqkU0rhY=" | |
48 | self.hex = ("0080f1429dbefae01b29a4d50cc5c5608bbc3c8ba0246aa42b424baa4" | |
49 | "534ae16") | |
50 | self.key_mgr._base_url = "http://host:9311/v1/" | |
51 | self.addCleanup(self._restore) | |
52 | ||
53 | def _restore(self): | |
54 | try: | |
55 | getattr(self, 'original_key') | |
56 | key_manager_key.SymmetricKey = self.original_key | |
57 | except AttributeError: | |
58 | return None | |
59 | ||
60 | def _build_mock_barbican(self): | |
61 | self.mock_barbican = mock.MagicMock(name='mock_barbican') | |
62 | ||
63 | # Set commonly used methods | |
64 | self.get = self.mock_barbican.secrets.get | |
65 | self.delete = self.mock_barbican.secrets.delete | |
66 | self.store = self.mock_barbican.secrets.store | |
67 | self.create = self.mock_barbican.secrets.create | |
68 | ||
69 | self.key_mgr._barbican_client = self.mock_barbican | |
70 | self.key_mgr._current_context = self.ctxt | |
71 | ||
72 | def _build_mock_symKey(self): | |
73 | self.mock_symKey = mock.Mock() | |
74 | ||
75 | def fake_sym_key(alg, key): | |
76 | self.mock_symKey.get_encoded.return_value = key | |
77 | self.mock_symKey.get_algorithm.return_value = alg | |
78 | return self.mock_symKey | |
79 | self.original_key = key_manager_key.SymmetricKey | |
80 | key_manager_key.SymmetricKey = fake_sym_key | |
81 | ||
82 | def test_copy_key(self): | |
83 | # Create metadata for original secret | |
84 | original_secret_metadata = mock.Mock() | |
85 | original_secret_metadata.algorithm = mock.sentinel.alg | |
86 | original_secret_metadata.bit_length = mock.sentinel.bit | |
87 | original_secret_metadata.name = mock.sentinel.name | |
88 | original_secret_metadata.expiration = mock.sentinel.expiration | |
89 | original_secret_metadata.mode = mock.sentinel.mode | |
90 | content_types = {'default': 'fake_type'} | |
91 | original_secret_metadata.content_types = content_types | |
92 | original_secret_data = mock.Mock() | |
93 | original_secret_metadata.payload = original_secret_data | |
94 | ||
95 | # Create href for copied secret | |
96 | copied_secret = mock.Mock() | |
97 | copied_secret.store.return_value = ( | |
98 | 'http://http://host:9311/v1/secrets/uuid') | |
99 | ||
100 | # Set get and create return values | |
101 | self.get.return_value = original_secret_metadata | |
102 | self.create.return_value = copied_secret | |
103 | ||
104 | # Create the mock key | |
105 | self._build_mock_symKey() | |
106 | ||
107 | # Copy the original | |
108 | self.key_mgr.copy_key(self.ctxt, self.key_id) | |
109 | ||
110 | # Assert proper methods were called | |
111 | self.get.assert_called_once_with(self.secret_ref) | |
112 | self.create.assert_called_once_with( | |
113 | payload=self.mock_symKey.get_encoded(), | |
114 | algorithm=mock.sentinel.alg, | |
115 | expiration=mock.sentinel.expiration) | |
116 | copied_secret.store.assert_called_once_with() | |
117 | ||
118 | def test_copy_null_context(self): | |
119 | self.key_mgr._barbican_client = None | |
120 | self.assertRaises(exception.Forbidden, | |
121 | self.key_mgr.copy_key, None, self.key_id) | |
122 | ||
123 | def test_create_key(self): | |
124 | # Create order_ref_url and assign return value | |
125 | order_ref_url = ("http://localhost:9311/v1/orders/" | |
126 | "4fe939b7-72bc-49aa-bd1e-e979589858af") | |
127 | key_order = mock.Mock() | |
128 | self.mock_barbican.orders.create_key.return_value = key_order | |
129 | key_order.submit.return_value = order_ref_url | |
130 | ||
131 | # Create order and assign return value | |
132 | order = mock.Mock() | |
133 | order.secret_ref = self.secret_ref | |
134 | self.mock_barbican.orders.get.return_value = order | |
135 | ||
136 | # Create the key, get the UUID | |
137 | returned_uuid = self.key_mgr.create_key(self.ctxt, | |
138 | algorithm='AES', | |
139 | length=256) | |
140 | ||
141 | self.mock_barbican.orders.get.assert_called_once_with(order_ref_url) | |
142 | self.assertEqual(self.key_id, returned_uuid) | |
143 | ||
144 | def test_create_null_context(self): | |
145 | self.key_mgr._barbican_client = None | |
146 | self.assertRaises(exception.Forbidden, | |
147 | self.key_mgr.create_key, None, 'AES', 256) | |
148 | ||
149 | def test_delete_null_context(self): | |
150 | self.key_mgr._barbican_client = None | |
151 | self.assertRaises(exception.Forbidden, | |
152 | self.key_mgr.delete_key, None, self.key_id) | |
153 | ||
154 | def test_delete_key(self): | |
155 | self.key_mgr.delete_key(self.ctxt, self.key_id) | |
156 | self.delete.assert_called_once_with(self.secret_ref) | |
157 | ||
158 | def test_delete_unknown_key(self): | |
159 | self.assertRaises(exception.KeyManagerError, | |
160 | self.key_mgr.delete_key, self.ctxt, None) | |
161 | ||
162 | def test_get_key(self): | |
163 | original_secret_metadata = mock.Mock() | |
164 | original_secret_metadata.algorithm = mock.sentinel.alg | |
165 | original_secret_metadata.bit_length = mock.sentinel.bit | |
166 | original_secret_data = mock.Mock() | |
167 | original_secret_metadata.payload = original_secret_data | |
168 | ||
169 | self.mock_barbican.secrets.get.return_value = original_secret_metadata | |
170 | key = self.key_mgr.get_key(self.ctxt, self.key_id) | |
171 | ||
172 | self.get.assert_called_once_with(self.secret_ref) | |
173 | self.assertEqual(key.get_encoded(), original_secret_data) | |
174 | ||
175 | def test_get_null_context(self): | |
176 | self.key_mgr._barbican_client = None | |
177 | self.assertRaises(exception.Forbidden, | |
178 | self.key_mgr.get_key, None, self.key_id) | |
179 | ||
180 | def test_get_unknown_key(self): | |
181 | self.assertRaises(exception.KeyManagerError, | |
182 | self.key_mgr.get_key, self.ctxt, None) | |
183 | ||
184 | def test_store_key_base64(self): | |
185 | # Create Key to store | |
186 | secret_key = array.array('B', [0x01, 0x02, 0xA0, 0xB3]).tolist() | |
187 | _key = key_manager_key.SymmetricKey('AES', secret_key) | |
188 | ||
189 | # Define the return values | |
190 | secret = mock.Mock() | |
191 | self.create.return_value = secret | |
192 | secret.store.return_value = self.secret_ref | |
193 | ||
194 | # Store the Key | |
195 | returned_uuid = self.key_mgr.store_key(self.ctxt, _key) | |
196 | ||
197 | self.create.assert_called_once_with(algorithm='AES', | |
198 | payload=secret_key, | |
199 | expiration=None) | |
200 | self.assertEqual(self.key_id, returned_uuid) | |
201 | ||
202 | def test_store_key_plaintext(self): | |
203 | # Create the plaintext key | |
204 | secret_key_text = "This is a test text key." | |
205 | _key = key_manager_key.SymmetricKey('AES', secret_key_text) | |
206 | ||
207 | # Store the Key | |
208 | self.key_mgr.store_key(self.ctxt, _key) | |
209 | self.create.assert_called_once_with(algorithm='AES', | |
210 | payload=secret_key_text, | |
211 | expiration=None) | |
212 | self.assertEqual(0, self.store.call_count) | |
213 | ||
214 | def test_store_null_context(self): | |
215 | self.key_mgr._barbican_client = None | |
216 | self.assertRaises(exception.Forbidden, | |
217 | self.key_mgr.store_key, None, None) |