Merge pull request #105 from Venafi/pm_updates

Service generated CSR enhancement for VaaS

Author: rvelaVenafi

tests/test_pm.py

@@ -19,14 +19,14 @@
 
 from test_env import (TPP_TOKEN_URL, CLOUD_APIKEY, CLOUD_URL, TPP_PM_ROOT, CLOUD_ENTRUST_CA_NAME,
                       CLOUD_DIGICERT_CA_NAME, TPP_CA_NAME, TPP_USER, TPP_PASSWORD, CLOUD_TEAM)
-from test_utils import timestamp
-from vcert import TPPTokenConnection, CloudConnection, Authentication, SCOPE_PM, logger, VenafiError
+from test_utils import get_tpp_policy_name, get_vaas_zone
+from vcert import TPPTokenConnection, CloudConnection, Authentication, SCOPE_PM, logger, VenafiError, KeyType
 from vcert.parser import json_parser, yaml_parser
 from vcert.parser.utils import parse_policy_spec
 from vcert.policy import (Policy, Subject, KeyPair, SubjectAltNames, Defaults, DefaultSubject, DefaultKeyPair,
                           PolicySpecification)
 from vcert.policy.pm_cloud import (CA_TYPE_DIGICERT, CA_TYPE_ENTRUST, validate_policy_spec as validate_ps_vaas,
-                                   get_ca_info, default_error_msg)
+                                   get_ca_info, default_error_msg, ipv4, ipv6)
 from vcert.policy.pm_tpp import (is_service_generated_csr, validate_policy_spec as validate_ps_tpp, no_match_error_msg,
                                  too_many_error_msg, unsupported_error_msg, supported_key_types,
                                  supported_rsa_key_sizes, supported_elliptic_curves)
@@ -50,7 +50,7 @@ def test_json_parsing(self):
         self._assert_policy_spec(ps)
 
     def test_json_serialization(self):
-        ps = PolicySpecification(policy=_get_policy_obj(), defaults=_get_defaults_obj())
+        ps = PolicySpecification(policy=get_policy_obj(), defaults=get_defaults_obj())
         json_parser.serialize(ps, 'test_json_serialization.json')
 
     def test_yaml_11_parsing(self):
@@ -61,7 +61,7 @@ def test_yaml_12_parsing(self):
         self._assert_policy_spec(ps)
 
     def test_yaml_serialization(self):
-        ps = PolicySpecification(policy=_get_policy_obj(), defaults=_get_defaults_obj())
+        ps = PolicySpecification(policy=get_policy_obj(), defaults=get_defaults_obj())
         yaml_parser.serialize(ps, 'test_yaml_serialization.yaml')
 
     def _assert_policy_spec(self, ps):
@@ -108,18 +108,18 @@ def test_create_policy_yaml(self):
         pass
 
     def test_create_policy_full(self):
-        policy = _get_policy_obj(ca_type=CA_TYPE_TPP)
+        policy = get_policy_obj(ca_type=CA_TYPE_TPP)
         policy.key_pair.rsa_key_sizes = [2048]
-        self._create_policy_tpp(policy=policy, defaults=_get_defaults_obj())
+        self._create_policy_tpp(policy=policy, defaults=get_defaults_obj())
 
     def test_create_policy_empty(self):
         self._create_policy_tpp()
 
     def test_create_policy_no_policy(self):
-        self._create_policy_tpp(defaults=_get_defaults_obj())
+        self._create_policy_tpp(defaults=get_defaults_obj())
 
     def test_create_policy_no_defaults(self):
-        policy = _get_policy_obj(ca_type=CA_TYPE_TPP)
+        policy = get_policy_obj(ca_type=CA_TYPE_TPP)
         policy.key_pair.rsa_key_sizes = [2048]
         self._create_policy_tpp(policy=policy)
 
@@ -149,7 +149,7 @@ def test_create_and_get_policy_with_contacts(self):
         self.assertEqual(1, len(result.users))
 
     def _create_policy_tpp(self, policy_spec=None, policy=None, defaults=None):
-        zone = f"{TPP_PM_ROOT}\\{_get_tpp_policy_name()}"
+        zone = f"{TPP_PM_ROOT}\\{get_tpp_policy_name()}"
         create_policy(self.tpp_conn, zone, policy_spec, policy, defaults)
 
 
@@ -171,27 +171,81 @@ def test_create_policy_yaml(self):
         pass
 
     def test_create_policy_full(self):
-        self._create_policy_cloud(policy=_get_policy_obj(), defaults=_get_defaults_obj())
+        self._create_policy_cloud(policy=get_policy_obj(), defaults=get_defaults_obj())
 
     def test_create_policy_empty(self):
         self._create_policy_cloud()
 
     def test_create_policy_no_policy(self):
