Merge pull request #70 from Venafi/vaas_pm_fix

Service generated CSR implementation

Author: rvelaVenafi

examples/get_cert_service_tpp.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+#
+# Copyright 2019 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.
+#
+
+from __future__ import print_function
+
+from vcert import (CertificateRequest, venafi_connection, RevocationRequest, CSR_ORIGIN_SERVICE)
+import string
+import random
+import logging
+import time
+from os import environ
+
+logging.basicConfig(level=logging.INFO)
+logging.getLogger("urllib3").setLevel(logging.ERROR)
+
+
+def main():
+    # Get credentials from environment variables
+    url = environ.get('TPP_TOKEN_URL')
+    user = environ.get('TPP_USER')
+    password = environ.get('TPP_PASSWORD')
+    zone = environ.get("TPP_ZONE")
+    server_trust_bundle = environ.get('TPP_TRUST_BUNDLE')
+
+    # 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"})
+
+    request = CertificateRequest(common_name=random_word(10) + ".venafi.example.com")
+    request.csr_origin = CSR_ORIGIN_SERVICE
+    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).
+    # 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
+    zone_config = conn.read_zone_conf(zone)
+    request.update_from_zone_config(zone_config)
+    conn.request_cert(request, zone)
+
+    # and wait for signing
+    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")
+    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)
+    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")
+    fn.write(new_cert.cert)
+    fn = open("/tmp/new_cert.key", "w")
+    fn.write(new_request.private_key_pem)
+    fn.close()
+
+
+def random_word(length):
+    letters = string.ascii_lowercase
+    return ''.join(random.choice(letters) for i in range(length))
+
+
+if __name__ == '__main__':
+    main()

tests/test_e2e.py

@@ -36,7 +36,7 @@
     RANDOM_DOMAIN, CLOUD_ZONE, \
     TPP_ZONE, TPP_ZONE_ECDSA
 from vcert import CloudConnection, CertificateRequest, TPPConnection, FakeConnection, ZoneConfig, RevocationRequest, \
-    TPPTokenConnection, CertField, KeyType, CustomField
+    TPPTokenConnection, CertField, KeyType, CustomField, CSR_ORIGIN_SERVICE
 from vcert.errors import ClientBadData, ServerUnexptedBehavior
 from vcert.pem import parse_pem
 
@@ -287,6 +287,15 @@ def test_tpp_token_enroll(self):
         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, 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 = [
@@ -496,7 +505,8 @@ def enroll_with_zone_update(conn, zone, cn=None):
     return cert, request.cert_guid
 
 
-def enroll(conn, zone, cn=None, private_key=None, public_key=None, password=None, csr=None, custom_fields=None):
+def enroll(conn, zone, cn=None, private_key=None, public_key=None, password=None, csr=None, custom_fields=None,
+           service_generated_csr=False):
     request = CertificateRequest(
         common_name=cn,
         private_key=private_key,
@@ -515,6 +525,8 @@ def enroll(conn, zone, cn=None, private_key=None, public_key=None, password=None
 
     if csr:
         request.csr = csr
+    elif service_generated_csr:
+        request.csr_origin = CSR_ORIGIN_SERVICE
 
     conn.request_cert(request, zone)
     cert = conn.retrieve_cert(request)
@@ -523,9 +535,10 @@ def enroll(conn, zone, cn=None, private_key=None, public_key=None, password=None
     # and save into file
     f = open("./cert.pem", "w")
     f.write(cert.full_chain)
-    f = open("./cert.key", "w")
-    f.write(request.private_key_pem)
-    f.close()
+    if not service_generated_csr:
+        f = open("./cert.key", "w")
+        f.write(request.private_key_pem)
+        f.close()
 
     cert = x509.load_pem_x509_certificate(cert.cert.encode(), default_backend())
     assert isinstance(cert, x509.Certificate)
@@ -551,11 +564,15 @@ def enroll(conn, zone, cn=None, private_key=None, public_key=None, password=None
             format=serialization.PublicFormat.SubjectPublicKeyInfo
         ).decode()
     else:
-        source_public_key_pem = request.public_key_pem
+        source_public_key_pem = request.public_key_pem if not service_generated_csr else None
     print(source_public_key_pem)
     print(cert_public_key_pem)
-    assert source_public_key_pem == cert_public_key_pem
-    return request.id, request.private_key_pem, cert, cert_public_key_pem, request.cert_guid
+
+    if not service_generated_csr:
+        assert source_public_key_pem == cert_public_key_pem
+    private_key_pem = request.private_key_pem if not service_generated_csr else None
+
+    return request.id, private_key_pem, cert, cert_public_key_pem, request.cert_guid
 
 
 def renew(conn, cert_id, pkey, sn, cn):

vcert/__init__.py

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

vcert/common.py

@@ -46,6 +46,9 @@
 MIME_ANY = "*/*"
 LOCALHOST = "127.0.0.1"
 DEFAULT_TIMEOUT = 180
+CSR_ORIGIN_PROVIDED = "provided"
+CSR_ORIGIN_LOCAL = "local"
+CSR_ORIGIN_SERVICE = "service"
 
 
 class CertField:
@@ -257,7 +260,8 @@ def __init__(self, cert_id=None,
                  locality=None,
                  origin=None,
                  custom_fields=None,
-                 timeout=DEFAULT_TIMEOUT
+                 timeout=DEFAULT_TIMEOUT,
+                 csr_origin=CSR_ORIGIN_LOCAL
                  ):
         """
         :param str cert_id: Certificate request id. Generating by server.
@@ -277,6 +281,7 @@ def __init__(self, cert_id=None,
         :param str origin: application identifier
         :param list[CustomField] custom_fields: list of custom fields values to be added to the certificate.
         :param int timeout: Timeout for the certificate to be retrieved from server. Measured in seconds.
+        :param str csr_origin: The origin of the CSR, either user provided, locally generated or service generated.
         """
 
         self.chain_option = "last"
@@ -302,6 +307,10 @@ def __init__(self, cert_id=None,
         self.locality = locality
         # CSR should be last, because it checks subject to match with over parameters
         self.csr = csr
+        if self.csr:
+            self.csr_origin = CSR_ORIGIN_PROVIDED
+        else:
+            self.csr_origin = csr_origin
         self.origin = origin or "Venafi VCert-Python"
         self.custom_fields = custom_fields
         self.cert_guid = None
@@ -330,6 +339,7 @@ def __setattr__(self, key, value):
             else:
                 raise ClientBadData("invalid private key type %s" % type(value))
         elif key == "csr":
+            self.csr_origin = CSR_ORIGIN_PROVIDED
             if isinstance(value, binary_type):
                 value = value.decode()
             elif not (isinstance(value, string_types) or value is None):
@@ -356,7 +366,7 @@ def __setattr__(self, key, value):
         self.__dict__[key] = value
 
     def _gen_key(self):
-        if self.key_type == None:
+        if self.key_type is None:
             self.key_type = KeyType(KeyType.RSA, 2048)
         if self.key_type.key_type == KeyType.RSA:
             self.private_key = rsa.generate_private_key(
@@ -439,7 +449,7 @@ def build_csr(self):
                     # string we'll convert it into a bytes object then insert our header. Otherwise, we'll just
                     # insert the header in the passed in bytes.
                     if isinstance(upn, str):
-                        bupn = bytes(upn,'utf-8')
+                        bupn = bytes(upn, 'utf-8')
                     else:
                         bupn = upn
                     values = [12,len(upn)]

vcert/connection_tpp.py

@@ -27,7 +27,8 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.x509 import SignatureAlgorithmOID as AlgOID
 
-from .common import MIME_JSON, CertField, ZoneConfig, Policy, KeyType
+from .common import MIME_JSON, CertField, ZoneConfig, Policy, KeyType, CSR_ORIGIN_LOCAL, CSR_ORIGIN_SERVICE, \
+    CSR_ORIGIN_PROVIDED
 from .connection_tpp_abstract import AbstractTPPConnection, URLS
 from .errors import (ServerUnexptedBehavior, ClientBadData, CertificateRequestError, AuthenticationError,
                      CertificateRenewError, VenafiError, RetrieveCertificateTimeoutError)
@@ -114,51 +115,6 @@ def auth(self):
             log.error("Authentication status is not %s but %s. Exiting" % (HTTPStatus.OK, status[0]))
             raise AuthenticationError
 
-    # TODO: Need to add service generated CSR implementation
-    def request_cert(self, request, zone):
-        if not request.csr:
-            request.build_csr()
-        request_data = {"PolicyDN": self._get_policy_dn(zone),
-                        "PKCS10": request.csr,
-                        "ObjectName": request.friendly_name,
-                        "DisableAutomaticRenewal": "true"}
-        if request.origin:
-            request_data["Origin"] = request.origin
-            ca_origin = {"Name": "Origin", "Value": request.origin}
-            if request_data.get("CASpecificAttributes"):
-                request_data["CASpecificAttributes"].append(ca_origin)
-            else:
-                request_data["CASpecificAttributes"] = [ca_origin]
-
-        if request.custom_fields:
-            custom_fields_map = {}
-            for c_field in request.custom_fields:
-                if custom_fields_map.get(c_field.name):
-                    custom_fields_map[c_field.name].append(c_field.value)
-                else:
-                    custom_fields_map[c_field.name] = [c_field.value]
-
-            for key in custom_fields_map:
-                custom_field_json = {
-                    "Name": key,
-                    "Values": custom_fields_map[key]
-                }
-                if request_data.get("CustomFields"):
-                    request_data["CustomFields"].append(custom_field_json)
-                else:
-                    request_data["CustomFields"] = [custom_field_json]
-
-        status, data = self._post(URLS.CERTIFICATE_REQUESTS, data=request_data)
-        if status == HTTPStatus.OK:
-            request.id = data['CertificateDN']
-            request.cert_guid = data['Guid']
-            log.debug("Certificate successfully requested with request id %s." % request.id)
-            log.debug("Certificate successfully requested with GUID %s." % request.cert_guid)
-            return True
-
-        log.error("Request status is not %s. %s." % HTTPStatus.OK, status)
-        raise CertificateRequestError
-
     def retrieve_cert(self, cert_request):
         log.debug("Getting certificate status for id %s" % cert_request.id)
 
@@ -336,19 +292,6 @@ def read_zone_conf(self, tag):
     def import_cert(self, request):
         raise NotImplementedError
 
-    @staticmethod
-    def _get_policy_dn(zone):
-        if zone is None:
-            log.error("Bad zone: %s" % zone)
-            raise ClientBadData
-        if re.match(r"^\\\\VED\\\\Policy", zone):
-            return zone
-        else:
-            if re.match(r"^\\\\", zone):
-                return r"\\VED\\Policy" + zone
-            else:
-                return r"\\VED\\Policy\\" + zone
-
     def search_by_thumbprint(self, thumbprint):
         """
         :param str thumbprint: