Codebase list python-castellan / 73e9601
Update Barbican wrapper Changes to Barbican key manager to support all managed objects and API changes. Updates to the functional tests will follow in another patch in the interest of shorter code reviews. Change-Id: I2af3e7c2f16f31dcd2b4484a6537d3114bc4b3bb Kaitlin Farr 8 years ago
3 changed file(s) with 398 addition(s) and 79 deletion(s). Raw diff Collapse all Expand all
+251
-52
castellan/key_manager/barbican_key_manager.py less more
1515 """
1616 Key manager implementation for Barbican
1717 """
18 from barbicanclient import client as barbican_client
19 from barbicanclient import exceptions as barbican_exceptions
20 from keystoneclient.auth import token_endpoint
18 import time
19
20 from cryptography.hazmat import backends
21 from cryptography.hazmat.primitives import serialization
22 from cryptography import x509 as cryptography_x509
23 from keystoneclient.auth import identity
2124 from keystoneclient import session
2225 from oslo_config import cfg
2326 from oslo_log import log as logging
2427 from oslo_utils import excutils
2528
2629 from castellan.common import exception
30 from castellan.common.objects import key as key_base_class
31 from castellan.common.objects import opaque_data as op_data
32 from castellan.common.objects import passphrase
33 from castellan.common.objects import private_key as pri_key
34 from castellan.common.objects import public_key as pub_key
2735 from castellan.common.objects import symmetric_key as sym_key
36 from castellan.common.objects import x_509
2837 from castellan.key_manager import key_manager
2938 from castellan.openstack.common import _i18n as u
3039
40 from barbicanclient import client as barbican_client
41 from barbicanclient import exceptions as barbican_exceptions
3142 from six.moves import urllib
3243
3344 barbican_opts = [
3445 cfg.StrOpt('barbican_endpoint',
35 default='http://localhost:9311/',
36 help='Use this endpoint to connect to Barbican'),
46 help='Use this endpoint to connect to Barbican, for example: '
47 '"http://localhost:9311/"'),
3748 cfg.StrOpt('barbican_api_version',
38 default='v1',
39 help='Version of the Barbican API'),
49 help='Version of the Barbican API, for example: "v1"'),
50 cfg.StrOpt('auth_endpoint',
51 default='http://localhost:5000/v3',
52 help='Use this endpoint to connect to Keystone'),
53 cfg.IntOpt('retry_delay',
54 default=1,
55 help='Number of seconds to wait before retrying poll for key '
56 'creation completion'),
57 cfg.IntOpt('number_of_retries',
58 default=60,
59 help='Number of times to retry poll for key creation '
60 'completion'),
4061 ]
4162
4263 BARBICAN_OPT_GROUP = 'barbican'
4667
4768 class BarbicanKeyManager(key_manager.KeyManager):
4869 """Key Manager Interface that wraps the Barbican client API."""
70
71 _secret_type_dict = {
72 op_data.OpaqueData: 'opaque',
73 passphrase.Passphrase: 'passphrase',
74 pri_key.PrivateKey: 'private',
75 pub_key.PublicKey: 'public',
76 sym_key.SymmetricKey: 'symmetric',
77 x_509.X509: 'certificate'}
4978
5079 def __init__(self, configuration):
5180 self._barbican_client = None
6089 :param context: the user context for authentication
6190 :return: a Barbican Client object
6291 :raises Forbidden: if the context is None
92 :raises KeyManagerError: if context is missing tenant or
93 tenant is None
6394 """
6495
6596 # Confirm context is provided, if not raise forbidden
6899 LOG.error(msg)
69100 raise exception.Forbidden(msg)
70101
102 if not hasattr(context, 'tenant') or context.tenant is None:
103 msg = u._("Unable to create Barbican Client without tenant "
104 "attribute in context object.")
105 LOG.error(msg)
106 raise exception.KeyManagerError(msg)
107
71108 if self._barbican_client and self._current_context == context:
72 return self._barbican_client
109 return self._barbican_client
73110
74111 try:
75112 self._current_context = context
76 sess = self._get_keystone_session(context)
77
113 auth = self._get_keystone_auth(context)
114 sess = session.Session(auth=auth)
115
116 self._barbican_endpoint = self._get_barbican_endpoint(auth, sess)
78117 self._barbican_client = barbican_client.Client(
79118 session=sess,
80119 endpoint=self._barbican_endpoint)
83122 with excutils.save_and_reraise_exception():
84123 LOG.error(u._LE("Error creating Barbican client: %s"), e)
85124
86 self._base_url = self._create_base_url()
125 self._base_url = self._create_base_url(auth,
126 sess,
127 self._barbican_endpoint)
87128
88129 return self._barbican_client
89130
90 def _get_keystone_session(self, context):
91 sess = session.Session.load_from_conf_options(
92 self.conf, BARBICAN_OPT_GROUP)
93
94 self._barbican_endpoint = self.conf.barbican.barbican_endpoint
95
96 auth = token_endpoint.Token(self._barbican_endpoint,
97 context.auth_token)
98 sess.auth = auth
99 return sess
100
101 def _create_base_url(self):
131 def _get_keystone_auth(self, context):
132 # TODO(kfarr): support keystone v2
133 auth = identity.v3.Token(
134 auth_url=self.conf.barbican.auth_endpoint,
135 token=context.auth_token,
136 project_id=context.tenant,
137 domain_id=context.user_domain,
138 project_domain_id=context.project_domain)
139 return auth
140
141 def _get_barbican_endpoint(self, auth, sess):
142 if self.conf.barbican.barbican_endpoint:
143 return self.conf.barbican.barbican_endpoint
144 else:
145 service_parameters = {'service_type': 'key-manager',
146 'service_name': 'barbican',
147 'interface': 'public'}
148 return auth.get_endpoint(sess, **service_parameters)
149
150 def _create_base_url(self, auth, sess, endpoint):
151 if self.conf.barbican.barbican_api_version:
152 api_version = self.conf.barbican.barbican_api_version
153 else:
154 discovery = auth.get_discovery(sess, url=endpoint)
155 raw_data = discovery.raw_version_data()
156 if len(raw_data) == 0:
157 msg = u._LE(
158 "Could not find discovery information for %s") % endpoint
159 LOG.error(msg)
160 raise exception.KeyManagerError(msg)
161 latest_version = raw_data[-1]
162 api_version = latest_version.get('id')
163
102164 base_url = urllib.parse.urljoin(
103 self._barbican_endpoint, self.conf.barbican.barbican_api_version)
165 endpoint, api_version)
104166 return base_url
105167
106168 def create_key(self, context, algorithm, length, expiration=None):
107169 """Creates a symmetric key.
108170
109171 :param context: contains information of the user and the environment
110 for the request (castellan/context.py)
172 for the request (castellan/context.py)
111173 :param algorithm: the algorithm associated with the secret
112174 :param length: the bit length of the secret
113175 :param expiration: the date the key will expire
124186 bit_length=length,
125187 expiration=expiration)
126188 order_ref = key_order.submit()
127 order = barbican_client.orders.get(order_ref)
189 order = self._get_active_order(barbican_client, order_ref)
128190 return self._retrieve_secret_uuid(order.secret_ref)
129191 except (barbican_exceptions.HTTPAuthError,
130192 barbican_exceptions.HTTPClientError,
135197 def create_key_pair(self, context, algorithm, length, expiration=None):
136198 """Creates an asymmetric key pair.
137199
138 Not implemented yet.
139
140 :param context: contains information of the user and the environment
141 for the request (castellan/context.py)
200 :param context: contains information of the user and the environment
201 for the request (castellan/context.py)
142202 :param algorithm: the algorithm associated with the secret
143203 :param length: the bit length of the secret
144204 :param expiration: the date the key will expire
145 :return: TODO: the UUIDs of the new key, in the order (private, public)
205 :return: the UUIDs of the new key, in the order (private, public)
146206 :raises NotImplementedError: until implemented
147207 :raises HTTPAuthError: if key creation fails with 401
148208 :raises HTTPClientError: if key creation failes with 4xx
149209 :raises HTTPServerError: if key creation fails with 5xx
150210 """
151 raise NotImplementedError()
211 barbican_client = self._get_barbican_client(context)
212
213 try:
214 key_pair_order = barbican_client.orders.create_asymmetric(
215 algorithm=algorithm,
216 bit_length=length,
217 expiration=expiration)
218
219 order_ref = key_pair_order.submit()
220 order = self._get_active_order(barbican_client, order_ref)
221 container = barbican_client.containers.get(order.container_ref)
222
223 private_key_uuid = self._retrieve_secret_uuid(
224 container.secret_refs['private_key'])
225 public_key_uuid = self._retrieve_secret_uuid(
226 container.secret_refs['public_key'])
227 return private_key_uuid, public_key_uuid
228 except (barbican_exceptions.HTTPAuthError,
229 barbican_exceptions.HTTPClientError,
230 barbican_exceptions.HTTPServerError) as e:
231 with excutils.save_and_reraise_exception():
232 LOG.error(u._LE("Error creating key pair: %s"), e)
233
234 def _get_barbican_object(self, barbican_client, managed_object):
235 """Converts the Castellan managed_object to a Barbican secret."""
236 try:
237 algorithm = managed_object.algorithm
238 bit_length = managed_object.bit_length
239 except AttributeError:
240 algorithm = None
241 bit_length = None
242
243 secret_type = self._secret_type_dict.get(type(managed_object),
244 'opaque')
245 payload = self._get_normalized_payload(managed_object.get_encoded(),
246 secret_type)
247 secret = barbican_client.secrets.create(payload=payload,
248 algorithm=algorithm,
249 bit_length=bit_length,
250 secret_type=secret_type)
251 return secret
252
253 def _get_normalized_payload(self, encoded_bytes, secret_type):
254 """Normalizes the bytes of the object.
255
256 Barbican expects certificates, public keys, and private keys in PEM
257 format, but Castellan expects these objects to be DER encoded bytes
258 instead.
259 """
260 if secret_type == 'public':
261 key = serialization.load_der_public_key(
262 encoded_bytes,
263 backend=backends.default_backend())
264 return key.public_bytes(
265 encoding=serialization.Encoding.PEM,
266 format=serialization.PublicFormat.SubjectPublicKeyInfo)
267 elif secret_type == 'private':
268 key = serialization.load_der_private_key(
269 encoded_bytes,
270 backend=backends.default_backend(),
271 password=None)
272 return key.private_bytes(
273 encoding=serialization.Encoding.PEM,
274 format=serialization.PrivateFormat.PKCS8,
275 encryption_algorithm=serialization.NoEncryption())
276 elif secret_type == 'certificate':
277 cert = cryptography_x509.load_der_x509_certificate(
278 encoded_bytes,
279 backend=backends.default_backend())
280 return cert.public_bytes(encoding=serialization.Encoding.PEM)
281 else:
282 return encoded_bytes
152283
153284 def store(self, context, managed_object, expiration=None):
154285 """Stores (i.e., registers) an object with the key manager.
167298 barbican_client = self._get_barbican_client(context)
168299
169300 try:
170 if managed_object.algorithm:
171 algorithm = managed_object.algorithm
172 else:
173 algorithm = None
174 encoded_object = managed_object.get_encoded()
175 # TODO(kfarr) add support for objects other than symmetric keys
176 secret = barbican_client.secrets.create(payload=encoded_object,
177 algorithm=algorithm,
178 expiration=expiration)
301 secret = self._get_barbican_object(barbican_client,
302 managed_object)
303 secret.expiration = expiration
179304 secret_ref = secret.store()
180305 return self._retrieve_secret_uuid(secret_ref)
181306 except (barbican_exceptions.HTTPAuthError,
198323 base_url += '/'
199324 return urllib.parse.urljoin(base_url, "secrets/" + key_id)
200325
326 def _get_active_order(self, barbican_client, order_ref):
327 """Returns the order when it is active.
328
329 Barbican key creation is done asynchronously, so this loop continues
330 checking until the order is active or a timeout occurs.
331 """
332 active = u'ACTIVE'
333 number_of_retries = self.conf.barbican.number_of_retries
334 retry_delay = self.conf.barbican.retry_delay
335 order = barbican_client.orders.get(order_ref)
336 time.sleep(.25)
337 for n in range(number_of_retries):
338 if order.status != active:
339 kwargs = {'attempt': n,
340 'total': number_of_retries,
341 'status': order.status,
342 'active': active,
343 'delay': retry_delay}
344 msg = u._LI("Retry attempt #%(attempt)i out of %(total)i: "
345 "Order status is '%(status)s'. Waiting for "
346 "'%(active)s', will retry in %(delay)s "
347 "seconds")
348 LOG.info(msg, kwargs)
349 time.sleep(retry_delay)
350 order = barbican_client.orders.get(order_ref)
351 else:
352 return order
353 msg = u._LE("Exceeded retries: Failed to find '%(active)s' status "
354 "within %(num_retries)i retries") % {'active': active,
355 'num_retries':
356 number_of_retries}
357 LOG.error(msg)
358 raise exception.KeyManagerError(msg)
359
201360 def _retrieve_secret_uuid(self, secret_ref):
202361 """Retrieves the UUID of the secret from the secret_ref.
203362
212371 return secret_ref.rpartition('/')[2]
213372
214373 def _get_secret_data(self, secret):
215 """Retrieves the secret data given a secret and content_type.
374 """Retrieves the secret data.
375
376 Converts the Barbican secret to bytes suitable for a Castellan object.
377 If the secret is a public key, private key, or certificate, the secret
378 is expected to be in PEM format and will be converted to DER.
216379
217380 :param secret: the secret from barbican with the payload of data
218381 :returns: the secret data
219382 """
220 # TODO(kfarr) support other types of keys
221 return secret.payload
383 if secret.secret_type == 'public':
384 key = serialization.load_pem_public_key(
385 secret.payload,
386 backend=backends.default_backend())
387 return key.public_bytes(
388 encoding=serialization.Encoding.DER,
389 format=serialization.PublicFormat.SubjectPublicKeyInfo)
390 elif secret.secret_type == 'private':
391 key = serialization.load_pem_private_key(
392 secret.payload,
393 backend=backends.default_backend(),
394 password=None)
395 return key.private_bytes(
396 encoding=serialization.Encoding.DER,
397 format=serialization.PrivateFormat.PKCS8,
398 encryption_algorithm=serialization.NoEncryption())
399 elif secret.secret_type == 'certificate':
400 cert = cryptography_x509.load_pem_x509_certificate(
401 secret.payload,
402 backend=backends.default_backend())
403 return cert.public_bytes(encoding=serialization.Encoding.DER)
404 else:
405 return secret.payload
406
407 def _get_castellan_object(self, secret):
408 """Creates a Castellan managed object given the Barbican secret.
409
410 :param secret: the secret from barbican with the payload of data
411 :returns: the castellan object
412 """
413 secret_type = op_data.OpaqueData
414 for castellan_type, barbican_type in self._secret_type_dict.items():
415 if barbican_type == secret.secret_type:
416 secret_type = castellan_type
417
418 secret_data = self._get_secret_data(secret)
419
420 if issubclass(secret_type, key_base_class.Key):
421 return secret_type(secret.algorithm,
422 secret.bit_length,
423 secret_data)
424 else:
425 return secret_type(secret_data)
222426
223427 def _get_secret(self, context, key_id):
224428 """Returns the metadata of the secret.
225429
226430 :param context: contains information of the user and the environment
227 for the request (castellan/context.py)
431 for the request (castellan/context.py)
228432 :param key_id: UUID of the secret
229433 :return: the secret's metadata
230434 :raises HTTPAuthError: if object retrieval fails with 401
249453 Currently only supports retrieving symmetric keys.
250454
251455 :param context: contains information of the user and the environment
252 for the request (castellan/context.py)
456 for the request (castellan/context.py)
253457 :param managed_object_id: the UUID of the object to retrieve
254458 :return: SymmetricKey representation of the key
255459 :raises HTTPAuthError: if object retrieval fails with 401
258462 """
259463 try:
260464 secret = self._get_secret(context, managed_object_id)
261 secret_data = self._get_secret_data(secret)
262 # TODO(kfarr) add support for other objects
263 key = sym_key.SymmetricKey(secret.algorithm,
264 secret.bit_length,
265 secret_data)
266 return key
465 return self._get_castellan_object(secret)
267466 except (barbican_exceptions.HTTPAuthError,
268467 barbican_exceptions.HTTPClientError,
269468 barbican_exceptions.HTTPServerError) as e:
+146
-24
castellan/tests/unit/key_manager/test_barbican_key_manager.py less more
1616 Test cases for the barbican key manager.
1717 """
1818
19 from barbicanclient import exceptions as barbican_exceptions
1920 import mock
2021 from oslo_config import cfg
2122
2223 from castellan.common import exception
23 from castellan.common.objects import symmetric_key as key_manager_key
24 from castellan.common.objects import symmetric_key as sym_key
2425 from castellan.key_manager import barbican_key_manager
2526 from castellan.tests.unit.key_manager import test_key_manager
2627
4748 self.hex = ("0080f1429dbefae01b29a4d50cc5c5608bbc3c8ba0246aa42b424baa4"
4849 "534ae16")
4950 self.key_mgr._base_url = "http://host:9311/v1/"
51
52 self.key_mgr.conf.barbican.number_of_retries = 3
53 self.key_mgr.conf.barbican.retry_delay = 1
54
5055 self.addCleanup(self._restore)
5156
5257 def _restore(self):
5358 try:
5459 getattr(self, 'original_key')
55 key_manager_key.SymmetricKey = self.original_key
60 sym_key.SymmetricKey = self.original_key
5661 except AttributeError:
5762 return None
5863
7984 # Create order and assign return value
8085 order = mock.Mock()
8186 order.secret_ref = self.secret_ref
87 order.status = u'ACTIVE'
8288 self.mock_barbican.orders.get.return_value = order
8389
8490 # Create the key, get the UUID
8995 self.mock_barbican.orders.get.assert_called_once_with(order_ref_url)
9096 self.assertEqual(self.key_id, returned_uuid)
9197
92 def test_create_null_context(self):
98 def test_create_key_null_context(self):
9399 self.key_mgr._barbican_client = None
94100 self.assertRaises(exception.Forbidden,
95101 self.key_mgr.create_key, None, 'AES', 256)
102
103 def test_create_key_with_error(self):
104 key_order = mock.Mock()
105 self.mock_barbican.orders.create_key.return_value = key_order
106 key_order.submit = mock.Mock(
107 side_effect=barbican_exceptions.HTTPClientError('test error'))
108 self.assertRaises(barbican_exceptions.HTTPClientError,
109 self.key_mgr.create_key, self.ctxt, 'AES', 256)
110
111 def test_create_key_pair(self):
112 # Create order_ref_url and assign return value
113 order_ref_url = ("http://localhost:9311/v1/orders/"
114 "f45bf211-a917-4ead-9aec-1c91e52609df")
115 asym_order = mock.Mock()
116 self.mock_barbican.orders.create_asymmetric.return_value = asym_order
117 asym_order.submit.return_value = order_ref_url
118
119 # Create order and assign return value
120 order = mock.Mock()
121 container_id = "16caa8f4-dd34-4fb3-bf67-6c20533a30e4"
122 container_ref = ("http://localhost:9311/v1/containers/" + container_id)
123 order.container_ref = container_ref
124 order.status = u'ACTIVE'
125 self.mock_barbican.orders.get.return_value = order
126
127 # Create container and assign return value
128 container = mock.Mock()
129 public_key_id = "43ed09c3-e551-4c24-b612-e619abe9b534"
130 pub_key_ref = ("http://localhost:9311/v1/secrets/" + public_key_id)
131 private_key_id = "32a0bc60-4e10-4269-9f17-f49767e99586"
132 priv_key_ref = ("http://localhost:9311/v1/secrets/" + private_key_id)
133 container.secret_refs = {'public_key': pub_key_ref,
134 'private_key': priv_key_ref}
135 self.mock_barbican.containers.get.return_value = container
136
137 # Create the keys, get the UUIDs
138 returned_private_uuid, returned_public_uuid = (
139 self.key_mgr.create_key_pair(self.ctxt,
140 algorithm='RSA',
141 length=2048))
142
143 self.mock_barbican.orders.get.assert_called_once_with(order_ref_url)
144 self.mock_barbican.containers.get.assert_called_once_with(
145 container_ref)
146
147 self.mock_barbican.orders.get.assert_called_once_with(order_ref_url)
148 self.assertEqual(private_key_id, returned_private_uuid)
149 self.assertEqual(public_key_id, returned_public_uuid)
150
151 def test_create_key_pair_null_context(self):
152 self.key_mgr._barbican_client = None
153 self.assertRaises(exception.Forbidden,
154 self.key_mgr.create_key_pair, None, 'RSA', 2048)
155
156 def test_create_key_pair_with_error(self):
157 asym_order = mock.Mock()
158 self.mock_barbican.orders.create_asymmetric.return_value = asym_order
159 asym_order.submit = mock.Mock(
160 side_effect=barbican_exceptions.HTTPClientError('test error'))
161 self.assertRaises(barbican_exceptions.HTTPClientError,
162 self.key_mgr.create_key_pair, self.ctxt, 'RSA', 2048)
96163
97164 def test_delete_null_context(self):
98165 self.key_mgr._barbican_client = None
106173 def test_delete_unknown_key(self):
107174 self.assertRaises(exception.KeyManagerError,
108175 self.key_mgr.delete, self.ctxt, None)
176
177 def test_delete_with_error(self):
178 self.mock_barbican.secrets.delete = mock.Mock(
179 side_effect=barbican_exceptions.HTTPClientError('test error'))
180 self.assertRaises(barbican_exceptions.HTTPClientError,
181 self.key_mgr.delete, self.ctxt, self.key_id)
109182
110183 def test_get_key(self):
111184 original_secret_metadata = mock.Mock()
112185 original_secret_metadata.algorithm = mock.sentinel.alg
113186 original_secret_metadata.bit_length = mock.sentinel.bit
114 original_secret_data = mock.Mock()
187 original_secret_metadata.secret_type = 'symmetric'
188 original_secret_data = b'test key'
115189 original_secret_metadata.payload = original_secret_data
116190
117191 self.mock_barbican.secrets.get.return_value = original_secret_metadata
118192 key = self.key_mgr.get(self.ctxt, self.key_id)
119193
120194 self.get.assert_called_once_with(self.secret_ref)
121 self.assertEqual(key.get_encoded(), original_secret_data)
195 self.assertEqual(original_secret_data, key.get_encoded())
122196
123197 def test_get_null_context(self):
124198 self.key_mgr._barbican_client = None
129203 self.assertRaises(exception.KeyManagerError,
130204 self.key_mgr.get, self.ctxt, None)
131205
132 def test_store_key_base64(self):
206 def test_get_with_error(self):
207 self.mock_barbican.secrets.get = mock.Mock(
208 side_effect=barbican_exceptions.HTTPClientError('test error'))
209 self.assertRaises(barbican_exceptions.HTTPClientError,
210 self.key_mgr.get, self.ctxt, self.key_id)
211
212 def test_store_key(self):
133213 # Create Key to store
134214 secret_key = bytes(b'\x01\x02\xA0\xB3')
135 _key = key_manager_key.SymmetricKey('AES',
136 len(secret_key) * 8,
137 secret_key)
215 key_length = len(secret_key) * 8
216 _key = sym_key.SymmetricKey('AES',
217 key_length,
218 secret_key)
138219
139220 # Define the return values
140221 secret = mock.Mock()
145226 returned_uuid = self.key_mgr.store(self.ctxt, _key)
146227
147228 self.create.assert_called_once_with(algorithm='AES',
229 bit_length=key_length,
148230 payload=secret_key,
149 expiration=None)
231 secret_type='symmetric')
150232 self.assertEqual(self.key_id, returned_uuid)
151233
152 def test_store_key_plaintext(self):
153 # Create the plaintext key
154 secret_key_text = "This is a test text key."
155 _key = key_manager_key.SymmetricKey('AES',
156 len(secret_key_text) * 8,
157 secret_key_text)
158
159 # Store the Key
160 self.key_mgr.store(self.ctxt, _key)
161 self.create.assert_called_once_with(algorithm='AES',
162 payload=secret_key_text,
163 expiration=None)
164 self.assertEqual(0, self.store.call_count)
165
166234 def test_store_null_context(self):
167235 self.key_mgr._barbican_client = None
168236 self.assertRaises(exception.Forbidden,
169237 self.key_mgr.store, None, None)
238
239 def test_store_with_error(self):
240 self.mock_barbican.secrets.create = mock.Mock(
241 side_effect=barbican_exceptions.HTTPClientError('test error'))
242 secret_key = bytes(b'\x01\x02\xA0\xB3')
243 key_length = len(secret_key) * 8
244 _key = sym_key.SymmetricKey('AES',
245 key_length,
246 secret_key)
247 self.assertRaises(barbican_exceptions.HTTPClientError,
248 self.key_mgr.store, self.ctxt, _key)
249
250 def test_get_active_order(self):
251 order_ref_url = ("http://localhost:9311/v1/orders/"
252 "4fe939b7-72bc-49aa-bd1e-e979589858af")
253
254 pending_order = mock.Mock()
255 pending_order.status = u'PENDING'
256 pending_order.order_ref = order_ref_url
257
258 active_order = mock.Mock()
259 active_order.secret_ref = self.secret_ref
260 active_order.status = u'ACTIVE'
261 active_order.order_ref = order_ref_url
262
263 self.mock_barbican.orders.get.side_effect = [pending_order,
264 active_order]
265
266 self.key_mgr._get_active_order(self.mock_barbican, order_ref_url)
267
268 self.assertEqual(2, self.mock_barbican.orders.get.call_count)
269
270 calls = [mock.call(order_ref_url), mock.call(order_ref_url)]
271 self.mock_barbican.orders.get.assert_has_calls(calls)
272
273 def test_get_active_order_timeout(self):
274 order_ref_url = ("http://localhost:9311/v1/orders/"
275 "4fe939b7-72bc-49aa-bd1e-e979589858af")
276
277 number_of_retries = self.key_mgr.conf.barbican.number_of_retries
278
279 pending_order = mock.Mock()
280 pending_order.status = u'PENDING'
281 pending_order.order_ref = order_ref_url
282
283 self.mock_barbican.orders.get.return_value = pending_order
284
285 self.assertRaises(exception.KeyManagerError,
286 self.key_mgr._get_active_order,
287 self.mock_barbican,
288 order_ref_url)
289
290 self.assertEqual(number_of_retries + 1,
291 self.mock_barbican.orders.get.call_count)
+1
-3
castellan/tests/utils.py less more
1313 # under the License.
1414
1515
16 """
17 These utilility functions are borrowed from Barbican's testing utilites.
18 """
16 """These utility functions are borrowed from Barbican's testing utilities."""
1917
2018
2119 def get_certificate_der():