Merge pull request #106 from Venafi/ec_support

Support for EC Key Types

Author: rvelaVenafi

tests/test_pm.py

@@ -19,8 +19,8 @@
 
 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)
-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,
@@ -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):
@@ -105,23 +105,23 @@ 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)
 
     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)
 
 
@@ -143,25 +143,25 @@ 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):
@@ -182,8 +182,30 @@ def test_csr_attributes_not_specified(self):
         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 _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
 
@@ -193,18 +215,14 @@ def _create_csr_attributes_policy(self, service_generated_csr=None):
         :param bool service_generated_csr:
         :rtype: common.Policy
         """
-        policy = _get_policy_obj()
+        policy = get_policy_obj()
         policy.key_pair.service_generated = service_generated_csr
-        zone = self._get_random_zone()
+        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
 
-    @staticmethod
-    def _get_random_zone():
-        return _get_zone()
-
 
 class TestLocalMethods(unittest.TestCase):
     def test_exceptions_tpp(self):
@@ -566,7 +584,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.'],
@@ -603,7 +621,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.',
@@ -617,24 +635,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}")