Merge pull request #111 from Venafi/pm_updates

Fixed issues with test suite on contacts test

Author: rvelaVenafi

tests/test_pm.py

@@ -139,10 +139,10 @@ def test_validate_identity(self):
 
     def test_create_and_get_policy_with_contacts(self):
         connector = self.tpp_conn
-        zone = f"{TPP_PM_ROOT}\\{_get_tpp_policy_name()}"
+        zone = f"{TPP_PM_ROOT}\\{get_tpp_policy_name()}"
         policy_specification = PolicySpecification()
-        policy_specification.policy = _get_policy_obj(ca_type=CA_TYPE_TPP)
-        policy_specification.defaults = _get_defaults_obj()
+        policy_specification.policy = get_policy_obj(ca_type=CA_TYPE_TPP)
+        policy_specification.defaults = get_defaults_obj()
         policy_specification.policy.key_pair.rsa_key_sizes = [2048]
         connector.set_policy(zone, policy_specification)
         result = connector.get_policy(zone)
@@ -247,22 +247,22 @@ def test_create_policy_uri_ip_email(self):
         self.assertTrue(ps.policy.subject_alt_names.ip_allowed)
 
     def test_create_policy_with_no_users(self):
-        zone = self._get_random_zone()
+        zone = get_vaas_zone()
         connector = self.cloud_conn
         policy_specification = PolicySpecification()
-        policy_specification.policy = _get_policy_obj()
-        policy_specification.defaults = _get_defaults_obj()
+        policy_specification.policy = get_policy_obj()
+        policy_specification.defaults = get_defaults_obj()
         connector.set_policy(zone, policy_specification)
         result = connector.get_policy(zone)
         self.assertEqual(1, len(result.users))
         self.assertEqual("jenkins@opensource.qa.venafi.io", result.users[0])
 
     def test_create_policy_with_users(self):
-        zone = self._get_random_zone()
+        zone = get_vaas_zone()
         connector = self.cloud_conn
         policy_specification = PolicySpecification()
-        policy_specification.policy = _get_policy_obj()
-        policy_specification.defaults = _get_defaults_obj()
+        policy_specification.policy = get_policy_obj()
+        policy_specification.defaults = get_defaults_obj()
         policy_specification.users = ["pki-admin@opensource.qa.venafi.io", "resource-owner@opensource.qa.venafi.io"]
         connector.set_policy(zone, policy_specification)
         result = connector.get_policy(zone)
@@ -271,13 +271,12 @@ def test_create_policy_with_users(self):
         self.assertIn("resource-owner@opensource.qa.venafi.io", result.users)
 
     def test_update_policy_with_no_users(self):
-        zone = self._get_random_zone()
+        zone = get_vaas_zone()
         connector = self.cloud_conn
         policy_specification = PolicySpecification()
-        policy_specification.policy = _get_policy_obj()
-        policy_specification.defaults = _get_defaults_obj()
-        policy_specification.users = ["pki-admin@opensource.qa.venafi.io",
-                                       "resource-owner@opensource.qa.venafi.io"]
+        policy_specification.policy = get_policy_obj()
+        policy_specification.defaults = get_defaults_obj()
+        policy_specification.users = ["pki-admin@opensource.qa.venafi.io", "resource-owner@opensource.qa.venafi.io"]
         connector.set_policy(zone, policy_specification)
         result = connector.get_policy(zone)
         self.assertEqual(2, len(result.users))
@@ -286,20 +285,20 @@ def test_update_policy_with_no_users(self):
 
         # Update Policy Specification with no users
         policy_specification2 = PolicySpecification()
-        policy_specification2.policy = _get_policy_obj()
-        policy_specification2.defaults = _get_defaults_obj()
+        policy_specification2.policy = get_policy_obj()
+        policy_specification2.defaults = get_defaults_obj()
         connector.set_policy(zone, policy_specification2)
         result2 = connector.get_policy(zone)
         self.assertEqual(2, len(result2.users))
         self.assertIn("pki-admin@opensource.qa.venafi.io", result2.users)
         self.assertIn("resource-owner@opensource.qa.venafi.io", result2.users)
 
     def test_update_policy_with_users(self):
-        zone = self._get_random_zone()
+        zone = get_vaas_zone()
         connector = self.cloud_conn
         policy_specification = PolicySpecification()
-        policy_specification.policy = _get_policy_obj()
-        policy_specification.defaults = _get_defaults_obj()
+        policy_specification.policy = get_policy_obj()
+        policy_specification.defaults = get_defaults_obj()
         policy_specification.users = ["jenkins@opensource.qa.venafi.io"]
         connector.set_policy(zone, policy_specification)
         result = connector.get_policy(zone)
@@ -308,22 +307,22 @@ def test_update_policy_with_users(self):
 
         # Update Policy Specification with users
         policy_specification2 = PolicySpecification()
-        policy_specification2.policy = _get_policy_obj()
-        policy_specification2.defaults = _get_defaults_obj()
+        policy_specification2.policy = get_policy_obj()
+        policy_specification2.defaults = get_defaults_obj()
         policy_specification2.users = ["pki-admin@opensource.qa.venafi.io",
-                                        "resource-owner@opensource.qa.venafi.io"]
+                                       "resource-owner@opensource.qa.venafi.io"]
         connector.set_policy(zone, policy_specification2)
         result2 = connector.get_policy(zone)
         self.assertEqual(2, len(result2.users))
         self.assertIn("pki-admin@opensource.qa.venafi.io", result2.users)
         self.assertIn("resource-owner@opensource.qa.venafi.io", result2.users)
 
     def test_create_policy_with_team(self):
-        zone = self._get_random_zone()
+        zone = get_vaas_zone()
         connector = self.cloud_conn
         policy_specification = PolicySpecification()
-        policy_specification.policy = _get_policy_obj()
-        policy_specification.defaults = _get_defaults_obj()
+        policy_specification.policy = get_policy_obj()
+        policy_specification.defaults = get_defaults_obj()
         policy_specification.users = [CLOUD_TEAM]
         connector.set_policy(zone, policy_specification)
         result = connector.get_policy(zone)
@@ -564,7 +563,11 @@ def test_exceptions_vaas(self):
                      elliptic_curves=["asd"],
                      service_generated=True)
         ps.policy.key_pair = kp
-        ps.policy.subject_alt_names = SubjectAltNames(dns_allowed=True,  email_allowed=True)
+        ps.policy.subject_alt_names = SubjectAltNames(
+            dns_allowed=True,
+            email_allowed=True,
+            upn_allowed=True
+        )
         s = Subject(orgs=["Venafi"],
                     org_units=["QA Venafi"],
                     localities=["Salt Lake City"],
@@ -584,18 +587,13 @@ def test_exceptions_vaas(self):
         ps.defaults.key_pair = dkp
 
         # validate key pair values
-        try:
-            validate_ps_vaas(ps)
-        except VenafiError as err:
-            msg = "Key Type values exceeded. Only one Key Type is allowed by VaaS"
-            self.assertEqual(err.args[0], msg)
-        ps.policy.key_pair.key_types = ["foo"]
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
             msg = f"Key Type [{ps.policy.key_pair.key_types[0]}] is not supported by VaaS"
             self.assertEqual(err.args[0], msg)
         ps.policy.key_pair.key_types = ["RSA"]
+
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
@@ -607,9 +605,9 @@ def test_exceptions_vaas(self):
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
-            msg = "Subject Alt name [SubjAltNameEmailAllowed] is not allowed by VaaS"
+            msg = "Subject Alt name [SubjAltNameUpnAllowed] is not allowed by VaaS"
             self.assertEqual(err.args[0], msg)
-        ps.policy.subject_alt_names.email_allowed = False
+        ps.policy.subject_alt_names.upn_allowed = False
 
         # validate default subject values against policy values
         try:
@@ -618,24 +616,28 @@ def test_exceptions_vaas(self):
             msg = default_error_msg.format('Organization', ds.org, s.orgs)
             self.assertEqual(err.args[0], msg)
         ps.defaults.subject.org = s.orgs[0]
+
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
             msg = default_error_msg.format('Org Units', ds.org_units, s.org_units)
             self.assertEqual(err.args[0], msg)
         ps.defaults.subject.org_units = s.org_units
+
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
             msg = default_error_msg.format('Localities', ds.locality, s.localities)
             self.assertEqual(err.args[0], msg)
         ps.defaults.subject.locality = s.localities[0]
+
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
             msg = default_error_msg.format('States', ds.state, s.states)
             self.assertEqual(err.args[0], msg)
         ps.defaults.subject.state = s.states[0]
+
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
@@ -649,24 +651,28 @@ def test_exceptions_vaas(self):
         except VenafiError as err:
             msg = default_error_msg.format('Key Types', dkp.key_type, kp.key_types)
             self.assertEqual(err.args[0], msg)
+
         ps.defaults.key_pair.key_type = kp.key_types[0]
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
             msg = default_error_msg.format('RSA Key Sizes', dkp.rsa_key_size, kp.rsa_key_sizes)
             self.assertEqual(err.args[0], msg)
+
         ps.defaults.key_pair.rsa_key_size = kp.rsa_key_sizes[0]
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
             msg = default_error_msg.format('Elliptic Curves', dkp.elliptic_curve, kp.elliptic_curves)
             self.assertEqual(err.args[0], msg)
+
         ps.defaults.key_pair.elliptic_curve = kp.elliptic_curves[0]
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
             msg = default_error_msg.format('Service Generated', dkp.service_generated, kp.service_generated)
             self.assertEqual(err.args[0], msg)
+
         ps.defaults.key_pair.service_generated = kp.service_generated
 
         # validate default values when policy is not defined
@@ -682,12 +688,14 @@ def test_exceptions_vaas(self):
         except VenafiError as err:
             msg = f"Default Key Type [{dkp2.key_type}] is not supported by VaaS"
             self.assertEqual(err.args[0], msg)
+
         ps.defaults.key_pair.key_type = "RSA"
         try:
             validate_ps_vaas(ps)
         except VenafiError as err:
-            msg = f"Default Key Size [{256}] is not supported by VaaS"
+            msg = f"Default RSA Key Size [{256}] is not supported by VaaS"
             self.assertEqual(err.args[0], msg)
+
         ps.defaults.key_pair.rsa_key_size = 4096
 
 

vcert/connection_cloud.py

@@ -933,12 +933,16 @@ def _get_service_generated_csr_attr(self, request, zone):
             csr_attr_map[CSR_ATTR_COUNTRY] = ps.defaults.subject.country
 
         if len(request.san_dns) > 0:
-            sans = {
-                CSR_ATTR_SANS_DNS: request.san_dns,
-                CSR_ATTR_SANS_IP_ADDR: request.ip_addresses,
-                CSR_ATTR_SANS_EMAIL_ADDR: request.email_addresses,
-                CSR_ATTR_SANS_URIS: request.uniform_resource_identifiers
-            }
+            sans = dict()
+            if request.san_dns and len(request.san_dns) > 0:
+                sans[CSR_ATTR_SANS_DNS] = request.san_dns
+            if request.ip_addresses and len(request.ip_addresses) > 0:
+                sans[CSR_ATTR_SANS_IP_ADDR] = request.ip_addresses
+            if request.email_addresses and len(request.email_addresses) > 0:
+                sans[CSR_ATTR_SANS_EMAIL_ADDR] = request.email_addresses
+            if request.uniform_resource_identifiers and len(request.uniform_resource_identifiers) > 0:
+                sans[CSR_ATTR_SANS_URIS] = request.uniform_resource_identifiers
+
             csr_attr_map[CSR_ATTR_SANS_BY_TYPE] = sans
 
         if request.key_type:
@@ -953,7 +957,7 @@ def _get_service_generated_csr_attr(self, request, zone):
                 req_kt_option = request.key_type.option
                 if request.key_type.key_type.lower() == KeyType.RSA:
                     policy_rsa_sizes = ps.policy.key_pair.rsa_key_sizes
-                    valid = value_matches_regex(value=req_kt_option, pattern_list=policy_rsa_sizes)
+                    valid = True if req_kt_option in policy_rsa_sizes else False
                     if not valid:
                         rsa_str = "RSA Key Size"
                         log.error(MSG_VALUE_NOT_MATCH_POLICY.format(rsa_str, f"{rsa_str}s", req_kt_option,
@@ -970,11 +974,11 @@ def _get_service_generated_csr_attr(self, request, zone):
             kt_param = {
                 CSR_ATTR_KEY_TYPE: request.key_type.key_type.upper()
             }
-            kt_option = request.key_type.option.upper()
-            if kt_option == KeyType.RSA:
-                kt_param[CSR_ATTR_KEY_LENGTH] = kt_option
-            elif request.key_type.key_type == KeyType.ECDSA:
-                kt_param[CSR_ATTR_KEY_CURVE] = kt_option
+            kt_type = request.key_type.key_type.lower()
+            if kt_type == KeyType.RSA:
+                kt_param[CSR_ATTR_KEY_LENGTH] = request.key_type.option
+            elif kt_type == KeyType.ECDSA:
+                kt_param[CSR_ATTR_KEY_CURVE] = request.key_type.option.upper()
 
             csr_attr_map[CSR_ATTR_KEY_TYPE_PARAMS] = kt_param
         elif ps.defaults and ps.defaults.key_pair:

vcert/policy/pm_cloud.py

@@ -558,6 +558,12 @@ def build_cit_request(ps, ca_details):
                 ec_kt['keyCurves'] = ['P256']
 
             key_types.append(ec_kt)
+    else:
+        rsa_kt = {
+         'keyType': KeyType.RSA.upper(),
+         'keyLengths': [2048]
+        }
+        key_types.append(rsa_kt)
 
     request['keyTypes'] = key_types