-        self._create_policy_cloud(defaults=_get_defaults_obj())
+        self._create_policy_cloud(defaults=get_defaults_obj())
 
     def test_create_policy_no_defaults(self):
-        self._create_policy_cloud(policy=_get_policy_obj())
+        self._create_policy_cloud(policy=get_policy_obj())
 
     def test_create_policy_entrust(self):
-        self._create_policy_cloud(policy=_get_policy_obj(ca_type=CA_TYPE_ENTRUST), defaults=_get_defaults_obj())
+        self._create_policy_cloud(policy=get_policy_obj(ca_type=CA_TYPE_ENTRUST), defaults=get_defaults_obj())
 
     def test_create_policy_digicert(self):
-        self._create_policy_cloud(policy=_get_policy_obj(ca_type=CA_TYPE_DIGICERT), defaults=_get_defaults_obj())
+        self._create_policy_cloud(policy=get_policy_obj(ca_type=CA_TYPE_DIGICERT), defaults=get_defaults_obj())
 
     def test_validate_domains(self):
-        policy = self._create_policy_cloud(policy=_get_policy_obj())
+        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 test_ec_key_pair(self):
+        policy = get_policy_obj()
+        kp = KeyPair(
+            key_types=['EC'],
+            rsa_key_sizes=[2048, 4096],
+            elliptic_curves=['P521', 'P384'],
+            reuse_allowed=False)
+        policy.key_pair = kp
+
+        defaults = get_defaults_obj()
+        defaults.key_pair = DefaultKeyPair(
+            key_type='EC',
+            rsa_key_size=2048,
+            elliptic_curve='P521')
+
+        ps = self._create_policy_cloud(policy=policy, defaults=defaults)
+        self.assertEqual(ps.policy.key_pair.key_types[0].upper(), KeyType.ECDSA.upper(), "Policy Key Type is not EC")
+        self.assertTrue(len(ps.policy.key_pair.elliptic_curves) == 2,
+                        f"Expected 2 accepted Elliptic Curves. Got {len(ps.policy.key_pair.elliptic_curves)}")
+        self.assertIn('P521', ['P521', 'P384'], "[P521] is not in the allowed Elliptic Curves list")
+        self.assertIn('P384', ['P521', 'P384'], "[P384] is not in the allowed Elliptic Curves list")
+
+    def test_create_policy_uri_ip_email(self):
+        policy = get_policy_obj()
+        policy.subject_alt_names.email_allowed = True
+        policy.subject_alt_names.uri_allowed = True
+        policy.subject_alt_names.ip_allowed = True
+        uri_protocols = ["https", "ldaps", "spiffe"]
+        policy.subject_alt_names.uri_protocols = uri_protocols
+        ps = self._create_policy_cloud(policy=policy, defaults=get_defaults_obj())
+        self.assertListEqual(ps.policy.subject_alt_names.ip_constraints, [ipv4, ipv6])
+        self.assertListEqual(ps.policy.subject_alt_names.uri_protocols, uri_protocols)
+        self.assertTrue(ps.policy.subject_alt_names.email_allowed)
+        self.assertTrue(ps.policy.subject_alt_names.uri_allowed)
+        self.assertTrue(ps.policy.subject_alt_names.ip_allowed)
+
     def test_create_policy_with_no_users(self):
         zone = self._get_random_zone()
         connector = self.cloud_conn
@@ -277,13 +331,23 @@ def test_create_policy_with_team(self):
         self.assertEqual(CLOUD_TEAM, result.users[0])
 
     def _create_policy_cloud(self, policy_spec=None, policy=None, defaults=None):
-        zone = self._get_random_zone()
+        zone = get_vaas_zone()
         response = create_policy(self.cloud_conn, zone, policy_spec, policy, defaults)
         return response
 
-    @staticmethod
-    def _get_random_zone():
-        return _get_zone()
+    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 = get_vaas_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
 
 
 class TestLocalMethods(unittest.TestCase):
@@ -646,7 +710,7 @@ def create_policy(connector, zone, policy_spec=None, policy=None, defaults=None)
 POLICY_DOMAINS = ['vfidev.com', 'vfidev.net', 'venafi.example']
 
 
