diff --git a/vmware_nsxlib/tests/unit/v3/test_cert.py b/vmware_nsxlib/tests/unit/v3/test_cert.py index 742c3946..8fc3fc6c 100644 --- a/vmware_nsxlib/tests/unit/v3/test_cert.py +++ b/vmware_nsxlib/tests/unit/v3/test_cert.py @@ -13,6 +13,7 @@ # License for the specific language governing permissions and limitations # under the License. # +import os from neutron_lib import exceptions from OpenSSL import crypto @@ -24,7 +25,7 @@ from vmware_nsxlib.tests.unit.v3 import test_client from vmware_nsxlib.v3 import client from vmware_nsxlib.v3 import client_cert from vmware_nsxlib.v3 import exceptions as nsxlib_exc -from vmware_nsxlib.v3 import trust_management +from vmware_nsxlib.v3 import trust_management as tm class DummyStorageDriver(dict): @@ -53,6 +54,7 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase): identity = 'drumknott' cert_id = "00000000-1111-2222-3333-444444444444" identity_id = "55555555-6666-7777-8888-999999999999" + node_id = "meh" def _get_mocked_response(self, status_code, results): return mocks.MockRequestsResponse( @@ -67,17 +69,31 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase): 'module_name': 'never mind', 'error message': 'bad luck'})) - def _get_mocked_trust(self, action): + def _get_mocked_trust(self, action, cert_pem): fake_responses = [] - if action == 'create': + if 'create' in action: # import cert and return its id results = [{'id': self.cert_id}] fake_responses.append(self._get_mocked_response(201, results)) # and then bind this id to principal identity fake_responses.append(self._get_mocked_response(201, [])) - elif action == 'delete': + if 'delete' in action: + nsx_style_pem = tm.NsxLibTrustManagement.remove_newlines_from_pem( + cert_pem) + # get certs list, including same cert imported twice edge case + results = [{'resource_type': 'Certificate', + 'id': 'dont care', + 'pem_encoded': 'some junk'}, + {'resource_type': 'Certificate', + 'id': 'some_other_cert_id', + 'pem_encoded': nsx_style_pem}, + {'resource_type': 'Certificate', + 'id': self.cert_id, + 'pem_encoded': nsx_style_pem}] + fake_responses.append(self._get_mocked_response(200, results)) + # get principal identities list results = [{'resource_type': 'Principal Identity', 'id': 'dont care', @@ -97,32 +113,10 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase): client.JSONRESTClient, url_prefix='api/v1', session_response=fake_responses) - return trust_management.NsxLibTrustManagement(mock_client, {}) - - def test_generate_cert(self): - """Test startup without certificate + certificate generation""" - - storage_driver = DummyStorageDriver() - # Prepare fake trust management for "cert create" requests - mocked_trust = self._get_mocked_trust('create') - cert = client_cert.ClientCertificateManager(self.identity, - mocked_trust, - storage_driver) - self.assertFalse(cert.exists()) - - cert.generate(subject={}, key_size=2048, valid_for_days=333, - node_id='meh') - - # verify client cert was generated and makes sense - self.assertTrue(cert.exists()) - self.assertEqual(332, cert.expires_in_days()) - cert_pem, key_pem = cert.get_pem() - - # verify cert ans PK were stored in storage - stored_cert, stored_key = storage_driver.get_cert(self.identity) - self.assertEqual(cert_pem, stored_cert) - self.assertEqual(key_pem, stored_key) + return tm.NsxLibTrustManagement(mock_client, {}) + def _verify_backend_create(self, mocked_trust, cert_pem): + """Verify API calls to create cert and identity on backend""" # verify API call to import cert on backend cert_pem = mocked_trust.remove_newlines_from_pem(cert_pem) base_uri = 'https://1.2.3.4/api/v1/trust-management' @@ -135,7 +129,7 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase): # verify API call to bind cert to identity on backend uri = base_uri + '/principal-identities' expected_body = {'name': self.identity, - 'node_id': 'meh', + 'node_id': self.node_id, 'permission_group': 'read_write_api_users', 'certificate_id': self.cert_id, 'is_protected': True} @@ -144,6 +138,51 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase): data=jsonutils.dumps(expected_body, sort_keys=True)) + def _verify_backend_delete(self, mocked_trust): + """Verify API calls to fetch and delete cert and identity""" + # verify API call to query identities in order to get cert id + base_uri = 'https://1.2.3.4/api/v1/trust-management' + uri = base_uri + '/principal-identities' + test_client.assert_json_call('get', mocked_trust.client, uri, + single_call=False) + # verify API call to delete openstack principal identity + uri = uri + '/' + self.identity_id + test_client.assert_json_call('delete', mocked_trust.client, uri, + single_call=False) + + # verify API call to delete certificate + uri = base_uri + '/certificates/' + self.cert_id + test_client.assert_json_call('delete', mocked_trust.client, uri, + single_call=False) + + def test_generate_cert(self): + """Test startup without certificate + certificate generation""" + + storage_driver = DummyStorageDriver() + # Prepare fake trust management for "cert create" requests + cert_pem, key_pem = storage_driver.get_cert(self.identity) + mocked_trust = self._get_mocked_trust('create', cert_pem) + cert = client_cert.ClientCertificateManager(self.identity, + mocked_trust, + storage_driver) + self.assertFalse(cert.exists()) + + cert.generate(subject={}, key_size=2048, valid_for_days=333, + node_id=self.node_id) + + # verify client cert was generated and makes sense + self.assertTrue(cert.exists()) + self.assertEqual(332, cert.expires_in_days()) + cert_pem, key_pem = cert.get_pem() + + # verify cert ans PK were stored in storage + stored_cert, stored_key = storage_driver.get_cert(self.identity) + self.assertEqual(cert_pem, stored_cert) + self.assertEqual(key_pem, stored_key) + + # verify backend API calls + self._verify_backend_create(mocked_trust, cert_pem) + # try to generate cert again and fail self.assertRaises(nsxlib_exc.ObjectAlreadyExists, cert.generate, {}) @@ -170,7 +209,8 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase): # get mocked backend driver for trust management, # prepared for get request, that preceeds delete operation - mocked_trust = self._get_mocked_trust('delete') + cert_pem, key_pem = storage_driver.get_cert(self.identity) + mocked_trust = self._get_mocked_trust('delete', cert_pem) cert = client_cert.ClientCertificateManager(self.identity, mocked_trust, @@ -182,21 +222,42 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase): self.assertFalse(cert.exists()) self.assertTrue(storage_driver.is_empty(self.identity)) - # verify API call to query identities in order to get cert id - base_uri = 'https://1.2.3.4/api/v1/trust-management' - uri = base_uri + '/principal-identities' - test_client.assert_json_call('get', mocked_trust.client, uri, - single_call=False) + self._verify_backend_delete(mocked_trust) - # verify API call to delete openstack principal identity - uri = uri + '/' + self.identity_id - test_client.assert_json_call('delete', mocked_trust.client, uri, - single_call=False) + def _test_import_and_delete_cert(self, with_pkey=True): + filename = '/tmp/test.pem' + # this driver simulates storage==none scenario + noop_driver = DummyStorageDriver() + cert, key = client_cert.generate_self_signed_cert_pair(4096, + 20, + 'sha256', + {}) - # verify API call to delete certificate - uri = base_uri + '/certificates/' + self.cert_id - test_client.assert_json_call('delete', mocked_trust.client, uri, - single_call=False) + cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) + key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, key) + with open(filename, 'wb') as f: + f.write(cert_pem) + if with_pkey: + f.write(key_pem) + + mocked_trust = self._get_mocked_trust('create_delete', + cert_pem) + cert = client_cert.ClientCertificateManager(self.identity, + mocked_trust, + noop_driver) + cert.import_pem(filename, self.node_id) + self._verify_backend_create(mocked_trust, cert_pem) + + cert.delete_pem(filename) + self._verify_backend_delete(mocked_trust) + + os.remove(filename) + + def test_import_and_delete_cert_pkey(self): + self._test_import_and_delete_cert(True) + + def test_import_and_delete_cert_only(self): + self._test_import_and_delete_cert(False) def test_get_certificate_details(self): """Test retrieving cert details for existing cert""" diff --git a/vmware_nsxlib/v3/client_cert.py b/vmware_nsxlib/v3/client_cert.py index 683ca520..15060f2f 100644 --- a/vmware_nsxlib/v3/client_cert.py +++ b/vmware_nsxlib/v3/client_cert.py @@ -26,7 +26,6 @@ from vmware_nsxlib.v3 import exceptions as nsxlib_exceptions LOG = log.getLogger(__name__) -NSX_ERROR_IDENTITY_EXISTS = 2027 CERT_SUBJECT_COUNTRY = 'country' CERT_SUBJECT_STATE = 'state' CERT_SUBJECT_ORG = 'organization' @@ -169,32 +168,22 @@ class ClientCertificateManager(object): def delete(self): """Delete existing certificate from storage and backend""" - if not self.exists(): + cert_pem, key_pem = self.get_pem() + if not cert_pem: return ok = True try: - # delete certificate and principal identity from backend - details = self._nsx_trust_management.get_identity_details( - self._identity) - - # TODO(annak): do not delete the identity once - # NSX supports multiple certificates per identity - # this will be required to support multiple openstack - # installations using same backend NSX - self._nsx_trust_management.delete_identity(details['id']) - if details['certificate_id']: - self._nsx_trust_management.delete_cert( - details['certificate_id']) - + self._nsx_trust_management.delete_cert_and_identity( + self._identity, cert_pem) except nsxlib_exceptions.ManagerError as e: LOG.error(_LE("Failed to clear certificate on backend: %s"), e) ok = False try: self._storage_driver.delete_cert(self._identity) - except Exception as e: - LOG.error(_LE("Failed to clear certificate on storage: %s"), e) + except Exception: + LOG.error(_LE("Failed to clear certificate in storage: %s"), e) ok = False self._cert = None @@ -211,12 +200,7 @@ class ClientCertificateManager(object): cert_pem, key_pem = self._storage_driver.get_cert(self._identity) return cert_pem is not None - def import_pem(self, filename, node_id=None): - """Import and register existing certificate in PEM format""" - - # TODO(annak): support PK import as well - self._validate_empty() - + def _get_cert_from_file(self, filename): with open(filename, 'r') as f: cert_pem = f.read() @@ -231,13 +215,35 @@ class ClientCertificateManager(object): raise nsxlib_exceptions.CertificateError( msg=_("Failed to import client certificate")) + return cert + + def import_pem(self, filename, node_id=None): + """Import and register existing certificate in PEM format""" + + # TODO(annak): support PK import as well + self._validate_empty() + + cert = self._get_cert_from_file(filename) # register on backend self._register_cert(cert, node_id or uuid.uuid4()) + cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) self._storage_driver.store_cert(self._identity, cert_pem, None) LOG.debug("Client certificate imported successfully") + def delete_pem(self, filename): + """Delete specified client certificate without storage verification""" + # This file may contain private key + # passing the pem through crypto will perform validation and + # stripp off the key + cert = self._get_cert_from_file(filename) + cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) + + self._nsx_trust_management.delete_cert_and_identity(self._identity, + cert_pem) + self._storage_driver.delete_cert(self._identity) + def _load_from_storage(self): """Returns certificate and key pair in PEM format""" @@ -328,11 +334,10 @@ class ClientCertificateManager(object): def _register_cert(self, cert, node_id): cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) - nsx_cert_id = self._nsx_trust_management.create_cert(cert_pem) - self._nsx_trust_management.create_identity(self._identity, - nsx_cert_id, - node_id, - 'read_write_api_users') + + self._nsx_trust_management.create_cert_and_identity(self._identity, + cert_pem, + node_id) class ClientCertProvider(object): diff --git a/vmware_nsxlib/v3/cluster.py b/vmware_nsxlib/v3/cluster.py index e59871e2..ea4e13b3 100644 --- a/vmware_nsxlib/v3/cluster.py +++ b/vmware_nsxlib/v3/cluster.py @@ -246,6 +246,12 @@ class Endpoint(object): self._state = EndpointState.INITIALIZED self._last_updated = datetime.datetime.now() + def regenerate_pool(self): + self.pool = pools.Pool(min_size=self.pool.min_size, + max_size=self.pool.max_size, + order_as_stack=True, + create=self.pool.create) + @property def last_updated(self): return self._last_updated @@ -404,6 +410,12 @@ class ClusteredAPI(object): with endpoint.pool.item() as conn: self._http_provider.validate_connection(self, endpoint, conn) endpoint.set_state(EndpointState.UP) + except exceptions.ClientCertificateNotTrusted: + LOG.warning(_LW("Failed to validate API cluster endpoint " + "'%(ep)s' due to untrusted client certificate"), + {'ep': endpoint}) + # regenerate connection pool based on new certificate + endpoint.regenerate_pool() except Exception as e: endpoint.set_state(EndpointState.DOWN) LOG.warning(_LW("Failed to validate API cluster endpoint " diff --git a/vmware_nsxlib/v3/trust_management.py b/vmware_nsxlib/v3/trust_management.py index e82a85e4..fa89430a 100644 --- a/vmware_nsxlib/v3/trust_management.py +++ b/vmware_nsxlib/v3/trust_management.py @@ -26,12 +26,15 @@ USER_GROUP_TYPES = [ class NsxLibTrustManagement(utils.NsxLibApiBase): - def remove_newlines_from_pem(self, pem): + @staticmethod + def remove_newlines_from_pem(pem): """NSX expects pem without newlines in certificate body BEGIN and END sections should be separated with newlines """ lines = pem.split(b'\n') + if len(lines) <= 1: + return pem result = lines[0] + b'\n' result += b''.join(lines[1:-2]) result += b'\n' + lines[-2] @@ -50,11 +53,14 @@ class NsxLibTrustManagement(utils.NsxLibApiBase): resource = CERT_SECTION + '/' + cert_id return self.client.get(resource) + def get_certs(self): + return self.client.get(CERT_SECTION)['results'] + def delete_cert(self, cert_id): resource = CERT_SECTION + '/' + cert_id self.client.delete(resource) - def create_identity(self, identity, cert_id, + def create_identity(self, name, cert_id, node_id, permission_group): # Validate permission group before sending to server if permission_group not in USER_GROUP_TYPES: @@ -62,15 +68,20 @@ class NsxLibTrustManagement(utils.NsxLibApiBase): operation='create_identity', arg_val=permission_group, arg_name='permission_group') - body = {'name': identity, 'certificate_id': cert_id, + body = {'name': name, 'certificate_id': cert_id, 'node_id': node_id, 'permission_group': permission_group, 'is_protected': True} self.client.create(ID_SECTION, body) - def delete_identity(self, identity): - resource = ID_SECTION + '/' + identity + def get_identities(self, name): + ids = self.client.get(ID_SECTION)['results'] + return [identity for identity in ids if identity['name'] == name] + + def delete_identity(self, identity_id): + resource = ID_SECTION + '/' + identity_id self.client.delete(resource) + # TODO(annak): kept for sake of short-term stability, remove this def get_identity_details(self, identity): results = self.client.get(ID_SECTION)['results'] for result in results: @@ -80,3 +91,41 @@ class NsxLibTrustManagement(utils.NsxLibApiBase): raise nsxlib_exc.ResourceNotFound( manager=self.client.nsx_api_managers, operation="Principal identity %s not found" % identity) + + def find_cert_and_identity(self, name, cert_pem): + nsx_style_pem = self.remove_newlines_from_pem(cert_pem) + certs = self.get_certs() + + cert_ids = [cert['id'] for cert in certs + if cert['pem_encoded'] == nsx_style_pem.decode('ascii')] + if not cert_ids: + raise nsxlib_exc.ResourceNotFound( + manager=self.client.nsx_api_managers, + operation="delete_certificate") + + identities = self.get_identities(name) + # should be zero or one matching identities + results = [identity for identity in identities + if identity['certificate_id'] in cert_ids] + + if not results: + raise nsxlib_exc.ResourceNotFound( + manager=self.client.nsx_api_managers, + operation="delete_identity") + + return results[0]['certificate_id'], results[0]['id'] + + def delete_cert_and_identity(self, name, cert_pem): + cert_id, identity_id = self.find_cert_and_identity(name, cert_pem) + self.delete_identity(identity_id) + self.delete_cert(cert_id) + + def create_cert_and_identity(self, name, cert_pem, + node_id, + permission_group='read_write_api_users'): + nsx_cert_id = self.create_cert(cert_pem) + try: + self.create_identity(name, nsx_cert_id, node_id, permission_group) + except nsxlib_exc.ManagerError as e: + self.delete_cert(nsx_cert_id) + raise e