- Added support for service generated CSR on certificate renewal.

- Fixed an issue with the zone parameter. Invalid use of the forward slash was not being checked.
- Updated example for service generated CSR. Expanded on comments, fixed errors.
- Cleaned up duplicated code.

Author: rvelaVenafi

examples/get_cert_service_tpp.py

@@ -17,11 +17,10 @@
 
 from __future__ import print_function
 
-from vcert import (CertificateRequest, venafi_connection, RevocationRequest, CSR_ORIGIN_SERVICE)
+from vcert import (CertificateRequest, venafi_connection, CSR_ORIGIN_SERVICE, CHAIN_OPTION_FIRST)
 import string
 import random
 import logging
-import time
 from os import environ
 
 logging.basicConfig(level=logging.INFO)
@@ -39,54 +38,60 @@ def main():
     # Connection will be chosen automatically based on which arguments are passed.
     # If token is passed Venafi Cloud connection will be used.
     # If user, password, and URL Venafi Platform (TPP) will be used.
-    conn = venafi_connection(url=url, user=user, password=password,
-                             http_request_kwargs={"verify": server_trust_bundle})
     # If your TPP server certificate signed with your own CA, or available only via proxy, you can specify
-    # a trust bundle using requests vars:
-    # conn = Connection(url=url, token=token, user=user, password=password,
-    #                  http_request_kwargs={"verify": "/path-to/bundle.pem"})
+    # a trust bundle using http_request_kwargs.
+    conn = venafi_connection(url=url, user=user, password=password, http_request_kwargs={"verify": server_trust_bundle})
 
+    # Build a Certificate request
     request = CertificateRequest(common_name=random_word(10) + ".venafi.example.com")
+    # Set the request to use a service generated CSR
     request.csr_origin = CSR_ORIGIN_SERVICE
+    # Include some Subject Alternative Names
     request.san_dns = ["www.client.venafi.example.com", "ww1.client.venafi.example.com"]
     request.email_addresses = ["e1@venafi.example.com", "e2@venafi.example.com"]
     request.ip_addresses = ["127.0.0.1", "192.168.1.1"]
     request.uniform_resource_identifiers = ["http://wgtest.com","https://ragnartest.com"]
     request.user_principal_names = ["e1@venafi.example.com", "e2@venafi.example.com"]
-    # Specify ordering certificates in chain. Root can be "first" or "last". By default it last. You also can
-    # specify "ignore" to ignore chain (supported only for Platform).
+    # Specify ordering certificates in chain. Root can be CHAIN_OPTION_FIRST ("first")
+    # or CHAIN_OPTION_LAST ("last"). By default it is CHAIN_OPTION_LAST.
+    # You can also specify CHAIN_OPTION_IGNORE ("ignore") to ignore chain (supported only for TPP).
+    # request.chain_option = CHAIN_OPTION_FIRST
     # To set Custom Fields for the certificate, specify an array of CustomField objects as name-value pairs
     # request.custom_fields = [
     #    CustomField(name="Cost Center", value="ABC123"),
     #    CustomField(name="Environment", value="Production"),
     #    CustomField(name="Environment", value="Staging")
     # ]
-    # Update certificate request from zone
+    #
+    # Update certificate request from zone.
     zone_config = conn.read_zone_conf(zone)
     request.update_from_zone_config(zone_config)
+    # Request the certificate.
     conn.request_cert(request, zone)
 
-    # and wait for signing
+    # Wait for the certificate to be retrieved.
+    # This operation may take some time to return, as it waits until the certificate is ISSUED or it timeout.
+    # Timeout is 180s by default. Can be changed using:
+    # request.timeout = 300
     cert = conn.retrieve_cert(request)
 
-    # after that print cert and key
-    print(cert.full_chain, request.private_key_pem, sep="\n")
-    # and save into file
-    f = open("/tmp/cert.pem", "w")
+    # Print the certificate
+    print(cert.full_chain)
+    # Save it into a file
+    f = open("./cert.pem", "w")
     f.write(cert.full_chain)
-    f = open("/tmp/cert.key", "w")
-    f.write(request.private_key_pem)
     f.close()
 
     print("Trying to renew certificate")
     new_request = CertificateRequest(cert_id=request.id)
+    # The renewal request should use a service generated CSR as well
+    # This may not be necessary and depends entirely on the settings of your Policy/Zone
+    new_request.csr_origin = CSR_ORIGIN_SERVICE
     conn.renew_cert(new_request)
     new_cert = conn.retrieve_cert(new_request)
-    print(new_cert.cert, new_request.private_key_pem, sep="\n")
-    fn = open("/tmp/new_cert.pem", "w")
+    print(new_cert.cert)
+    fn = open("./new_cert.pem", "w")
     fn.write(new_cert.cert)
-    fn = open("/tmp/new_cert.key", "w")
-    fn.write(new_request.private_key_pem)
     fn.close()
 
 

examples/get_cert_token.py

@@ -33,7 +33,7 @@ def main():
     user = environ.get('TPP_USER')
     password = environ.get('TPP_PASSWORD')
     url = environ.get('TPP_TOKEN_URL')
-    zone = environ.get("ZONE")
+    zone = environ.get("TPP_ZONE")
     fake = environ.get('FAKE')
 
     if fake:
@@ -54,7 +54,7 @@ def main():
     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.uniform_resource_identifiers = [u"http://wgtest.com",u"https://ragnartest.com"]
-    request.user_principal_names = [u"e1@venafi.example.com", u"e2@venafi.example.com"] 
+    request.user_principal_names = [u"e1@venafi.example.com", u"e2@venafi.example.com"]
     # Specify ordering certificates in chain. Root can be "first" or "last". By default its last. You also can
     # specify "ignore" to ignore chain (supported only for Platform).
     # To set Custom Fields for the certificate, specify an array of CustomField objects as name-value pairs

vcert/__init__.py

@@ -16,7 +16,7 @@
 
 from .common import CertificateRequest, CommonConnection, RevocationRequest, ZoneConfig, CertField, KeyType, \
     CustomField, Authentication, SCOPE_CM, SCOPE_PM, SCOPE_SSH, CSR_ORIGIN_LOCAL, CSR_ORIGIN_PROVIDED, \
-    CSR_ORIGIN_SERVICE
+    CSR_ORIGIN_SERVICE, CHAIN_OPTION_FIRST, CHAIN_OPTION_IGNORE, CHAIN_OPTION_LAST
 from .connection_cloud import CloudConnection
 from .connection_tpp import TPPConnection
 from .connection_tpp_token import TPPTokenConnection

vcert/common.py

@@ -49,6 +49,9 @@
 CSR_ORIGIN_PROVIDED = "provided"
 CSR_ORIGIN_LOCAL = "local"
 CSR_ORIGIN_SERVICE = "service"
+CHAIN_OPTION_FIRST = "first"
+CHAIN_OPTION_LAST = "last"
+CHAIN_OPTION_IGNORE = "ignore"
 
 
 class CertField:
@@ -284,7 +287,7 @@ def __init__(self, cert_id=None,
         :param str csr_origin: The origin of the CSR, either user provided, locally generated or service generated.
         """
 
-        self.chain_option = "last"
+        self.chain_option = CHAIN_OPTION_LAST  # "last"
         self.san_dns = san_dns or []
         self.email_addresses = email_addresses
         self.ip_addresses = ip_addresses or []

vcert/connection_tpp.py

@@ -174,64 +174,6 @@ def revoke_cert(self, request):
 
         raise ServerUnexptedBehavior
 
-    def renew_cert(self, request, reuse_key=False):
-        if not request.id and not request.thumbprint:
-            log.debug("Request id or thumbprint must be specified for TPP")
-            raise CertificateRenewError
-        if not request.id and request.thumbprint:
-            request.id = self.search_by_thumbprint(request.thumbprint)
-        if reuse_key:
-            log.debug("Trying to renew certificate %s" % request.id)
-            status, data = self._post(URLS.CERTIFICATE_RENEW, data={"CertificateDN": request.id})
-            if not data['Success']:
-                raise CertificateRenewError
-            return
-        cert = self.retrieve_cert(request)
-        cert = x509.load_pem_x509_certificate(cert.cert.encode(), default_backend())
-        for a in cert.subject:
-            if a.oid == x509.NameOID.COMMON_NAME:
-                request.common_name = a.value
-            elif a.oid == x509.NameOID.COUNTRY_NAME:
-                request.country = a.value
-            elif a.oid == x509.NameOID.LOCALITY_NAME:
-                request.locality = a.value
-            elif a.oid == x509.NameOID.STATE_OR_PROVINCE_NAME:
-                request.province = a.value
-            elif a.oid == x509.NameOID.ORGANIZATION_NAME:
-                request.organization = a.value
-            elif a.oid == x509.NameOID.ORGANIZATIONAL_UNIT_NAME:
-                request.organizational_unit = a.value
-        for e in cert.extensions:
-            if e.oid == x509.OID_SUBJECT_ALTERNATIVE_NAME:
-                request.san_dns = list([x.value for x in e.value if isinstance(x, x509.DNSName)])
-                request.email_addresses = list([x.value for x in e.value if isinstance(x, x509.RFC822Name)])
-                request.ip_addresses = list([x.value.exploded for x in e.value if isinstance(x, x509.IPAddress)])
-                # remove header bytes from ASN1 encoded UPN field before setting it in the request object
-                upns = []
-                for x in e.value:
-                    if isinstance(x, x509.OtherName):
-                        upns.append(x.value[2::])
-                request.user_principal_names = upns
-                request.uniform_resource_identifiers = list(
-                    [x.value for x in e.value if isinstance(x, x509.UniformResourceIdentifier)])
-        if cert.signature_algorithm_oid in (AlgOID.ECDSA_WITH_SHA1, AlgOID.ECDSA_WITH_SHA224, AlgOID.ECDSA_WITH_SHA256,
-                                            AlgOID.ECDSA_WITH_SHA384, AlgOID.ECDSA_WITH_SHA512):
-            request.key_type = (KeyType.ECDSA, KeyType.ALLOWED_CURVES[0])
-        else:
-            request.key_type = KeyType(KeyType.RSA, 2048)  # todo: make parsing key size
-        if not request.csr:
-            request.build_csr()
-        status, data = self._post(URLS.CERTIFICATE_RENEW,
-                                  data={"CertificateDN": request.id, "PKCS10": request.csr})
-        if status == HTTPStatus.OK:
-            if "CertificateDN" in data:
-                request.id = data['CertificateDN']
-            log.debug("Certificate successfully requested with request id %s." % request.id)
-            return True
-
-        log.error("Request status is not %s. %s." % HTTPStatus.OK, status)
-        raise CertificateRequestError
-
     @staticmethod
     def _parse_zone_config_to_policy(data):
         # todo: parse over values to regexps (dont forget tests!)
@@ -284,28 +226,14 @@ def _parse_zone_data_to_object(data):
         return z
 
     def read_zone_conf(self, tag):
-        status, data = self._post(URLS.ZONE_CONFIG, {"PolicyDN": self._get_policy_dn(tag)})
+        status, data = self._post(URLS.ZONE_CONFIG, {"PolicyDN": self._normalize_zone(tag)})
         if status != HTTPStatus.OK:
             raise ServerUnexptedBehavior("Server returns %d status on reading zone configuration." % status)
         return self._parse_zone_data_to_object(data)
 
     def import_cert(self, request):
         raise NotImplementedError
 
-    def search_by_thumbprint(self, thumbprint):
-        """
-        :param str thumbprint:
-        """
-        thumbprint = re.sub(r'[^\dabcdefABCDEF]', "", thumbprint)
-        thumbprint = thumbprint.upper()
-        status, data = self._get(URLS.CERTIFICATE_SEARCH, params={"Thumbprint": thumbprint})
-        if status != HTTPStatus.OK:
-            raise ServerUnexptedBehavior
-
-        if not data['Certificates']:
-            raise ClientBadData("Certificate not found by thumbprint")
-        return data['Certificates'][0]['DN']
-
     def _read_config_dn(self, dn, attribute_name):
         status, data = self._post(URLS.CONFIG_READ_DN, {
             "ObjectDN": dn,