Service generated CSR enhancement for VaaS

Added support for VaaS policy attributes 'csrUploadAllowed' and 'keyGeneratedByVenafiAllowed'

Author: rvelaVenafi

tests/test_pm.py

@@ -164,11 +164,43 @@ def test_validate_domains(self):
         policy = self._create_policy_cloud(policy=_get_policy_obj())
         self.assertListEqual(policy.policy.domains, POLICY_DOMAINS)
 
+    def test_csr_attributes_service(self):
+        cit = self._create_csr_attributes_policy(service_generated_csr=True)
+
+        self.assertFalse(cit.csr_upload_allowed, "csrUploadAllowed attribute is not False")
+        self.assertTrue(cit.key_generated_by_venafi_allowed, "keyGeneratedByVenafiAllowed is not True")
+
+    def test_csr_attributes_local(self):
+        cit = self._create_csr_attributes_policy(service_generated_csr=False)
+
+        self.assertTrue(cit.csr_upload_allowed, "csrUploadAllowed attribute is not True")
+        self.assertFalse(cit.key_generated_by_venafi_allowed, "keyGeneratedByVenafiAllowed is not False")
+
+    def test_csr_attributes_not_specified(self):
+        cit = self._create_csr_attributes_policy()
+
+        self.assertTrue(cit.csr_upload_allowed, "csrUploadAllowed attribute is not True")
+        self.assertTrue(cit.key_generated_by_venafi_allowed, "keyGeneratedByVenafiAllowed is not True")
+
     def _create_policy_cloud(self, policy_spec=None, policy=None, defaults=None):
         zone = self._get_random_zone()
         response = create_policy(self.cloud_conn, zone, policy_spec, policy, defaults)
         return response
 
+    def _create_csr_attributes_policy(self, service_generated_csr=None):
+        """
+
+        :param bool service_generated_csr:
+        :rtype: common.Policy
+        """
+        policy = _get_policy_obj()
+        policy.key_pair.service_generated = service_generated_csr
+        zone = self._get_random_zone()
+        create_policy(connector=self.cloud_conn, zone=zone, policy_spec=None, policy=policy)
+        cit = self.cloud_conn._get_template_by_id(zone)
+
+        return cit
+
     @staticmethod
     def _get_random_zone():
         return _get_zone()

vcert/common.py

@@ -179,7 +179,8 @@ def __init__(self, policy_id=None, company_id=None, name=None, system_generated=
                  subject_ou_regexes=None, subject_st_regexes=None, subject_l_regexes=None, subject_c_regexes=None,
                  san_regexes=None, key_types=None, key_reuse=None, cert_authority=None, cert_authority_account_id=None,
                  cert_authority_product_option_id=None, priority=None, modification_date=None, status=None, reason=None,
-                 validity_period=None, recommended_settings=None):
+                 validity_period=None, recommended_settings=None, csr_upload_allowed=None,
+                 key_generated_by_venafi_allowed=None):
         """
         :param str policy_id:
         :param str company_id:
@@ -204,6 +205,8 @@ def __init__(self, policy_id=None, company_id=None, name=None, system_generated=
         :param str reason:
         :param str validity_period:
         :param vaas_utils.RecommendedSettings recommended_settings:
+        :param bool csr_upload_allowed:
+        :param bool key_generated_by_venafi_allowed:
         """
         self.id = policy_id
         self.company_id = company_id
@@ -229,6 +232,8 @@ def __init__(self, policy_id=None, company_id=None, name=None, system_generated=
         self.reason = reason
         self.validity_period = validity_period
         self.recommended_settings = recommended_settings
+        self.csr_upload_allowed = csr_upload_allowed
+        self.key_generated_by_venafi_allowed = key_generated_by_venafi_allowed
 
     def __repr__(self):
         return "Policy:\n" + "\n".join([f"  {k}: {v}" for k, v in (

vcert/connection_cloud.py

@@ -247,7 +247,9 @@ def _parse_policy_response_to_object(d):
             d['status'] if 'status' in d else None,
             d['reason'] if 'reason' in d else None,
             d['validityPeriod'] if 'validityPeriod' in d else None,
-            None
+            None,
+            d['csrUploadAllowed'] if 'csrUploadAllowed' in d else None,
+            d['keyGeneratedByVenafiAllowed'] if 'keyGeneratedByVenafiAllowed' in d else None
         )
         for kt in d.get('keyTypes', []):
             key_type = kt['keyType'].lower()

vcert/policy/pm_cloud.py

@@ -103,6 +103,15 @@ def build_policy_spec(cit, ca_info, subject_cn_to_str=True):
         kp.rsa_key_sizes = key_sizes
 
     kp.reuse_allowed = cit.key_reuse
+    if cit.key_generated_by_venafi_allowed is True and cit.csr_upload_allowed is True:
+        kp.service_generated = None
+    elif cit.key_generated_by_venafi_allowed:
+        kp.service_generated = True
+        create_kp = True
+    elif cit.csr_upload_allowed:
+        kp.service_generated = False
+        create_kp = True
+
     p.key_pair = kp if create_kp else None
 
     sans = SubjectAltNames(False, False, False, False, False)
@@ -412,6 +421,14 @@ def build_cit_request(ps, ca_details):
     else:
         request['keyReuse'] = False
 
+    if ps.policy and ps.policy.key_pair and ps.policy.key_pair.service_generated is not None:
+        is_serv_gen = ps.policy.key_pair.service_generated
+        request['csrUploadAllowed'] = not is_serv_gen
+        request['keyGeneratedByVenafiAllowed'] = is_serv_gen
+    else:
+        request['csrUploadAllowed'] = True
+        request['keyGeneratedByVenafiAllowed'] = True
+
     r_settings = dict()
     if ps.defaults and ps.defaults.subject:
         if ps.defaults.subject.org: