Support multiple client certificate per identity
When openstack runs in HA mode, admin might choose to assign two separate client certificates for each openstack host. This is possible with storage_type=none. This change allows deleting cert and identity based not only on identity name, but on cert pem. In addition, allow faster cluster recovery in case of certificate change. Change-Id: Ia4eea874cfa2bf4befc724b719e53e936292e11f
This commit is contained in:
parent
627964757b
commit
1ac9c11b03
|
@ -13,6 +13,7 @@
|
||||||
# License for the specific language governing permissions and limitations
|
# License for the specific language governing permissions and limitations
|
||||||
# under the License.
|
# under the License.
|
||||||
#
|
#
|
||||||
|
import os
|
||||||
|
|
||||||
from neutron_lib import exceptions
|
from neutron_lib import exceptions
|
||||||
from OpenSSL import crypto
|
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
|
||||||
from vmware_nsxlib.v3 import client_cert
|
from vmware_nsxlib.v3 import client_cert
|
||||||
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
|
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):
|
class DummyStorageDriver(dict):
|
||||||
|
@ -53,6 +54,7 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
|
||||||
identity = 'drumknott'
|
identity = 'drumknott'
|
||||||
cert_id = "00000000-1111-2222-3333-444444444444"
|
cert_id = "00000000-1111-2222-3333-444444444444"
|
||||||
identity_id = "55555555-6666-7777-8888-999999999999"
|
identity_id = "55555555-6666-7777-8888-999999999999"
|
||||||
|
node_id = "meh"
|
||||||
|
|
||||||
def _get_mocked_response(self, status_code, results):
|
def _get_mocked_response(self, status_code, results):
|
||||||
return mocks.MockRequestsResponse(
|
return mocks.MockRequestsResponse(
|
||||||
|
@ -67,17 +69,31 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
|
||||||
'module_name': 'never mind',
|
'module_name': 'never mind',
|
||||||
'error message': 'bad luck'}))
|
'error message': 'bad luck'}))
|
||||||
|
|
||||||
def _get_mocked_trust(self, action):
|
def _get_mocked_trust(self, action, cert_pem):
|
||||||
|
|
||||||
fake_responses = []
|
fake_responses = []
|
||||||
if action == 'create':
|
if 'create' in action:
|
||||||
# import cert and return its id
|
# import cert and return its id
|
||||||
results = [{'id': self.cert_id}]
|
results = [{'id': self.cert_id}]
|
||||||
fake_responses.append(self._get_mocked_response(201, results))
|
fake_responses.append(self._get_mocked_response(201, results))
|
||||||
# and then bind this id to principal identity
|
# and then bind this id to principal identity
|
||||||
fake_responses.append(self._get_mocked_response(201, []))
|
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
|
# get principal identities list
|
||||||
results = [{'resource_type': 'Principal Identity',
|
results = [{'resource_type': 'Principal Identity',
|
||||||
'id': 'dont care',
|
'id': 'dont care',
|
||||||
|
@ -97,32 +113,10 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
|
||||||
client.JSONRESTClient,
|
client.JSONRESTClient,
|
||||||
url_prefix='api/v1', session_response=fake_responses)
|
url_prefix='api/v1', session_response=fake_responses)
|
||||||
|
|
||||||
return trust_management.NsxLibTrustManagement(mock_client, {})
|
return tm.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)
|
|
||||||
|
|
||||||
|
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
|
# verify API call to import cert on backend
|
||||||
cert_pem = mocked_trust.remove_newlines_from_pem(cert_pem)
|
cert_pem = mocked_trust.remove_newlines_from_pem(cert_pem)
|
||||||
base_uri = 'https://1.2.3.4/api/v1/trust-management'
|
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
|
# verify API call to bind cert to identity on backend
|
||||||
uri = base_uri + '/principal-identities'
|
uri = base_uri + '/principal-identities'
|
||||||
expected_body = {'name': self.identity,
|
expected_body = {'name': self.identity,
|
||||||
'node_id': 'meh',
|
'node_id': self.node_id,
|
||||||
'permission_group': 'read_write_api_users',
|
'permission_group': 'read_write_api_users',
|
||||||
'certificate_id': self.cert_id,
|
'certificate_id': self.cert_id,
|
||||||
'is_protected': True}
|
'is_protected': True}
|
||||||
|
@ -144,6 +138,51 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
|
||||||
data=jsonutils.dumps(expected_body,
|
data=jsonutils.dumps(expected_body,
|
||||||
sort_keys=True))
|
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
|
# try to generate cert again and fail
|
||||||
self.assertRaises(nsxlib_exc.ObjectAlreadyExists,
|
self.assertRaises(nsxlib_exc.ObjectAlreadyExists,
|
||||||
cert.generate, {})
|
cert.generate, {})
|
||||||
|
@ -170,7 +209,8 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
|
||||||
|
|
||||||
# get mocked backend driver for trust management,
|
# get mocked backend driver for trust management,
|
||||||
# prepared for get request, that preceeds delete operation
|
# 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,
|
cert = client_cert.ClientCertificateManager(self.identity,
|
||||||
mocked_trust,
|
mocked_trust,
|
||||||
|
@ -182,21 +222,42 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
|
||||||
self.assertFalse(cert.exists())
|
self.assertFalse(cert.exists())
|
||||||
self.assertTrue(storage_driver.is_empty(self.identity))
|
self.assertTrue(storage_driver.is_empty(self.identity))
|
||||||
|
|
||||||
# verify API call to query identities in order to get cert id
|
self._verify_backend_delete(mocked_trust)
|
||||||
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
|
def _test_import_and_delete_cert(self, with_pkey=True):
|
||||||
uri = uri + '/' + self.identity_id
|
filename = '/tmp/test.pem'
|
||||||
test_client.assert_json_call('delete', mocked_trust.client, uri,
|
# this driver simulates storage==none scenario
|
||||||
single_call=False)
|
noop_driver = DummyStorageDriver()
|
||||||
|
cert, key = client_cert.generate_self_signed_cert_pair(4096,
|
||||||
|
20,
|
||||||
|
'sha256',
|
||||||
|
{})
|
||||||
|
|
||||||
# verify API call to delete certificate
|
cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
|
||||||
uri = base_uri + '/certificates/' + self.cert_id
|
key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, key)
|
||||||
test_client.assert_json_call('delete', mocked_trust.client, uri,
|
with open(filename, 'wb') as f:
|
||||||
single_call=False)
|
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):
|
def test_get_certificate_details(self):
|
||||||
"""Test retrieving cert details for existing cert"""
|
"""Test retrieving cert details for existing cert"""
|
||||||
|
|
|
@ -26,7 +26,6 @@ from vmware_nsxlib.v3 import exceptions as nsxlib_exceptions
|
||||||
|
|
||||||
LOG = log.getLogger(__name__)
|
LOG = log.getLogger(__name__)
|
||||||
|
|
||||||
NSX_ERROR_IDENTITY_EXISTS = 2027
|
|
||||||
CERT_SUBJECT_COUNTRY = 'country'
|
CERT_SUBJECT_COUNTRY = 'country'
|
||||||
CERT_SUBJECT_STATE = 'state'
|
CERT_SUBJECT_STATE = 'state'
|
||||||
CERT_SUBJECT_ORG = 'organization'
|
CERT_SUBJECT_ORG = 'organization'
|
||||||
|
@ -169,32 +168,22 @@ class ClientCertificateManager(object):
|
||||||
|
|
||||||
def delete(self):
|
def delete(self):
|
||||||
"""Delete existing certificate from storage and backend"""
|
"""Delete existing certificate from storage and backend"""
|
||||||
if not self.exists():
|
cert_pem, key_pem = self.get_pem()
|
||||||
|
if not cert_pem:
|
||||||
return
|
return
|
||||||
|
|
||||||
ok = True
|
ok = True
|
||||||
try:
|
try:
|
||||||
# delete certificate and principal identity from backend
|
self._nsx_trust_management.delete_cert_and_identity(
|
||||||
details = self._nsx_trust_management.get_identity_details(
|
self._identity, cert_pem)
|
||||||
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'])
|
|
||||||
|
|
||||||
except nsxlib_exceptions.ManagerError as e:
|
except nsxlib_exceptions.ManagerError as e:
|
||||||
LOG.error(_LE("Failed to clear certificate on backend: %s"), e)
|
LOG.error(_LE("Failed to clear certificate on backend: %s"), e)
|
||||||
ok = False
|
ok = False
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self._storage_driver.delete_cert(self._identity)
|
self._storage_driver.delete_cert(self._identity)
|
||||||
except Exception as e:
|
except Exception:
|
||||||
LOG.error(_LE("Failed to clear certificate on storage: %s"), e)
|
LOG.error(_LE("Failed to clear certificate in storage: %s"), e)
|
||||||
ok = False
|
ok = False
|
||||||
|
|
||||||
self._cert = None
|
self._cert = None
|
||||||
|
@ -211,12 +200,7 @@ class ClientCertificateManager(object):
|
||||||
cert_pem, key_pem = self._storage_driver.get_cert(self._identity)
|
cert_pem, key_pem = self._storage_driver.get_cert(self._identity)
|
||||||
return cert_pem is not None
|
return cert_pem is not None
|
||||||
|
|
||||||
def import_pem(self, filename, node_id=None):
|
def _get_cert_from_file(self, filename):
|
||||||
"""Import and register existing certificate in PEM format"""
|
|
||||||
|
|
||||||
# TODO(annak): support PK import as well
|
|
||||||
self._validate_empty()
|
|
||||||
|
|
||||||
with open(filename, 'r') as f:
|
with open(filename, 'r') as f:
|
||||||
cert_pem = f.read()
|
cert_pem = f.read()
|
||||||
|
|
||||||
|
@ -231,13 +215,35 @@ class ClientCertificateManager(object):
|
||||||
raise nsxlib_exceptions.CertificateError(
|
raise nsxlib_exceptions.CertificateError(
|
||||||
msg=_("Failed to import client certificate"))
|
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
|
# register on backend
|
||||||
self._register_cert(cert, node_id or uuid.uuid4())
|
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)
|
self._storage_driver.store_cert(self._identity, cert_pem, None)
|
||||||
|
|
||||||
LOG.debug("Client certificate imported successfully")
|
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):
|
def _load_from_storage(self):
|
||||||
"""Returns certificate and key pair in PEM format"""
|
"""Returns certificate and key pair in PEM format"""
|
||||||
|
|
||||||
|
@ -328,11 +334,10 @@ class ClientCertificateManager(object):
|
||||||
|
|
||||||
def _register_cert(self, cert, node_id):
|
def _register_cert(self, cert, node_id):
|
||||||
cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
|
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,
|
self._nsx_trust_management.create_cert_and_identity(self._identity,
|
||||||
nsx_cert_id,
|
cert_pem,
|
||||||
node_id,
|
node_id)
|
||||||
'read_write_api_users')
|
|
||||||
|
|
||||||
|
|
||||||
class ClientCertProvider(object):
|
class ClientCertProvider(object):
|
||||||
|
|
|
@ -246,6 +246,12 @@ class Endpoint(object):
|
||||||
self._state = EndpointState.INITIALIZED
|
self._state = EndpointState.INITIALIZED
|
||||||
self._last_updated = datetime.datetime.now()
|
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
|
@property
|
||||||
def last_updated(self):
|
def last_updated(self):
|
||||||
return self._last_updated
|
return self._last_updated
|
||||||
|
@ -404,6 +410,12 @@ class ClusteredAPI(object):
|
||||||
with endpoint.pool.item() as conn:
|
with endpoint.pool.item() as conn:
|
||||||
self._http_provider.validate_connection(self, endpoint, conn)
|
self._http_provider.validate_connection(self, endpoint, conn)
|
||||||
endpoint.set_state(EndpointState.UP)
|
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:
|
except Exception as e:
|
||||||
endpoint.set_state(EndpointState.DOWN)
|
endpoint.set_state(EndpointState.DOWN)
|
||||||
LOG.warning(_LW("Failed to validate API cluster endpoint "
|
LOG.warning(_LW("Failed to validate API cluster endpoint "
|
||||||
|
|
|
@ -26,12 +26,15 @@ USER_GROUP_TYPES = [
|
||||||
|
|
||||||
class NsxLibTrustManagement(utils.NsxLibApiBase):
|
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
|
"""NSX expects pem without newlines in certificate body
|
||||||
|
|
||||||
BEGIN and END sections should be separated with newlines
|
BEGIN and END sections should be separated with newlines
|
||||||
"""
|
"""
|
||||||
lines = pem.split(b'\n')
|
lines = pem.split(b'\n')
|
||||||
|
if len(lines) <= 1:
|
||||||
|
return pem
|
||||||
result = lines[0] + b'\n'
|
result = lines[0] + b'\n'
|
||||||
result += b''.join(lines[1:-2])
|
result += b''.join(lines[1:-2])
|
||||||
result += b'\n' + lines[-2]
|
result += b'\n' + lines[-2]
|
||||||
|
@ -50,11 +53,14 @@ class NsxLibTrustManagement(utils.NsxLibApiBase):
|
||||||
resource = CERT_SECTION + '/' + cert_id
|
resource = CERT_SECTION + '/' + cert_id
|
||||||
return self.client.get(resource)
|
return self.client.get(resource)
|
||||||
|
|
||||||
|
def get_certs(self):
|
||||||
|
return self.client.get(CERT_SECTION)['results']
|
||||||
|
|
||||||
def delete_cert(self, cert_id):
|
def delete_cert(self, cert_id):
|
||||||
resource = CERT_SECTION + '/' + cert_id
|
resource = CERT_SECTION + '/' + cert_id
|
||||||
self.client.delete(resource)
|
self.client.delete(resource)
|
||||||
|
|
||||||
def create_identity(self, identity, cert_id,
|
def create_identity(self, name, cert_id,
|
||||||
node_id, permission_group):
|
node_id, permission_group):
|
||||||
# Validate permission group before sending to server
|
# Validate permission group before sending to server
|
||||||
if permission_group not in USER_GROUP_TYPES:
|
if permission_group not in USER_GROUP_TYPES:
|
||||||
|
@ -62,15 +68,20 @@ class NsxLibTrustManagement(utils.NsxLibApiBase):
|
||||||
operation='create_identity',
|
operation='create_identity',
|
||||||
arg_val=permission_group,
|
arg_val=permission_group,
|
||||||
arg_name='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,
|
'node_id': node_id, 'permission_group': permission_group,
|
||||||
'is_protected': True}
|
'is_protected': True}
|
||||||
self.client.create(ID_SECTION, body)
|
self.client.create(ID_SECTION, body)
|
||||||
|
|
||||||
def delete_identity(self, identity):
|
def get_identities(self, name):
|
||||||
resource = ID_SECTION + '/' + identity
|
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)
|
self.client.delete(resource)
|
||||||
|
|
||||||
|
# TODO(annak): kept for sake of short-term stability, remove this
|
||||||
def get_identity_details(self, identity):
|
def get_identity_details(self, identity):
|
||||||
results = self.client.get(ID_SECTION)['results']
|
results = self.client.get(ID_SECTION)['results']
|
||||||
for result in results:
|
for result in results:
|
||||||
|
@ -80,3 +91,41 @@ class NsxLibTrustManagement(utils.NsxLibApiBase):
|
||||||
raise nsxlib_exc.ResourceNotFound(
|
raise nsxlib_exc.ResourceNotFound(
|
||||||
manager=self.client.nsx_api_managers,
|
manager=self.client.nsx_api_managers,
|
||||||
operation="Principal identity %s not found" % identity)
|
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
|
||||||
|
|
Loading…
Reference in New Issue