Merge pull request #76 from Venafi/flexible_validity_period

- Added support for flexible validity periods.

Author: rvelaVenafi

tests/test_e2e.py

@@ -14,11 +14,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import random
-import string
 from os import environ
 
-from future.backports.datetime import datetime
 from six import text_type
 
 FAKE = environ.get("FAKE")
@@ -43,12 +40,3 @@
 
 if RANDOM_DOMAIN and not isinstance(RANDOM_DOMAIN, text_type):
     RANDOM_DOMAIN = RANDOM_DOMAIN.decode()
-
-
-def random_word(length):
-    letters = string.ascii_lowercase
-    return ''.join(random.choice(letters) for _ in range(length))
-
-
-def timestamp():
-    return datetime.today().strftime('%Y.%m.%d-%Hh%Mm%Ss')

tests/test_env.py

@@ -161,7 +161,7 @@
 
 chain = [ca_cert, ca_cert]
 
-log = logger.get_child("test-local-methods")
+log = logger.get_child("test-local")
 
 
 class TestLocalMethods(unittest.TestCase):
@@ -347,4 +347,3 @@ def test_pkcs12_plain_pk(self):
         cert = Certificate(cert=pkcs12_plain_cert, chain=chain, key=pkcs12_plain_pk)
         output = cert.as_pkcs12()
         log.info("PKCS12 created successfully:\n%s" % output)
-        pass

tests/test_local_methods.py

@@ -15,14 +15,14 @@
 # limitations under the License.
 #
 
-import logging
 import os
 import unittest
 from pprint import pformat
 
 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, timestamp
-from vcert import TPPTokenConnection, CloudConnection, Authentication, SCOPE_PM
+    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
 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, \
@@ -33,8 +33,7 @@
 POLICY_SPEC_YAML = '/resources/policy_specification.yaml'
 CA_TYPE_TPP = 'TPP'
 
-logging.basicConfig(level=logging.DEBUG)
-logger = logging.getLogger('vcert-test')
+log = logger.get_child("test-pm")
 
 
 class TestParsers(unittest.TestCase):
@@ -167,8 +166,8 @@ def create_policy(connector, zone, policy_spec=None, policy=None, defaults=None)
     connector.set_policy(zone, policy_spec)
     resp = connector.get_policy(zone)
     data = parse_policy_spec(resp)
-    logger.debug('Created Policy at %s' % zone)
-    logger.debug(pformat(data))
+    log.debug('Created Policy at %s' % zone)
+    log.debug(pformat(data))
     return resp
 
 
@@ -251,7 +250,7 @@ def _get_tpp_policy_name():
 
 def _resolve_resources_path(path):
     resources_dir = os.path.dirname(__file__)
-    logger.debug('Testing root folder: [%s]' % resources_dir)
+    log.debug('Testing root folder: [%s]' % resources_dir)
     resolved_path = ('.%s' % path) if resources_dir.endswith('tests') else ('./tests%s' % path)
-    logger.debug('resolved path: [%s]' % resolved_path)
+    log.debug('resolved path: [%s]' % resolved_path)
     return resolved_path

tests/test_pm.py

@@ -20,14 +20,13 @@
 import unittest
 
 from assets import SSH_CERT_DATA, SSH_PRIVATE_KEY, SSH_PUBLIC_KEY
-from test_env import timestamp, TPP_TOKEN_URL, TPP_USER, TPP_PASSWORD, TPP_SSH_CADN
+from test_env import TPP_TOKEN_URL, TPP_USER, TPP_PASSWORD, TPP_SSH_CADN
+from test_utils import timestamp
 from vcert import CommonConnection, SSHCertRequest, TPPTokenConnection, Authentication, \
-    SCOPE_SSH, write_ssh_files
+    SCOPE_SSH, write_ssh_files, logger
 from vcert.ssh_utils import SSHRetrieveResponse, SSHKeyPair
 
-logging.basicConfig(level=logging.DEBUG)
-logger = logging.getLogger('vcert-test')
-
+log = logger.get_child("test-ssh")
 
 SERVICE_GENERATED_NO_KEY_ERROR = "%s key data is %s empty for Certificate %s"  # type: str
 SSH_CERT_DATA_ERROR = "Certificate data is empty for Certificate %s"  # type: str
@@ -84,7 +83,7 @@ def test_write_ssh_files(self):
         with open(full_path, "r") as priv_key_file:
             s_priv_key = priv_key_file.read()
             expected_priv_key = SSH_PRIVATE_KEY
-            if platform.system() is not "Windows":
+            if platform.system() != "Windows":
                 expected_priv_key = expected_priv_key.replace("\r\n", "\n")
 
             self.assertTrue(expected_priv_key == s_priv_key, err_msg % "SSH Private Key")

tests/test_ssh.py

@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+#
+# Copyright 2021 Venafi, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import binascii
+import time
+import unittest
+from datetime import datetime, timedelta
+
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes
+
+from assets import TEST_KEY_ECDSA, TEST_KEY_RSA_4096, TEST_KEY_RSA_2048_ENCRYPTED
+from test_env import TPP_ZONE, TPP_ZONE_ECDSA, TPP_USER, TPP_PASSWORD, TPP_TOKEN_URL
+from test_utils import random_word, enroll, renew, renew_by_thumbprint, renew_without_key_reuse, \
+    enroll_with_zone_update, simple_enroll
+from vcert import CustomField, KeyType, RevocationRequest, CertificateRequest, IssuerHint, logger, \
+    TPPTokenConnection
+from vcert.errors import ClientBadData, ServerUnexptedBehavior
+
+log = logger.get_child("test-tpp-token")
+
+
+class TestTPPTokenMethods(unittest.TestCase):
+    def __init__(self, *args, **kwargs):
+        self.tpp_zone = TPP_ZONE
+        self.tpp_zone_ecdsa = TPP_ZONE_ECDSA
+        self.tpp_conn = TPPTokenConnection(url=TPP_TOKEN_URL, user=TPP_USER, password=TPP_PASSWORD,
+                                           http_request_kwargs={"verify": "/tmp/chain.pem"})
+        super(TestTPPTokenMethods, self).__init__(*args, **kwargs)
+
+    def test_tpp_token_enroll(self):
+        cn = random_word(10) + ".venafi.example.com"
+        try:
+            cert_id, pkey, cert, _, cert_guid = enroll(self.tpp_conn, self.tpp_zone, cn)
+            cert_config = self.tpp_conn._get_certificate_details(cert_guid)
+            self.assertEqual(cert_config["Origin"], "Venafi VCert-Python")
+        except Exception as err:
+            self.fail("Error in test: %s" % err.message)
+
+    def test_tpp_token_enroll_with_service_generated_csr(self):
+        cn = random_word(10) + ".venafi.example.com"
+        try:
+            _, _, _, _, cert_guid = enroll(self.tpp_conn, self.tpp_zone, cn=cn, password="FooBarPass123",
+                                           service_generated_csr=True)
+            cert_config = self.tpp_conn._get_certificate_details(cert_guid)
+            self.assertEqual(cert_config["Origin"], "Venafi VCert-Python")
+        except Exception as err:
+            self.fail("Error in test: %s" % err.message)
+
+    def test_tpp_token_enroll_with_custom_fields(self):
+        cn = random_word(10) + ".venafi.example.com"
+        custom_fields = [
+            CustomField(name="custom", value="pythonTest"),
+            CustomField(name="cfList", value="item2"),
+            CustomField(name="cfListMulti", value="tier1"),
+            CustomField(name="cfListMulti", value="tier4")
+        ]
+        try:
+            cert_id, pkey, cert, _, cert_guid = enroll(conn=self.tpp_conn, zone=self.tpp_zone, cn=cn,
+                                                       custom_fields=custom_fields)
+            cert_config = self.tpp_conn._get_certificate_details(cert_guid)
+            self.assertEqual(cert_config["Origin"], "Venafi VCert-Python")
+        except Exception as err:
+            self.fail("Error in test: %s" % err.__str__)
+
+    def test_tpp_token_enroll_origin(self):
+        cn = random_word(10) + ".venafi.example.com"
+        try:
+            cert_id, pkey, cert, _, _ = enroll(self.tpp_conn, self.tpp_zone, cn)
+        except Exception as err:
+            self.fail("Error in test: %s" % err.__str__())
+
+    def test_tpp_token_renew(self):
+        cn = random_word(10) + ".venafi.example.com"
+        cert_id, pkey, cert, _, _ = enroll(self.tpp_conn, self.tpp_zone, cn)
+        cert = renew(self.tpp_conn, cert_id, pkey, cert.serial_number, cn)
+
+    def test_tpp_token_renew_twice(self):
+        cn = random_word(10) + ".venafi.example.com"
+        cert_id, pkey, cert, _, _ = enroll(self.tpp_conn, self.tpp_zone, cn)
+        time.sleep(5)
+        renew(self.tpp_conn, cert_id, pkey, cert.serial_number, cn)
+        time.sleep(5)
+        renew(self.tpp_conn, cert_id, pkey, cert.serial_number, cn)
+
+    def test_tpp_token_renew_by_thumbprint(self):
+        cn = random_word(10) + ".venafi.example.com"
+        cert_id, pkey, cert, _, _ = enroll(self.tpp_conn, self.tpp_zone, cn)
+        renew_by_thumbprint(self.tpp_conn, cert)
+
+    def test_tpp_token_renew_without_key_reuse(self):
+        renew_without_key_reuse(self, self.tpp_conn, self.tpp_zone)
+
+    def test_tpp_token_enroll_ecdsa(self):
+        cn = random_word(10) + ".venafi.example.com"
+        enroll(self.tpp_conn, self.tpp_zone_ecdsa, cn, TEST_KEY_ECDSA[0], TEST_KEY_ECDSA[1])
+
+    def test_tpp_token_enroll_with_custom_key(self):
+        cn = random_word(10) + ".venafi.example.com"
+        enroll(self.tpp_conn, self.tpp_zone, cn, TEST_KEY_RSA_4096[0], TEST_KEY_RSA_4096[1])
+
+    def test_tpp_token_enroll_with_encrypted_key(self):
+        cn = random_word(10) + ".venafi.example.com"
+        enroll(self.tpp_conn, self.tpp_zone, cn, TEST_KEY_RSA_2048_ENCRYPTED[0], TEST_KEY_RSA_2048_ENCRYPTED[1],
+               'venafi')
+
+    def test_tpp_token_enroll_with_custom_csr(self):
+        key = open("/tmp/csr-test.key.pem").read()
+        csr = open("/tmp/csr-test.csr.csr").read()
+        enroll(self.tpp_conn, self.tpp_zone, private_key=key, csr=csr)
+
+    def test_tpp_token_enroll_with_zone_update_and_custom_origin(self):
+        cn = random_word(10) + ".venafi.example.com"
+        cert, cert_guid = enroll_with_zone_update(self.tpp_conn, self.tpp_zone_ecdsa, cn)
+        cert_config = self.tpp_conn._get_certificate_details(cert_guid)
+        self.assertEqual(cert_config["Origin"], "Python-SDK ECDSA")
+        cert = x509.load_pem_x509_certificate(cert.cert.encode(), default_backend())
+        key = cert.public_key()
+        self.assertEqual(key.curve.name, "secp521r1")
+
+    def test_tpp_token_read_zone_config(self):
+        zone = self.tpp_conn.read_zone_conf(self.tpp_zone)
+        self.assertEqual(zone.country.value, "US")
+        self.assertEqual(zone.province.value, "Utah")
+        self.assertEqual(zone.locality.value, "Salt Lake")
+        self.assertEqual(zone.organization.value, "Venafi Inc.")
+        self.assertEqual(zone.organizational_unit.value, ["Integrations"])
+        self.assertEqual(zone.key_type.key_type, KeyType.RSA)
+        self.assertEqual(zone.key_type.option, 2048)
+
+    def test_tpp_token_read_zone_unknown_zone(self):
+        with self.assertRaises(Exception):
+            self.tpp_conn.read_zone_conf("fdsfsd")
+
+    def test_tpp_token_retrieve_non_issued(self):
+        with self.assertRaises(Exception):
+            self.tpp_conn.retrieve_cert(self.tpp_zone + "\\devops\\vcert\\test-non-issued.example.com")
+
+    def test_tpp_token_search_by_thumbprint(self):
+        req, cert = simple_enroll(self.tpp_conn, self.tpp_zone)
+        cert = x509.load_pem_x509_certificate(cert.cert.encode(), default_backend())
+        fingerprint = binascii.hexlify(cert.fingerprint(hashes.SHA1())).decode()
+        found = self.tpp_conn.search_by_thumbprint(fingerprint)
+        self.assertEqual(found, req.id)
+
+    def test_token_revoke_not_issued(self):
+        req = RevocationRequest(req_id=self.tpp_zone + '\\not-issued.example.com')
+        with self.assertRaises(Exception):
+            self.tpp_conn.revoke_cert(req)
+        req = RevocationRequest(thumbprint="2b25ff9f8725dfee37c6a7adcba31897b12e921d")
+        with self.assertRaises(Exception):
+            self.tpp_conn.revoke_cert(req)
+        req = RevocationRequest()
+        with self.assertRaises(Exception):
+            self.tpp_conn.revoke_cert(req)
+
+    def test_token_revoke_normal(self):
+        req, cert = simple_enroll(self.tpp_conn, self.tpp_zone)
+        rev_req = RevocationRequest(req_id=req.id)
+        self.tpp_conn.revoke_cert(rev_req)
+        time.sleep(1)
+        with self.assertRaises(Exception):
+            self.tpp_conn.renew_cert(req)
+
+    def test_token_revoke_without_disable(self):
+        req, cert = simple_enroll(self.tpp_conn, self.tpp_zone)
+        rev_req = RevocationRequest(req_id=req.id, disable=False)
+        self.tpp_conn.revoke_cert(rev_req)
+        time.sleep(1)
+        self.tpp_conn.renew_cert(req)
+
+    def test_token_revoke_normal_thumbprint(self):
+        req, cert = simple_enroll(self.tpp_conn, self.tpp_zone)
+        cert = x509.load_pem_x509_certificate(cert.cert.encode(), default_backend())
+        thumbprint = binascii.hexlify(cert.fingerprint(hashes.SHA1())).decode()
+        rev_req = RevocationRequest(thumbprint=thumbprint)
+        self.tpp_conn.revoke_cert(rev_req)
+        time.sleep(1)
+        with self.assertRaises(Exception):
+            self.tpp_conn.renew_cert(req)
+
+    def test_tpp_token_enroll_valid_hours(self):
+        cn = random_word(10) + ".venafi.example.com"
+        request = CertificateRequest(common_name=cn)
+
+        request.san_dns = [u"www.client.venafi.example.com", u"ww1.client.venafi.example.com"]
+        request.email_addresses = [u"e1@venafi.example.com", u"e2@venafi.example.com"]
+        request.ip_addresses = [u"127.0.0.1", u"192.168.1.1"]
+        request.user_principal_names = [u"e1@venafi.example.com", u"e2@venafi.example.com"]
+        request.uniform_resource_identifiers = [u"https://www.venafi.com", u"https://venafi.cloud"]
+
+        custom_fields = [
+            CustomField(name="custom", value="pythonTest"),
+            CustomField(name="cfList", value="item2"),
+            CustomField(name="cfListMulti", value="tier1"),
+            CustomField(name="cfListMulti", value="tier4")
+        ]
+
+        request.custom_fields = custom_fields
+        request.validity_hours = 144
+        request.issuer_hint = IssuerHint.MICROSOFT
+        expected_date = datetime.utcnow() + timedelta(hours=request.validity_hours)
+
+        self.tpp_conn.request_cert(request, self.tpp_zone)
+        cert = self.tpp_conn.retrieve_cert(request)
+
+        cert = x509.load_pem_x509_certificate(cert.cert.encode(), default_backend())
+        assert isinstance(cert, x509.Certificate)
+        expiration_date = cert.not_valid_after
+        # Due to some roundings and delays in operations on the server side, the certificate expiration date
+        # is not exactly the same as the one used in the request. A gap is allowed in this scenario to compensate
+        # this delays and roundings.
+        delta = timedelta(seconds=60)
+        date_format = "%Y-%m-%d %H:%M:%S"
+        self.assertAlmostEqual(expected_date, expiration_date, delta=delta,
+                               msg="Delta between expected and expiration date is too big.\nExpected: %s\nGot: %s\n"
+                                   "Expected_delta: %s seconds."
+                                   % (expected_date.strftime(date_format), expiration_date.strftime(date_format),
+                                      delta.total_seconds()))
+
+    def test_get_access_token(self):
+        try:
+            token_info = self.tpp_conn.get_access_token()
+            self.assertIsNotNone(token_info)
+            self.assertIsNotNone(token_info.access_token)
+            self.assertIsNotNone(token_info.refresh_token)
+            self.assertIsNotNone(token_info.expires)
+        except ClientBadData:
+            self.fail("Error in Test Data")
+        except ServerUnexptedBehavior as sub:
+            self.fail("Error from server: %s" % sub.__str__())
+
+    def test_refresh_access_token(self):
+        try:
+            self.tpp_conn.get_access_token()
+            refresh_info = self.tpp_conn.refresh_access_token()
+            self.assertIsNotNone(refresh_info)
+            self.assertIsNotNone(refresh_info.access_token)
+            self.assertIsNotNone(refresh_info.refresh_token)
+            self.assertIsNotNone(refresh_info.expires)
+        except ClientBadData:
+            self.fail("Error in Test Data")
+        except ServerUnexptedBehavior as sub:
+            self.fail("Error from server: %s" % sub.__str__())
+
+    def test_revoke_access_token(self):
+        try:
+            self.tpp_conn.get_access_token()
+            status, resp = self.tpp_conn.revoke_access_token()
+            self.assertEqual(status, 200)
+        except Exception as err:
+            self.fail("Error happened: %s" % err.__str__())
+
+        cn = random_word(10) + ".venafi.example.com"
+        with self.assertRaises(Exception):
+            enroll(self.tpp_conn, self.tpp_zone, cn)