Add Certificate controller for TLS support.

The Certificate controller has 2 operations:

1. POST

Generate X509 certificate using bay's CA cert.
Below is an example of Certificate POST API request using magnum command:

Example Request:

    curl -X POST -H 'Content-Type: application/json' \
        -d '{"bay_uuid": "<bay_uuid>", "csr": "<csr>"}' \
        http://localhost:9511/v1/certificates

This creates a X509 certificate signed by the given bay's CA and returns
it. No database information is stored in Magnum against it. For each POST
request, a new certificate is generated.

Example Response:

    {"bay_uuid": "<bay_uuid>", "csr": "<csr>",
     "pem": "<pem encoded certifiacte>"}

2. GET
Fetches the CA cert associated with a bay. Below is an example of CA GET
API request using magnum command:

Example Request:

    curl -X GET http://localhost:9511/v1/certificates/<bay_uuid>

This fetches stored CA cert for the given Bay, which can be used to validate
any client and node certificates signed by the Bay's CA. The value for each
is fetched from Barbican or Magnum db based on the different configuration
for storage of certificates.

Example Response:

    {"bay_uuid": "<bay_uuid>", "pem": "<pem encoded certifiacte>"}

Co-Authored-By: Andrew Melton <andrew.melton@rackspace.com>

Change-Id: I4b72cc1e1bddc7a7c7eeb0ab22d3769a666ccb2b
Partially-Implements: bp secure-kubernetes
changes/79/214179/10
Madhuri 7 years ago committed by Andrew Melton
parent d6916e8bb1
commit cefc5a9ac3
  1. 5
      etc/magnum/policy.json
  2. 11
      magnum/api/controllers/v1/__init__.py
  3. 155
      magnum/api/controllers/v1/certificate.py
  4. 2
      magnum/cmd/conductor.py
  5. 7
      magnum/conductor/api.py
  6. 45
      magnum/conductor/handlers/ca_conductor.py
  7. 12
      magnum/conductor/handlers/common/cert_manager.py
  8. 6
      magnum/objects/__init__.py
  9. 45
      magnum/objects/certificate.py
  10. 4
      magnum/tests/unit/api/controllers/test_root.py
  11. 185
      magnum/tests/unit/api/controllers/v1/test_certificate.py
  12. 8
      magnum/tests/unit/api/utils.py
  13. 74
      magnum/tests/unit/conductor/handlers/common/test_cert_manager.py
  14. 57
      magnum/tests/unit/conductor/handlers/test_ca_conductor.py
  15. 16
      magnum/tests/unit/conductor/test_rpcapi.py

@ -51,5 +51,8 @@
"container:detail": "rule:default",
"container:get": "rule:default",
"container:get_all": "rule:default",
"container:update": "rule:default"
"container:update": "rule:default",
"certificate:create": "rule:default",
"certificate:get": "rule:default"
}

@ -27,6 +27,7 @@ from magnum.api.controllers import base as controllers_base
from magnum.api.controllers import link
from magnum.api.controllers.v1 import bay
from magnum.api.controllers.v1 import baymodel
from magnum.api.controllers.v1 import certificate
from magnum.api.controllers.v1 import container
from magnum.api.controllers.v1 import node
from magnum.api.controllers.v1 import pod
@ -105,6 +106,9 @@ class V1(controllers_base.APIBase):
x509keypairs = [link.Link]
certificates = [link.Link]
"""Links to the certificates resource"""
@staticmethod
def convert():
v1 = V1()
@ -160,6 +164,12 @@ class V1(controllers_base.APIBase):
pecan.request.host_url,
'x509keypairs', '',
bookmark=True)]
v1.certificates = [link.Link.make_link('self', pecan.request.host_url,
'certificates', ''),
link.Link.make_link('bookmark',
pecan.request.host_url,
'certificates', '',
bookmark=True)]
return v1
@ -174,6 +184,7 @@ class Controller(rest.RestController):
rcs = rc.ReplicationControllersController()
services = service.ServicesController()
x509keypairs = x509keypair.X509KeyPairController()
certificates = certificate.CertificateController()
@expose.expose(V1)
def get(self):

@ -0,0 +1,155 @@
# Copyright 2015 NEC Corporation. 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.
import datetime
import pecan
from pecan import rest
import wsme
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
from magnum.api.controllers import base
from magnum.api.controllers import link
from magnum.api.controllers.v1 import types
from magnum.api.controllers.v1 import utils as api_utils
from magnum.common import exception
from magnum.common import policy
from magnum import objects
class Certificate(base.APIBase):
"""API representation of a certificate.
This class enforces type checking and value constraints, and converts
between the internal object model and the API representation of a
certificate.
"""
_bay_uuid = None
"""uuid or logical name of bay"""
_bay = None
def _get_bay_uuid(self):
return self._bay_uuid
def _set_bay_uuid(self, value):
if value and self._bay_uuid != value:
try:
self._bay = api_utils.get_rpc_resource('Bay', value)
self._bay_uuid = self._bay.uuid
except exception.BayNotFound as e:
# Change error code because 404 (NotFound) is inappropriate
# response for a POST request to create a Bay
e.code = 400 # BadRequest
raise e
elif value == wtypes.Unset:
self._bay_uuid = wtypes.Unset
bay_uuid = wsme.wsproperty(wtypes.text, _get_bay_uuid,
_set_bay_uuid, mandatory=True)
"""The bay UUID or id"""
links = wsme.wsattr([link.Link], readonly=True)
"""A list containing a self link and associated certificate links"""
csr = wtypes.StringType(min_length=1)
""""The Certificate Signing Request"""
pem = wtypes.StringType()
""""The Signed Certificate"""
def __init__(self, **kwargs):
super(Certificate, self).__init__()
self.fields = []
for field in objects.Certificate.fields:
# Skip fields we do not expose.
if not hasattr(self, field):
continue
self.fields.append(field)
setattr(self, field, kwargs.get(field, wtypes.Unset))
def get_bay(self):
if not self._bay:
self._bay = api_utils.get_rpc_resource('Bay', self.bay_uuid)
return self._bay
@staticmethod
def _convert_with_links(certificate, url, expand=True):
if not expand:
certificate.unset_fields_except(['bay_uuid', 'csr', 'pem'])
certificate.links = [link.Link.make_link('self', url,
'certificates',
certificate.bay_uuid),
link.Link.make_link('bookmark', url,
'certificates',
certificate.bay_uuid,
bookmark=True)]
return certificate
@classmethod
def convert_with_links(cls, rpc_cert, expand=True):
cert = Certificate(**rpc_cert.as_dict())
return cls._convert_with_links(cert,
pecan.request.host_url, expand)
@classmethod
def sample(cls, expand=True):
sample = cls(bay_uuid='7ae81bb3-dec3-4289-8d6c-da80bd8001ae',
created_at=datetime.datetime.utcnow(),
csr='AAA....AAA')
return cls._convert_with_links(sample, 'http://localhost:9511', expand)
class CertificateController(rest.RestController):
"""REST controller for Certificate."""
def __init__(self):
super(CertificateController, self).__init__()
_custom_actions = {
'detail': ['GET'],
}
@policy.enforce_wsgi("certificate", "get")
@wsme_pecan.wsexpose(Certificate, types.uuid_or_name)
def get_one(self, bay_ident):
"""Retrieve information about the given certificate.
:param bay_ident: UUID of a bay or
logical name of the bay.
"""
rpc_bay = api_utils.get_rpc_resource('Bay', bay_ident)
certificate = pecan.request.rpcapi.get_ca_certificate(rpc_bay)
return Certificate.convert_with_links(certificate)
@policy.enforce_wsgi("certificate", "create")
@wsme_pecan.wsexpose(Certificate, body=Certificate, status_code=201)
def post(self, certificate):
"""Create a new certificate.
:param certificate: a certificate within the request body.
"""
certificate_dict = certificate.as_dict()
context = pecan.request.context
certificate_dict['project_id'] = context.project_id
certificate_dict['user_id'] = context.user_id
cert_obj = objects.Certificate(context, **certificate_dict)
new_cert = pecan.request.rpcapi.sign_certificate(certificate.get_bay(),
cert_obj)
return Certificate.convert_with_links(new_cert)

@ -27,6 +27,7 @@ from magnum.common import rpc_service
from magnum.common import service as magnum_service
from magnum.common import short_id
from magnum.conductor.handlers import bay_conductor
from magnum.conductor.handlers import ca_conductor
from magnum.conductor.handlers import conductor_listener
from magnum.conductor.handlers import docker_conductor
from magnum.conductor.handlers import k8s_conductor
@ -56,6 +57,7 @@ def main():
bay_conductor.Handler(),
x509keypair_conductor.Handler(),
conductor_listener.Handler(),
ca_conductor.Handler(),
]
if (not os.path.isfile(cfg.CONF.bay.k8s_atomic_template_path)

@ -162,6 +162,13 @@ class API(rpc_service.API):
def x509keypair_list(self, context, limit, marker, sort_key, sort_dir):
return objects.X509KeyPair.list(context, limit, marker,
sort_key, sort_dir)
# CA operations
def sign_certificate(self, bay, certificate):
return self._call('sign_certificate', bay=bay, certificate=certificate)
def get_ca_certificate(self, bay):
return self._call('get_ca_certificate', bay=bay)
class ListenerAPI(rpc_service.API):

@ -0,0 +1,45 @@
# Copyright 2015 NEC Corporation. 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.
"""Magnum CA RPC handler."""
from oslo_log import log as logging
from magnum.conductor.handlers.common import cert_manager
from magnum import objects
LOG = logging.getLogger(__name__)
class Handler(object):
"""These are the backend operations. They are executed by the backend
service. API calls via AMQP (within the ReST API) trigger the
handlers to be called.
"""
def __init__(self):
super(Handler, self).__init__()
def sign_certificate(self, context, bay, certificate):
LOG.debug("Creating self signed x509 certificate")
signed_cert = cert_manager.sign_node_certificate(bay,
certificate.csr)
certificate.pem = signed_cert
return certificate
def get_ca_certificate(self, context, bay):
ca_cert = cert_manager.get_bay_ca_certificate(bay)
certificate = objects.Certificate.from_object_bay(bay)
certificate.pem = ca_cert
return certificate

@ -83,3 +83,15 @@ def generate_certificates_to_bay(bay):
bay.ca_cert_ref = ca_cert_ref
bay.magnum_cert_ref = magnum_cert_ref
def get_bay_ca_certificate(bay):
ca_cert = cert_manager.get_backend().CertManager.get_cert(bay.ca_cert_uuid)
return ca_cert.get_certificate()
def sign_node_certificate(bay, csr):
ca_cert = cert_manager.get_backend().CertManager.get_cert(bay.ca_cert_uuid)
node_cert = x509.sign(csr, bay.name, ca_cert.get_private_key(),
ca_cert.get_private_key_passphrase())
return node_cert

@ -15,6 +15,7 @@
from magnum.objects import bay
from magnum.objects import baylock
from magnum.objects import baymodel
from magnum.objects import certificate
from magnum.objects import container
from magnum.objects import node
from magnum.objects import pod
@ -32,7 +33,7 @@ Pod = pod.Pod
ReplicationController = rc.ReplicationController
Service = service.Service
X509KeyPair = x509keypair.X509KeyPair
Certificate = certificate.Certificate
__all__ = (Bay,
BayLock,
BayModel,
@ -41,4 +42,5 @@ __all__ = (Bay,
Pod,
ReplicationController,
Service,
X509KeyPair)
X509KeyPair,
Certificate)

@ -0,0 +1,45 @@
# coding=utf-8
#
#
# 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 oslo_versionedobjects import fields
from magnum.objects import base
@base.MagnumObjectRegistry.register
class Certificate(base.MagnumPersistentObject, base.MagnumObject,
base.MagnumObjectDictCompat):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'project_id': fields.StringField(nullable=True),
'user_id': fields.StringField(nullable=True),
'bay_uuid': fields.StringField(nullable=True),
'csr': fields.StringField(nullable=True),
'pem': fields.StringField(nullable=True),
}
@classmethod
def from_object_bay(cls, bay):
return cls(project_id=bay.project_id,
user_id=bay.user_id,
bay_uuid=bay.uuid)
@classmethod
def from_db_bay(cls, bay):
return cls(project_id=bay['project_id'],
user_id=bay['user_id'],
bay_uuid=bay['uuid'])