-def _get_policy_obj(ca_type=None):
+def get_policy_obj(ca_type=None):
     policy = Policy(
         subject=Subject(
             orgs=['OSS Venafi, Inc.'],
@@ -683,7 +747,7 @@ def _get_policy_obj(ca_type=None):
     return policy
 
 
-def _get_defaults_obj():
+def get_defaults_obj():
     defaults = Defaults(
         d_subject=DefaultSubject(
             org='OSS Venafi, Inc.',
@@ -697,24 +761,3 @@ def _get_defaults_obj():
             elliptic_curve='P521'),
         auto_installed=False)
     return defaults
-
-
-def _get_app_name():
-    name = 'vcert-python-app-{}'
-    return name
-
-
-def _get_cit_name():
-    cit_name = 'vcert-python-cit-{}'
-    return cit_name
-
-
-def _get_zone():
-    time = timestamp()
-    zone = f"{_get_app_name().format(time)}\\{_get_cit_name().format(time)}"
-    return zone
-
-
-def _get_tpp_policy_name():
-    time = timestamp()
-    return f"{_get_app_name().format(time)}"

tests/test_utils.py

@@ -39,6 +39,27 @@ def timestamp():
     return datetime.today().strftime('%Y.%m.%d-%Hh%Mm%Ss')
 
 
+def _get_app_name():
+    name = 'vcert-python-app-{}'
+    return name
+
+
+def _get_cit_name():
+    cit_name = 'vcert-python-cit-{}'
+    return cit_name
+
+
+def get_vaas_zone():
+    t = timestamp()
+    zone = f"{_get_app_name().format(t)}\\{_get_cit_name().format(t)}"
+    return zone
+
+
+def get_tpp_policy_name():
+    t = timestamp()
+    return f"{_get_app_name().format(t)}"
+
+
 def simple_enroll(conn, zone):
     req = CertificateRequest(common_name=f"{random_word(12)}.venafi.example.com")
     conn.request_cert(req, zone)

tests/test_vaas.py

@@ -21,11 +21,15 @@
 
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
 from cryptography.x509.oid import NameOID
 
+from policy import KeyPair, DefaultKeyPair, PolicySpecification
 from test_env import CLOUD_ZONE, CLOUD_APIKEY, CLOUD_URL, RANDOM_DOMAIN
-from test_utils import random_word, enroll, renew, renew_by_thumbprint, renew_without_key_reuse, simple_enroll
+from test_pm import get_policy_obj, get_defaults_obj
+from test_utils import random_word, enroll, renew, renew_by_thumbprint, renew_without_key_reuse, simple_enroll, \
+    get_vaas_zone
 from vcert import CloudConnection, KeyType, CertificateRequest, CustomField, logger, CSR_ORIGIN_SERVICE
 
 log = logger.get_child("test-vaas")
@@ -165,3 +169,49 @@ def test_cloud_enroll_service_generated_csr(self):
 
         output = cert_object.as_pkcs12('FooBarPass123')
         log.info(f"PKCS12 created successfully for certificate with CN: {cn}")
+
+    def test_enroll_ec_key_certificate(self):
+        policy = get_policy_obj()
+        kp = KeyPair(
+            key_types=['EC'],
+            elliptic_curves=['P521', 'P384'],
+            reuse_allowed=False)
+        policy.key_pair = kp
+
+        defaults = get_defaults_obj()
+        defaults.key_pair = DefaultKeyPair(
+            key_type='EC',
+            elliptic_curve='P521')
+
+        policy_spec = PolicySpecification()
+        policy_spec.policy = policy
+        policy_spec.defaults = defaults
+
+        zone = get_vaas_zone()
+
+        self.cloud_conn.set_policy(zone, policy_spec)
+        password = 'FooBarPass123'
+
+        request = CertificateRequest(
+            common_name=f"{random_word(10)}.venafi.example",
+            key_type=KeyType(
+                key_type="ec",
+                option="P384"
+            ),
+            csr_origin=CSR_ORIGIN_SERVICE,
+            key_password=password
+        )
+
+        self.cloud_conn.request_cert(request, zone)
+        cert = self.cloud_conn.retrieve_cert(request)
+
+        p_key = None
+        try:
+            p_key = serialization.load_pem_private_key(data=cert.key.encode(), password=password.encode(),
+                                                       backend=default_backend())
+        except Exception as e:
+            log.error(msg=f"Error parsing Private Key: {e.message}")
+
+        if p_key:
+            self.assertIsInstance(p_key, EllipticCurvePrivateKey, "returned private key is not of type Elliptic Curve")
+            self.assertEqual(p_key.curve.key_size, 384, f"Private Key expected curve: 384. Got: {p_key.curve.key_size}")

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,11 @@ 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
+        self.email_regexes = None  # type: list[str]
+        self.ip_constraints_regexes = None  # type: list[str]
+        self.uri_regexes = None  # type: list[str]
 
     def __repr__(self):
         return "Policy:\n" + "\n".join([f"  {k}: {v}" for k, v in (