Merge "Pass node ID and user permissions when creating NSX identity"

vmware_nsxlib/tests/unit/v3/test_cert.py

@@ -77,22 +77,6 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
             # and then bind this id to principal identity
             fake_responses.append(self._get_mocked_response(201, []))
 
-        elif action == 'retry-create':
-            # simulate "identity already exists" failure
-            results = [{'id': self.cert_id}]
-            fake_responses.append(self._get_mocked_response(201, results))
-            fake_responses.append(self._get_mocked_error_response(400, 2027))
-            # after error generate code will retry identity deletion:
-            # first get indentities
-            results = [{'resource_type': 'Principal Identity',
-                        'id': self.identity_id,
-                        'name': self.identity,
-                        'certificate_id': self.cert_id}]
-            # then delete identity
-            fake_responses.append(self._get_mocked_response(200, results))
-            # then retry identity create
-            fake_responses.append(self._get_mocked_response(204, []))
-
         elif action == 'delete':
             # get principal identities list
             results = [{'resource_type': 'Principal Identity',
@@ -126,7 +110,8 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
                                                     storage_driver)
         self.assertFalse(cert.exists())
 
-        cert.generate(subject={}, key_size=2048, valid_for_days=333)
+        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())
@@ -150,7 +135,10 @@ 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,
-                         'certificate_id': self.cert_id}
+                         'node_id': 'meh',
+                         'permission_group': 'read_write_api_users',
+                         'certificate_id': self.cert_id,
+                         'is_protected': True}
         test_client.assert_json_call('post', mocked_trust.client, uri,
                                      single_call=False,
                                      data=jsonutils.dumps(expected_body,
@@ -160,27 +148,6 @@ class NsxV3ClientCertificateTestCase(nsxlib_testcase.NsxClientTestCase):
         self.assertRaises(nsxlib_exc.ObjectAlreadyExists,
                           cert.generate, {})
 
-    def test_generate_cert_with_retry(self):
-        """Test startup without certificate + certificate generation"""
-
-        storage_driver = DummyStorageDriver()
-        # Prepare fake trust management for "cert create" requests
-        mocked_trust = self._get_mocked_trust('retry-create')
-        cert = client_cert.ClientCertificateManager(self.identity,
-                                                    mocked_trust,
-                                                    storage_driver)
-        self.assertFalse(cert.exists())
-        cert.generate(subject={}, key_size=4096, valid_for_days=3)
-
-        # verify client cert was generated and makes sense
-        self.assertTrue(cert.exists())
-
-        # verify cert ans PK were stored in storage
-        cert_pem, key_pem = cert.get_pem()
-        stored_cert, stored_key = storage_driver.get_cert(self.identity)
-        self.assertEqual(cert_pem, stored_cert)
-        self.assertEqual(key_pem, stored_key)
-
     def _prepare_storage_with_existing_cert(self, key_size, days, alg, subj):
         # prepare storage driver with existing cert and key
         # this test simulates system startup

vmware_nsxlib/v3/client_cert.py

@@ -16,6 +16,7 @@
 import datetime
 from OpenSSL import crypto
 from time import time
+import uuid
 
 from neutron_lib import exceptions
 from oslo_log import log
@@ -142,7 +143,7 @@ class ClientCertificateManager(object):
         self._key = None
 
     def generate(self, subject, key_size=2048, valid_for_days=3650,
-                 signature_alg='sha256'):
+                 signature_alg='sha256', node_id=None):
         """Generate new certificate and register it in the system
 
         Generate certificate with RSA key based on arguments provided,
@@ -157,7 +158,7 @@ class ClientCertificateManager(object):
                                                    subject)
 
         # register on backend
-        self._register_cert(cert)
+        self._register_cert(cert, node_id or uuid.uuid4())
 
         # save in storage
         cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
@@ -210,7 +211,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):
+    def import_pem(self, filename, node_id=None):
         """Import and register existing certificate in PEM format"""
 
         # TODO(annak): support PK import as well
@@ -231,7 +232,7 @@ class ClientCertificateManager(object):
                 msg=_("Failed to import client certificate"))
 
         # register on backend
-        self._register_cert(cert)
+        self._register_cert(cert, node_id or uuid.uuid4())
 
         self._storage_driver.store_cert(self._identity, cert_pem, None)
 
@@ -325,25 +326,13 @@ class ClientCertificateManager(object):
 
         return cert, key
 
-    def _register_cert(self, cert):
+    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)
-        try:
+        self._nsx_trust_management.create_identity(self._identity,
-            self._nsx_trust_management.create_identity(self._identity,
+                                                   nsx_cert_id,
-                                                       nsx_cert_id)
+                                                   node_id,
-        except nsxlib_exceptions.ManagerError as e:
+                                                   'read_write_api_users')
-            if e.error_code != NSX_ERROR_IDENTITY_EXISTS:
-                raise e
-
-            # principal identity already exists - this can happen
-            # due to temporary error on deletion. Worth retrying.
-            # TODO(annak): remove this code once
-            # NSX supports multiple certificates per identity
-            details = self._nsx_trust_management.get_identity_details(
-                self._identity)
-            self._nsx_trust_management.delete_identity(details['id'])
-            self._nsx_trust_management.create_identity(self._identity,
-                                                       nsx_cert_id)
 
 
 class ClientCertProvider(object):

vmware_nsxlib/v3/trust_management.py

@@ -12,13 +12,16 @@
 #    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 vmware_nsxlib.v3 import exceptions as nsxlib_exc
 from vmware_nsxlib.v3 import utils
 
 BASE_SECTION = 'trust-management'
 CERT_SECTION = BASE_SECTION + '/certificates'
 ID_SECTION = BASE_SECTION + '/principal-identities'
+USER_GROUP_TYPES = [
+    'read_only_api_users',
+    'read_write_api_users',
+    'superusers']
 
 
 class NsxLibTrustManagement(utils.NsxLibApiBase):
@@ -51,8 +54,17 @@ class NsxLibTrustManagement(utils.NsxLibApiBase):
         resource = CERT_SECTION + '/' + cert_id
         self.client.delete(resource)
 
-    def create_identity(self, identity, cert_id):
+    def create_identity(self, identity, cert_id,
-        body = {'name': identity, 'certificate_id': cert_id}
+                        node_id, permission_group):
+        # Validate permission group before sending to server
+        if permission_group not in USER_GROUP_TYPES:
+            raise nsxlib_exc.InvalidInput(
+                operation='create_identity',
+                arg_val=permission_group,
+                arg_name='permission_group')
+        body = {'name': identity, '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):