@ -73,6 +73,10 @@ class TestRootController(api_base.FunctionalTest):
u'x509keypairs': [{u'href': u'http://localhost/v1/x509keypairs/',
u'rel': u'self'},
{u'href': u'http://localhost/x509keypairs/',
u'rel': u'bookmark'}],
u'certificates': [{u'href': u'http://localhost/v1/certificates/',
u'rel': u'self'},
{u'href': u'http://localhost/certificates/',
u'rel': u'bookmark'}]}
response = self.app.get('/v1/')

@ -0,0 +1,185 @@
# 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 mock
from oslo_policy import policy
from magnum.api.controllers.v1 import certificate as api_cert
from magnum.common import utils
from magnum.tests import base
from magnum.tests.unit.api import base as api_base
from magnum.tests.unit.api import utils as apiutils
from magnum.tests.unit.objects import utils as obj_utils
class TestCertObject(base.TestCase):
@mock.patch('magnum.api.controllers.v1.utils.get_rpc_resource')
def test_cert_init(self, mock_get_rpc_resource):
cert_dict = apiutils.cert_post_data()
mock_bay = mock.MagicMock()
mock_bay.uuid = cert_dict['bay_uuid']
mock_get_rpc_resource.return_value = mock_bay
cert = api_cert.Certificate(**cert_dict)
self.assertEqual(cert.bay_uuid, cert_dict['bay_uuid'])
self.assertEqual(cert.csr, cert_dict['csr'])
self.assertEqual(cert.pem, cert_dict['pem'])
class TestGetCertificate(api_base.FunctionalTest):
def setUp(self):
super(TestGetCertificate, self).setUp()
self.bay = obj_utils.create_test_bay(self.context)
conductor_api_patcher = mock.patch('magnum.conductor.api.API')
self.conductor_api_class = conductor_api_patcher.start()
self.conductor_api = mock.MagicMock()
self.conductor_api_class.return_value = self.conductor_api
self.addCleanup(conductor_api_patcher.stop)
def test_get_one(self):
fake_cert = apiutils.cert_post_data()
mock_cert = mock.MagicMock()
mock_cert.as_dict.return_value = fake_cert
self.conductor_api.get_ca_certificate.return_value = mock_cert
response = self.get_json('/certificates/%s' % self.bay.uuid)
self.assertEqual(response['bay_uuid'], self.bay.uuid)
self.assertEqual(response['csr'], fake_cert['csr'])
self.assertEqual(response['pem'], fake_cert['pem'])
def test_get_one_by_name(self):
fake_cert = apiutils.cert_post_data()
mock_cert = mock.MagicMock()
mock_cert.as_dict.return_value = fake_cert
self.conductor_api.get_ca_certificate.return_value = mock_cert
response = self.get_json('/certificates/%s' % self.bay.name)
self.assertEqual(response['bay_uuid'], self.bay.uuid)
self.assertEqual(response['csr'], fake_cert['csr'])
self.assertEqual(response['pem'], fake_cert['pem'])
def test_get_one_by_name_not_found(self):
response = self.get_json('/certificates/not_found',
expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_get_one_by_name_multiple_bay(self):
obj_utils.create_test_bay(self.context, name='test_bay',
uuid=utils.generate_uuid())
obj_utils.create_test_bay(self.context, name='test_bay',
uuid=utils.generate_uuid())
response = self.get_json('/certificates/test_bay',
expect_errors=True)
self.assertEqual(409, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_links(self):
fake_cert = apiutils.cert_post_data()
mock_cert = mock.MagicMock()
mock_cert.as_dict.return_value = fake_cert
self.conductor_api.get_ca_certificate.return_value = mock_cert
response = self.get_json('/certificates/%s' % self.bay.uuid)
self.assertIn('links', response.keys())
self.assertEqual(2, len(response['links']))
self.assertIn(self.bay.uuid, response['links'][0]['href'])
for l in response['links']:
bookmark = l['rel'] == 'bookmark'
self.assertTrue(self.validate_link(l['href'], bookmark=bookmark))
class TestPost(api_base.FunctionalTest):
def setUp(self):
super(TestPost, self).setUp()
self.bay = obj_utils.create_test_bay(self.context)
conductor_api_patcher = mock.patch('magnum.conductor.api.API')
self.conductor_api_class = conductor_api_patcher.start()
self.conductor_api = mock.MagicMock()
self.conductor_api_class.return_value = self.conductor_api
self.addCleanup(conductor_api_patcher.stop)
self.conductor_api.sign_certificate.side_effect = self._fake_sign
@staticmethod
def _fake_sign(bay, cert):
cert.pem = 'fake-pem'
return cert
def test_create_cert(self, ):
new_cert = apiutils.cert_post_data(bay_uuid=self.bay.uuid)
del new_cert['pem']
response = self.post_json('/certificates', new_cert)
self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int)
self.assertEqual(response.json['bay_uuid'], new_cert['bay_uuid'])
self.assertEqual(response.json['pem'], 'fake-pem')
def test_create_cert_by_bay_name(self, ):
new_cert = apiutils.cert_post_data(bay_uuid=self.bay.name)
del new_cert['pem']
response = self.post_json('/certificates', new_cert)
self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int)
self.assertEqual(response.json['bay_uuid'], self.bay.uuid)
self.assertEqual(response.json['pem'], 'fake-pem')
def test_create_cert_bay_not_found(self, ):
new_cert = apiutils.cert_post_data(bay_uuid='not_found')
del new_cert['pem']
response = self.post_json('/certificates', new_cert,
expect_errors=True)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
class TestCertPolicyEnforcement(api_base.FunctionalTest):
def setUp(self):
super(TestCertPolicyEnforcement, self).setUp()
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: "project:non_fake"})
exc = self.assertRaises(policy.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith("disallowed by policy"))
def test_policy_disallow_get_one(self):
self._common_policy_check(
"certificate:get", self.get_json,
'/certificates/ce5da569-4f65-4272-9199-fac8c9fbc9d4')
def test_policy_disallow_create(self):
cert = apiutils.cert_post_data()
self._common_policy_check(
"certificate:create", self.post_json, '/certificates', cert)

@ -44,6 +44,14 @@ def bay_post_data(**kw):
return remove_internal(bay, internal)
def cert_post_data(**kw):
return {
'bay_uuid': kw.get('bay_uuid', '5d12f6fd-a196-4bf0-ae4c-1f639a523a52'),
'csr': kw.get('csr', 'fake-csr'),
'pem': kw.get('pem', 'fake-pem')
}
def pod_post_data(**kw):
pod = utils.get_test_pod(**kw)
if 'manifest' not in pod:

@ -14,12 +14,23 @@
import mock
from magnum.common.cert_manager import get_backend
from magnum.conductor.handlers.common import cert_manager
from magnum.tests import base
class CertManagerTestCase(base.BaseTestCase):
def setUp(self):
super(CertManagerTestCase, self).setUp()
cert_manager_patcher = mock.patch.object(cert_manager, 'cert_manager')
self.cert_manager = cert_manager_patcher.start()
self.addCleanup(cert_manager_patcher.stop)
self.cert_manager_backend = mock.MagicMock()
self.cert_manager.get_backend.return_value = self.cert_manager_backend
self.cert_manager_backend.CertManager = mock.MagicMock()
self.CertManager = self.cert_manager_backend.CertManager
@mock.patch('magnum.common.x509.operations.generate_ca_certificate')
@mock.patch('magnum.common.short_id.generate_id')
@ -33,17 +44,15 @@ class CertManagerTestCase(base.BaseTestCase):
mock_generate_id.return_value = expected_ca_password
mock_generate_ca_cert.return_value = expected_ca_cert
with mock.patch.object(get_backend().CertManager,
'store_cert') as mock_store_cert:
mock_store_cert.return_value = expected_ca_cert_ref
self.assertEqual(
cert_manager._generate_ca_cert(expected_ca_name),
(expected_ca_cert_ref, expected_ca_cert,
expected_ca_password))
self.CertManager.store_cert.return_value = expected_ca_cert_ref
self.assertEqual(
cert_manager._generate_ca_cert(expected_ca_name),
(expected_ca_cert_ref, expected_ca_cert,
expected_ca_password))
mock_generate_ca_cert.assert_called_once_with(
expected_ca_name, encryption_password=expected_ca_password)
mock_store_cert.assert_called_once_with(
self.CertManager.store_cert.assert_called_once_with(
certificate=expected_ca_cert['certificate'],
private_key=expected_ca_cert['private_key'],
private_key_passphrase=expected_ca_password,
@ -66,13 +75,12 @@ class CertManagerTestCase(base.BaseTestCase):
mock_generate_id.return_value = expected_password
mock_generate_cert.return_value = expected_cert
with mock.patch.object(get_backend().CertManager,
'store_cert') as mock_store_cert:
mock_store_cert.return_value = expected_cert_ref
self.assertEqual(
cert_manager._generate_client_cert(
expected_ca_name, expected_ca_cert, expected_ca_password),
expected_cert_ref)
self.CertManager.store_cert.return_value = expected_cert_ref
self.assertEqual(
cert_manager._generate_client_cert(
expected_ca_name, expected_ca_cert, expected_ca_password),
expected_cert_ref)
mock_generate_cert.assert_called_once_with(
expected_ca_name,
@ -81,7 +89,7 @@ class CertManagerTestCase(base.BaseTestCase):
encryption_password=expected_password,
ca_key_password=expected_ca_password,
)
mock_store_cert.assert_called_once_with(
self.CertManager.store_cert.assert_called_once_with(
certificate=expected_cert['certificate'],
private_key=expected_cert['private_key'],
private_key_passphrase=expected_password,
@ -115,3 +123,35 @@ class CertManagerTestCase(base.BaseTestCase):
mock_generate_ca_cert.assert_called_once_with(expected_ca_name)
mock_generate_client_cert.assert_called_once_with(
expected_ca_name, expected_ca_cert, expected_ca_password)
@mock.patch('magnum.common.x509.operations.sign')
def test_sign_node_certificate(self, mock_x509_sign):
mock_bay = mock.MagicMock()
mock_ca_cert = mock.MagicMock()
mock_ca_cert.get_private_key.return_value = mock.sentinel.priv_key
passphrase = mock.sentinel.passphrase
mock_ca_cert.get_private_key_passphrase.return_value = passphrase
self.CertManager.get_cert.return_value = mock_ca_cert
mock_csr = mock.MagicMock()
mock_x509_sign.return_value = mock.sentinel.signed_cert
bay_ca_cert = cert_manager.sign_node_certificate(mock_bay, mock_csr)
self.CertManager.get_cert.assert_called_once_with(
mock_bay.ca_cert_uuid)
mock_x509_sign.assert_called_once_with(mock_csr, mock_bay.name,
mock.sentinel.priv_key,
passphrase)
self.assertEqual(bay_ca_cert, mock.sentinel.signed_cert)
def test_get_bay_ca_certificate(self):
mock_bay = mock.MagicMock()
mock_ca_cert = mock.MagicMock()
mock_ca_cert.get_certificate.return_value = mock.sentinel.certificate
self.CertManager.get_cert.return_value = mock_ca_cert
bay_ca_cert = cert_manager.get_bay_ca_certificate(mock_bay)
self.CertManager.get_cert.assert_called_once_with(
mock_bay.ca_cert_uuid)
self.assertEqual(bay_ca_cert, mock.sentinel.certificate)

@ -0,0 +1,57 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from magnum.conductor.handlers import ca_conductor
from magnum.tests import base
import mock
from mock import patch
class TestSignConductor(base.TestCase):
def setUp(self):
super(TestSignConductor, self).setUp()
self.ca_handler = ca_conductor.Handler()
@patch.object(ca_conductor, 'cert_manager')
def test_sign_certificate(self, mock_cert_manager):
mock_bay = mock.MagicMock()
mock_certificate = mock.MagicMock()
mock_certificate.csr = 'fake-csr'
mock_cert_manager.sign_node_certificate.return_value = 'fake-pem'
actual_cert = self.ca_handler.sign_certificate(self.context,
mock_bay,
mock_certificate)
mock_cert_manager.sign_node_certificate.assert_called_once_with(
mock_bay, 'fake-csr'
)
self.assertEqual(actual_cert.pem, 'fake-pem')
@patch.object(ca_conductor, 'cert_manager')
def test_get_ca_certificate(self, mock_cert_manager):
mock_bay = mock.MagicMock()
mock_bay.uuid = 'bay-uuid'
mock_bay.user_id = 'user-id'
mock_bay.project_id = 'project-id'
mock_cert_manager.get_bay_ca_certificate.return_value = 'fake-pem'
actual_cert = self.ca_handler.get_ca_certificate(self.context,
mock_bay)
self.assertEqual(actual_cert.bay_uuid, mock_bay.uuid)
self.assertEqual(actual_cert.user_id, mock_bay.user_id)
self.assertEqual(actual_cert.project_id, mock_bay.project_id)
self.assertEqual(actual_cert.pem, 'fake-pem')

@ -18,6 +18,7 @@ import copy
import mock
from magnum.conductor import api as conductor_rpcapi
from magnum import objects
from magnum.tests.unit.db import base
from magnum.tests.unit.db import utils as dbutils
@ -33,6 +34,8 @@ class RPCAPITestCase(base.DbTestCase):
self.fake_service = dbutils.get_test_service(driver='fake-driver')
self.fake_x509keypair = dbutils.get_test_x509keypair(
driver='fake-driver')
self.fake_certificate = objects.Certificate.from_db_bay(self.fake_bay)
self.fake_certificate.csr = 'fake-csr'
def _test_rpcapi(self, method, rpc_method, **kwargs):
rpcapi_cls = kwargs.pop('rpcapi_cls', conductor_rpcapi.API)
@ -250,3 +253,16 @@ class RPCAPITestCase(base.DbTestCase):
'call',
version='1.1',
uuid=self.fake_x509keypair['name'])
def test_sign_certificate(self):
self._test_rpcapi('sign_certificate',
'call',
version='1.0',
bay=self.fake_bay,
certificate=self.fake_certificate)
def test_get_ca_certificate(self):
self._test_rpcapi('get_ca_certificate',
'call',
version='1.0',
bay=self.fake_bay)

Loading…
Cancel
Save