Barbican implementation for Certificates
A Barbican implementation of CertManager and a placeholder implementation of CertGenerator (not supported yet). Change-Id: Icdbf883a733101c84b9a7bb933782ef166b929f7 Partially-implements: blueprint tls-data-security
This commit is contained in:
Adam Harwell
parent
ac4fe48813
commit
0f7e269821
|
|
@ -0,0 +1,102 @@
|
|||
# Copyright (c) 2014 Rackspace US, 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.
|
||||
|
||||
"""
|
||||
Common classes for Barbican certificate handling
|
||||
"""
|
||||
|
||||
from barbicanclient import client as barbican_client
|
||||
from keystoneclient.auth.identity import v3 as keystone_client
|
||||
from keystoneclient import session
|
||||
from oslo.config import cfg
|
||||
|
||||
from octavia.certificates.common import cert
|
||||
from octavia.openstack.common import excutils
|
||||
from octavia.openstack.common import gettextutils
|
||||
from octavia.openstack.common import log as logging
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
CONF = cfg.CONF
|
||||
CONF.import_group('keystone_authtoken', 'octavia.common.config')
|
||||
|
||||
|
||||
class BarbicanCert(cert.Cert):
|
||||
"""Representation of a Cert based on the Barbican CertificateContainer."""
|
||||
def __init__(self, cert_container):
|
||||
if not isinstance(cert_container,
|
||||
barbican_client.containers.CertificateContainer):
|
||||
raise TypeError(gettextutils._LE(
|
||||
"Retrieved Barbican Container is not of the correct type "
|
||||
"(certificate)."))
|
||||
self._cert_container = cert_container
|
||||
|
||||
def get_certificate(self):
|
||||
return self._cert_container.certificate.payload
|
||||
|
||||
def get_intermediates(self):
|
||||
return self._cert_container.intermediates.payload
|
||||
|
||||
def get_private_key(self):
|
||||
return self._cert_container.private_key.payload
|
||||
|
||||
def get_private_key_passphrase(self):
|
||||
return self._cert_container.private_key_passphrase.payload
|
||||
|
||||
|
||||
class BarbicanKeystoneAuth(object):
|
||||
_keystone_session = None
|
||||
_barbican_client = None
|
||||
|
||||
@classmethod
|
||||
def _get_keystone_session(cls):
|
||||
"""Initializes a Keystone session.
|
||||
|
||||
:return: a Keystone Session object
|
||||
:raises Exception: if the session cannot be established
|
||||
"""
|
||||
if not cls._keystone_session:
|
||||
try:
|
||||
kc = keystone_client.Password(
|
||||
auth_url=CONF.keystone_authtoken.auth_uri,
|
||||
username=CONF.keystone_authtoken.admin_user,
|
||||
password=CONF.keystone_authtoken.admin_password,
|
||||
project_id=CONF.keystone_authtoken.admin_project_id
|
||||
)
|
||||
cls._keystone_session = session.Session(auth=kc)
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(gettextutils._LE(
|
||||
"Error creating Keystone session: %s"), e)
|
||||
return cls._keystone_session
|
||||
|
||||
@classmethod
|
||||
def get_barbican_client(cls):
|
||||
"""Creates a Barbican client object.
|
||||
|
||||
:return: a Barbican Client object
|
||||
:raises Exception: if the client cannot be created
|
||||
"""
|
||||
if not cls._barbican_client:
|
||||
try:
|
||||
cls._barbican_client = barbican_client.Client(
|
||||
session=cls._get_keystone_session()
|
||||
)
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(gettextutils._LE(
|
||||
"Error creating Barbican client: %s"), e)
|
||||
return cls._barbican_client
|
||||
|
|
@ -29,4 +29,4 @@ CONF.register_opts(certgen_opts, group='certificates')
|
|||
|
||||
def API():
|
||||
cls = importutils.import_class(CONF.certgen.cert_generator_class)
|
||||
return cls()
|
||||
return cls()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,39 @@
|
|||
# Copyright (c) 2014 Rackspace US, 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.
|
||||
|
||||
"""
|
||||
Cert generator implementation for Barbican
|
||||
"""
|
||||
from octavia.certificates.generator import cert_gen
|
||||
from octavia.openstack.common import log as logging
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BarbicanCertGenerator(cert_gen.CertGenerator):
|
||||
"""Certificate Generator that wraps the Barbican client API."""
|
||||
|
||||
@staticmethod
|
||||
def sign_cert(csr, validity):
|
||||
"""Signs a certificate using our private CA based on the specified CSR.
|
||||
|
||||
:param csr: A Certificate Signing Request
|
||||
:param validity: Valid for <validity> seconds from the current time
|
||||
|
||||
:return: Signed certificate
|
||||
:raises Exception: if certificate signing fails
|
||||
"""
|
||||
raise NotImplementedError("Barbican does not yet support signing.")
|
||||
|
|
@ -29,4 +29,4 @@ CONF.register_opts(certmgr_opts, group='certificates')
|
|||
|
||||
def API():
|
||||
cls = importutils.import_class(CONF.certmgr.cert_manager_class)
|
||||
return cls()
|
||||
return cls()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,205 @@
|
|||
# Copyright (c) 2014 Rackspace US, 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.
|
||||
|
||||
"""
|
||||
Cert manager implementation for Barbican
|
||||
"""
|
||||
from oslo.config import cfg
|
||||
|
||||
from octavia.certificates.common import barbican as barbican_common
|
||||
from octavia.certificates.manager import cert_mgr
|
||||
from octavia.openstack.common import excutils
|
||||
from octavia.openstack.common import gettextutils
|
||||
from octavia.openstack.common import log as logging
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
CONF = cfg.CONF
|
||||
|
||||
|
||||
class BarbicanCertManager(cert_mgr.CertManager):
|
||||
"""Certificate Manager that wraps the Barbican client API."""
|
||||
@staticmethod
|
||||
def store_cert(certificate, private_key, intermediates=None,
|
||||
private_key_passphrase=None, expiration=None,
|
||||
name='Octavia TLS Cert', **kwargs):
|
||||
"""Stores a certificate in the certificate manager.
|
||||
|
||||
:param certificate: PEM encoded TLS certificate
|
||||
:param private_key: private key for the supplied certificate
|
||||
:param intermediates: ordered and concatenated intermediate certs
|
||||
:param private_key_passphrase: optional passphrase for the supplied key
|
||||
:param expiration: the expiration time of the cert in ISO 8601 format
|
||||
:param name: a friendly name for the cert
|
||||
|
||||
:returns: the container_ref of the stored cert
|
||||
:raises Exception: if certificate storage fails
|
||||
"""
|
||||
connection = barbican_common.BarbicanKeystoneAuth.get_barbican_client()
|
||||
|
||||
LOG.info(gettextutils._LI(
|
||||
"Storing certificate container '{0}' in Barbican."
|
||||
).format(name))
|
||||
|
||||
certificate_secret = None
|
||||
private_key_secret = None
|
||||
intermediates_secret = None
|
||||
pkp_secret = None
|
||||
|
||||
try:
|
||||
certificate_secret = connection.secrets.create(
|
||||
payload=certificate,
|
||||
expiration=expiration,
|
||||
name="Certificate"
|
||||
)
|
||||
private_key_secret = connection.secrets.create(
|
||||
payload=private_key,
|
||||
expiration=expiration,
|
||||
name="Private Key"
|
||||
)
|
||||
certificate_container = connection.containers.create_certificate(
|
||||
name=name,
|
||||
certificate=certificate_secret,
|
||||
private_key=private_key_secret
|
||||
)
|
||||
if intermediates:
|
||||
intermediates_secret = connection.secrets.create(
|
||||
payload=intermediates,
|
||||
expiration=expiration,
|
||||
name="Intermediates"
|
||||
)
|
||||
certificate_container.intermediates = intermediates_secret
|
||||
if private_key_passphrase:
|
||||
pkp_secret = connection.secrets.create(
|
||||
payload=private_key_passphrase,
|
||||
expiration=expiration,
|
||||
name="Private Key Passphrase"
|
||||
)
|
||||
certificate_container.private_key_passphrase = pkp_secret
|
||||
|
||||
certificate_container.store()
|
||||
return certificate_container.container_ref
|
||||
except Exception as e:
|
||||
for i in [certificate_secret, private_key_secret,
|
||||
intermediates_secret, pkp_secret]:
|
||||
if i and i.secret_ref:
|
||||
old_ref = i.secret_ref
|
||||
try:
|
||||
i.delete()
|
||||
LOG.info(gettextutils._LI(
|
||||
"Deleted secret {0} ({1}) during rollback."
|
||||
).format(i.name, old_ref))
|
||||
except Exception:
|
||||
LOG.warning(gettextutils._LW(
|
||||
"Failed to delete {0} ({1}) during rollback. This "
|
||||
"might not be a problem."
|
||||
).format(i.name, old_ref))
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(gettextutils._LE(
|
||||
"Error storing certificate data: {0}"
|
||||
).format(str(e)))
|
||||
|
||||
@staticmethod
|
||||
def get_cert(cert_ref, service_name='Octavia', resource_ref=None,
|
||||
check_only=False, **kwargs):
|
||||
"""Retrieves the specified cert and registers as a consumer.
|
||||
|
||||
:param cert_ref: the UUID of the cert to retrieve
|
||||
:param service_name: Friendly name for the consuming service
|
||||
:param resource_ref: Full HATEOAS reference to the consuming resource
|
||||
:param check_only: Read Certificate data without registering
|
||||
|
||||
:return: octavia.certificates.common.Cert representation of the
|
||||
certificate data
|
||||
:raises Exception: if certificate retrieval fails
|
||||
"""
|
||||
connection = barbican_common.BarbicanKeystoneAuth.get_barbican_client()
|
||||
|
||||
LOG.info(gettextutils._LI(
|
||||
"Loading certificate container {0} from Barbican."
|
||||
).format(cert_ref))
|
||||
try:
|
||||
if check_only:
|
||||
cert_container = connection.containers.get(
|
||||
container_ref=cert_ref
|
||||
)
|
||||
else:
|
||||
cert_container = connection.containers.register_consumer(
|
||||
container_ref=cert_ref,
|
||||
name=service_name,
|
||||
url=resource_ref
|
||||
)
|
||||
return barbican_common.BarbicanCert(cert_container)
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(gettextutils._LE(
|
||||
"Error getting {0}: {1}"
|
||||
).format(cert_ref, str(e)))
|
||||
|
||||
@staticmethod
|
||||
def delete_cert(cert_ref, service_name='Octavia', resource_ref=None,
|
||||
**kwargs):
|
||||
"""Deregister as a consumer for the specified cert.
|
||||
|
||||
:param cert_ref: the UUID of the cert to retrieve
|
||||
:param service_name: Friendly name for the consuming service
|
||||
:param resource_ref: Full HATEOAS reference to the consuming resource
|
||||
|
||||
:raises Exception: if deregistration fails
|
||||
"""
|
||||
connection = barbican_common.BarbicanKeystoneAuth.get_barbican_client()
|
||||
|
||||
LOG.info(gettextutils._LI(
|
||||
"Deregistering as a consumer of {0} in Barbican."
|
||||
).format(cert_ref))
|
||||
try:
|
||||
connection.containers.remove_consumer(
|
||||
container_ref=cert_ref,
|
||||
name=service_name,
|
||||
url=resource_ref
|
||||
)
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(gettextutils._LE(
|
||||
"Error deregistering as a consumer of {0}: {1}"
|
||||
).format(cert_ref, str(e)))
|
||||
|
||||
@staticmethod
|
||||
def _actually_delete_cert(cert_ref):
|
||||
"""Deletes the specified cert. Very dangerous. Do not recommend.
|
||||
|
||||
:param cert_ref: the UUID of the cert to delete
|
||||
:raises Exception: if certificate deletion fails
|
||||
"""
|
||||
connection = barbican_common.BarbicanKeystoneAuth.get_barbican_client()
|
||||
|
||||
LOG.info(gettextutils._LI(
|
||||
"Recursively deleting certificate container {0} from Barbican."
|
||||
).format(cert_ref))
|
||||
try:
|
||||
certificate_container = connection.containers.get(cert_ref)
|
||||
certificate_container.certificate.delete()
|
||||
if certificate_container.intermediates:
|
||||
certificate_container.intermediates.delete()
|
||||
if certificate_container.private_key_passphrase:
|
||||
certificate_container.private_key_passphrase.delete()
|
||||
certificate_container.private_key.delete()
|
||||
certificate_container.delete()
|
||||
except Exception as e:
|
||||
with excutils.save_and_reraise_exception():
|
||||
LOG.error(gettextutils._LE(
|
||||
"Error recursively deleting container {0}: {1}"
|
||||
).format(cert_ref, str(e)))
|
||||
|
|
@ -41,9 +41,10 @@ class CertManager(object):
|
|||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def get_cert(self, cert_ref, **kwargs):
|
||||
def get_cert(self, cert_ref, check_only=False, **kwargs):
|
||||
"""Retrieves the specified cert.
|
||||
|
||||
If check_only is True, don't perform any sort of registration.
|
||||
If the specified cert does not exist, a CertificateStorageException
|
||||
should be raised.
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -0,0 +1,127 @@
|
|||
# Copyright 2014 Rackspace
|
||||
#
|
||||
# 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 barbicanclient import client as barbican_client
|
||||
from keystoneclient import session
|
||||
import mock
|
||||
|
||||
import octavia.certificates.common.barbican as barbican_common
|
||||
import octavia.tests.unit.base as base
|
||||
|
||||
|
||||
class TestBarbicanAuth(base.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Reset the session and client
|
||||
barbican_common.BarbicanKeystoneAuth._keystone_session = None
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client = None
|
||||
|
||||
super(TestBarbicanAuth, self).setUp()
|
||||
|
||||
def test_get_keystone_client(self):
|
||||
# There should be no existing session
|
||||
self.assertIsNone(
|
||||
barbican_common.BarbicanKeystoneAuth._keystone_session
|
||||
)
|
||||
|
||||
# Get us a session
|
||||
ks1 = barbican_common.BarbicanKeystoneAuth._get_keystone_session()
|
||||
|
||||
# Our returned session should also be the saved session
|
||||
self.assertIsInstance(
|
||||
barbican_common.BarbicanKeystoneAuth._keystone_session,
|
||||
session.Session
|
||||
)
|
||||
self.assertIs(
|
||||
barbican_common.BarbicanKeystoneAuth._keystone_session,
|
||||
ks1
|
||||
)
|
||||
|
||||
# Getting the session again should return the same object
|
||||
ks2 = barbican_common.BarbicanKeystoneAuth._get_keystone_session()
|
||||
self.assertIs(ks1, ks2)
|
||||
|
||||
def test_get_barbican_client(self):
|
||||
# There should be no existing client
|
||||
self.assertIsNone(
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client
|
||||
)
|
||||
|
||||
# Mock out the keystone session and get the client
|
||||
barbican_common.BarbicanKeystoneAuth._keystone_session = (
|
||||
mock.MagicMock()
|
||||
)
|
||||
bc1 = barbican_common.BarbicanKeystoneAuth.get_barbican_client()
|
||||
|
||||
# Our returned client should also be the saved client
|
||||
self.assertIsInstance(
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client,
|
||||
barbican_client.Client
|
||||
)
|
||||
self.assertIs(
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client,
|
||||
bc1
|
||||
)
|
||||
|
||||
# Getting the session again should return the same object
|
||||
bc2 = barbican_common.BarbicanKeystoneAuth.get_barbican_client()
|
||||
self.assertIs(bc1, bc2)
|
||||
|
||||
|
||||
class TestBarbicanCert(base.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Certificate data
|
||||
self.certificate = "My Certificate"
|
||||
self.intermediates = "My Intermediates"
|
||||
self.private_key = "My Private Key"
|
||||
self.private_key_passphrase = "My Private Key Passphrase"
|
||||
|
||||
self.certificate_secret = barbican_client.secrets.Secret(
|
||||
api=mock.MagicMock(),
|
||||
payload=self.certificate
|
||||
)
|
||||
self.intermediates_secret = barbican_client.secrets.Secret(
|
||||
api=mock.MagicMock(),
|
||||
payload=self.intermediates
|
||||
)
|
||||
self.private_key_secret = barbican_client.secrets.Secret(
|
||||
api=mock.MagicMock(),
|
||||
payload=self.private_key
|
||||
)
|
||||
self.private_key_passphrase_secret = barbican_client.secrets.Secret(
|
||||
api=mock.MagicMock(),
|
||||
payload=self.private_key_passphrase
|
||||
)
|
||||
|
||||
super(TestBarbicanCert, self).setUp()
|
||||
|
||||
def test_barbican_cert(self):
|
||||
container = barbican_client.containers.CertificateContainer(
|
||||
api=mock.MagicMock(),
|
||||
certificate=self.certificate_secret,
|
||||
intermediates=self.intermediates_secret,
|
||||
private_key=self.private_key_secret,
|
||||
private_key_passphrase=self.private_key_passphrase_secret
|
||||
)
|
||||
# Create a cert
|
||||
cert = barbican_common.BarbicanCert(
|
||||
cert_container=container
|
||||
)
|
||||
|
||||
# Validate the cert functions
|
||||
self.assertEqual(cert.get_certificate(), self.certificate)
|
||||
self.assertEqual(cert.get_intermediates(), self.intermediates)
|
||||
self.assertEqual(cert.get_private_key(), self.private_key)
|
||||
self.assertEqual(cert.get_private_key_passphrase(),
|
||||
self.private_key_passphrase)
|
||||
|
|
@ -0,0 +1,66 @@
|
|||
# Copyright 2014 Rackspace
|
||||
#
|
||||
# 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.
|
||||
import uuid
|
||||
|
||||
import mock
|
||||
from OpenSSL import crypto
|
||||
|
||||
import octavia.certificates.generator.barbican as barbican_cert_gen
|
||||
import octavia.tests.unit.base as base
|
||||
|
||||
|
||||
class TestBarbicanGenerator(base.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Make a fake Order and contents
|
||||
self.barbican_endpoint = 'http://localhost:9311/v1'
|
||||
self.container_uuid = uuid.uuid4()
|
||||
|
||||
# TODO(rm_work): fill this section, right now it is placeholder data
|
||||
self.order_uuid = uuid.uuid4()
|
||||
self.order_ref = '{0}/orders/{1}'.format(
|
||||
self.barbican_endpoint, self.container_uuid
|
||||
)
|
||||
|
||||
key = crypto.PKey()
|
||||
key.generate_key(crypto.TYPE_RSA, 1024)
|
||||
req = crypto.X509Req()
|
||||
req.set_pubkey(key)
|
||||
self.certificate_signing_request = crypto.dump_certificate_request(
|
||||
crypto.FILETYPE_PEM, req
|
||||
)
|
||||
|
||||
order = mock.Mock()
|
||||
self.order = order
|
||||
|
||||
super(TestBarbicanGenerator, self).setUp()
|
||||
|
||||
def test_sign_cert(self):
|
||||
# TODO(rm_work): Update this test when Barbican supports this, right
|
||||
# now this is all guesswork
|
||||
self.skipTest("Barbican does not yet support signing.")
|
||||
|
||||
# Mock out the client
|
||||
bc = mock.MagicMock()
|
||||
bc.orders.create.return_value = self.order
|
||||
barbican_cert_gen.BarbicanCertGenerator._barbican_client = bc
|
||||
|
||||
# Attempt to order a cert signing
|
||||
barbican_cert_gen.BarbicanCertGenerator.sign_cert(
|
||||
csr=self.certificate_signing_request
|
||||
)
|
||||
|
||||
# create order should be called once
|
||||
# should get back a valid order
|
||||
bc.orders.create.assert_called_once_with()
|
||||
|
|
@ -0,0 +1,238 @@
|
|||
# Copyright 2014 Rackspace
|
||||
#
|
||||
# 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.
|
||||
import uuid
|
||||
|
||||
from barbicanclient import containers
|
||||
from barbicanclient import secrets
|
||||
import mock
|
||||
|
||||
import octavia.certificates.common.barbican as barbican_common
|
||||
import octavia.certificates.common.cert as cert
|
||||
import octavia.certificates.manager.barbican as barbican_cert_mgr
|
||||
import octavia.tests.unit.base as base
|
||||
|
||||
|
||||
class TestBarbicanManager(base.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Make a fake Container and contents
|
||||
self.barbican_endpoint = 'http://localhost:9311/v1'
|
||||
self.container_uuid = uuid.uuid4()
|
||||
|
||||
self.container_ref = '{0}/containers/{1}'.format(
|
||||
self.barbican_endpoint, self.container_uuid
|
||||
)
|
||||
|
||||
self.name = 'My Fancy Cert'
|
||||
self.private_key = mock.Mock(spec=secrets.Secret)
|
||||
self.certificate = mock.Mock(spec=secrets.Secret)
|
||||
self.intermediates = mock.Mock(spec=secrets.Secret)
|
||||
self.private_key_passphrase = mock.Mock(spec=secrets.Secret)
|
||||
|
||||
container = mock.Mock(spec=containers.CertificateContainer)
|
||||
container.container_ref = self.container_ref
|
||||
container.name = self.name
|
||||
container.private_key = self.private_key
|
||||
container.certificate = self.certificate
|
||||
container.intermediates = self.intermediates
|
||||
container.private_key_passphrase = self.private_key_passphrase
|
||||
self.container = container
|
||||
|
||||
self.empty_container = mock.Mock(spec=containers.CertificateContainer)
|
||||
|
||||
self.secret1 = mock.Mock(spec=secrets.Secret)
|
||||
self.secret2 = mock.Mock(spec=secrets.Secret)
|
||||
self.secret3 = mock.Mock(spec=secrets.Secret)
|
||||
self.secret4 = mock.Mock(spec=secrets.Secret)
|
||||
|
||||
super(TestBarbicanManager, self).setUp()
|
||||
|
||||
def test_store_cert(self):
|
||||
# Mock out the client
|
||||
bc = mock.MagicMock()
|
||||
bc.containers.create_certificate.return_value = self.empty_container
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client = bc
|
||||
|
||||
# Attempt to store a cert
|
||||
barbican_cert_mgr.BarbicanCertManager.store_cert(
|
||||
certificate=self.certificate,
|
||||
private_key=self.private_key,
|
||||
intermediates=self.intermediates,
|
||||
private_key_passphrase=self.private_key_passphrase,
|
||||
name=self.name
|
||||
)
|
||||
|
||||
# create_secret should be called four times with our data
|
||||
calls = [
|
||||
mock.call(payload=self.certificate, expiration=None,
|
||||
name=mock.ANY),
|
||||
mock.call(payload=self.private_key, expiration=None,
|
||||
name=mock.ANY),
|
||||
mock.call(payload=self.intermediates, expiration=None,
|
||||
name=mock.ANY),
|
||||
mock.call(payload=self.private_key_passphrase, expiration=None,
|
||||
name=mock.ANY)
|
||||
]
|
||||
bc.secrets.create.assert_has_calls(calls, any_order=True)
|
||||
|
||||
# create_certificate should be called once
|
||||
self.assertEqual(bc.containers.create_certificate.call_count, 1)
|
||||
|
||||
# Container should be stored once
|
||||
self.empty_container.store.assert_called_once_with()
|
||||
|
||||
def test_store_cert_failure(self):
|
||||
# Mock out the client
|
||||
bc = mock.MagicMock()
|
||||
bc.containers.create_certificate.return_value = self.empty_container
|
||||
test_secrets = [
|
||||
self.secret1,
|
||||
self.secret2,
|
||||
self.secret3,
|
||||
self.secret4
|
||||
]
|
||||
bc.secrets.create.side_effect = test_secrets
|
||||
self.empty_container.store.side_effect = ValueError()
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client = bc
|
||||
|
||||
# Attempt to store a cert
|
||||
self.assertRaises(
|
||||
ValueError,
|
||||
barbican_cert_mgr.BarbicanCertManager.store_cert,
|
||||
certificate=self.certificate,
|
||||
private_key=self.private_key,
|
||||
intermediates=self.intermediates,
|
||||
private_key_passphrase=self.private_key_passphrase,
|
||||
name=self.name
|
||||
)
|
||||
|
||||
# create_secret should be called four times with our data
|
||||
calls = [
|
||||
mock.call(payload=self.certificate, expiration=None,
|
||||
name=mock.ANY),
|
||||
mock.call(payload=self.private_key, expiration=None,
|
||||
name=mock.ANY),
|
||||
mock.call(payload=self.intermediates, expiration=None,
|
||||
name=mock.ANY),
|
||||
mock.call(payload=self.private_key_passphrase, expiration=None,
|
||||
name=mock.ANY)
|
||||
]
|
||||
bc.secrets.create.assert_has_calls(calls, any_order=True)
|
||||
|
||||
# create_certificate should be called once
|
||||
self.assertEqual(bc.containers.create_certificate.call_count, 1)
|
||||
|
||||
# Container should be stored once
|
||||
self.empty_container.store.assert_called_once_with()
|
||||
|
||||
# All secrets should be deleted (or at least an attempt made)
|
||||
for s in test_secrets:
|
||||
s.delete.assert_called_once_with()
|
||||
|
||||
def test_get_cert(self):
|
||||
# Mock out the client
|
||||
bc = mock.MagicMock()
|
||||
bc.containers.register_consumer.return_value = self.container
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client = bc
|
||||
|
||||
# Get the container data
|
||||
data = barbican_cert_mgr.BarbicanCertManager.get_cert(
|
||||
cert_ref=self.container_ref,
|
||||
resource_ref=self.container_ref,
|
||||
service_name='Octavia'
|
||||
)
|
||||
|
||||
# 'register_consumer' should be called once with the container_ref
|
||||
bc.containers.register_consumer.assert_called_once_with(
|
||||
container_ref=self.container_ref,
|
||||
url=self.container_ref,
|
||||
name='Octavia'
|
||||
)
|
||||
|
||||
# The returned data should be a Cert object with the correct values
|
||||
self.assertIsInstance(data, cert.Cert)
|
||||
self.assertEqual(data.get_private_key(),
|
||||
self.private_key.payload)
|
||||
self.assertEqual(data.get_certificate(),
|
||||
self.certificate.payload)
|
||||
self.assertEqual(data.get_intermediates(),
|
||||
self.intermediates.payload)
|
||||
self.assertEqual(data.get_private_key_passphrase(),
|
||||
self.private_key_passphrase.payload)
|
||||
|
||||
def test_get_cert_no_registration(self):
|
||||
# Mock out the client
|
||||
bc = mock.MagicMock()
|
||||
bc.containers.get.return_value = self.container
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client = bc
|
||||
|
||||
# Get the container data
|
||||
data = barbican_cert_mgr.BarbicanCertManager.get_cert(
|
||||
cert_ref=self.container_ref, check_only=True
|
||||
)
|
||||
|
||||
# 'get' should be called once with the container_ref
|
||||
bc.containers.get.assert_called_once_with(
|
||||
container_ref=self.container_ref
|
||||
)
|
||||
|
||||
# The returned data should be a Cert object with the correct values
|
||||
self.assertIsInstance(data, cert.Cert)
|
||||
self.assertEqual(data.get_private_key(),
|
||||
self.private_key.payload)
|
||||
self.assertEqual(data.get_certificate(),
|
||||
self.certificate.payload)
|
||||
self.assertEqual(data.get_intermediates(),
|
||||
self.intermediates.payload)
|
||||
self.assertEqual(data.get_private_key_passphrase(),
|
||||
self.private_key_passphrase.payload)
|
||||
|
||||
def test_delete_cert(self):
|
||||
# Mock out the client
|
||||
bc = mock.MagicMock()
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client = bc
|
||||
|
||||
# Attempt to deregister as a consumer
|
||||
barbican_cert_mgr.BarbicanCertManager.delete_cert(
|
||||
cert_ref=self.container_ref,
|
||||
resource_ref=self.container_ref,
|
||||
service_name='Octavia'
|
||||
)
|
||||
|
||||
# remove_consumer should be called once with the container_ref
|
||||
bc.containers.remove_consumer.assert_called_once_with(
|
||||
container_ref=self.container_ref,
|
||||
url=self.container_ref,
|
||||
name='Octavia'
|
||||
)
|
||||
|
||||
def test_actually_delete_cert(self):
|
||||
# Mock out the client
|
||||
bc = mock.MagicMock()
|
||||
bc.containers.get.return_value = self.container
|
||||
barbican_common.BarbicanKeystoneAuth._barbican_client = bc
|
||||
|
||||
# Attempt to store a cert
|
||||
barbican_cert_mgr.BarbicanCertManager._actually_delete_cert(
|
||||
cert_ref=self.container_ref
|
||||
)
|
||||
|
||||
# All secrets should be deleted
|
||||
self.container.certificate.delete.assert_called_once_with()
|
||||
self.container.private_key.delete.assert_called_once_with()
|
||||
self.container.intermediates.delete.assert_called_once_with()
|
||||
self.container.private_key_passphrase.delete.assert_called_once_with()
|
||||
|
||||
# Container should be deleted once
|
||||
self.container.delete.assert_called_once_with()
|
||||
|
|
@ -27,6 +27,7 @@ oslo.config>=1.4.0.0a3
|
|||
oslo.db>=0.4.0 # Apache-2.0
|
||||
oslo.messaging>=1.4.0.0a3
|
||||
oslo.rootwrap>=1.3.0.0a1
|
||||
python-barbicanclient>=3.0
|
||||
python-keystoneclient>=0.11.1
|
||||
python-novaclient>=2.17.0
|
||||
posix_ipc
|
||||
|
|
|
|||
Loading…
Reference in New Issue