Merge pull request #88 from Venafi/svc_gen_csr_fixes

Added step to fill empty request attributes with zone default values

Author: rvelaVenafi

vcert/common.py

@@ -183,7 +183,7 @@ def __init__(self, policy_id=None, company_id=None, name=None, system_generated=
         :param str status:
         :param str reason:
         :param str validity_period:
-        :param RecommendedSettings recommended_settings:
+        :param vaas_utils.RecommendedSettings recommended_settings:
         """
         self.id = policy_id
         self.company_id = company_id
@@ -264,11 +264,11 @@ def __init__(self, cert_id=None,
         :param str friendly_name: Name for certificate in the platform. If not specified common name will be used.
         :param str common_name: Common name of certificate. Usually domain name.
         :param str thumbprint: Certificate thumbprint. Can be used for identifying certificate on the platform.
-        :param organization:
-        :param organizational_unit:
-        :param country:
-        :param province:
-        :param locality:
+        :param str organization:
+        :param str or list[str] organizational_unit:
+        :param str country:
+        :param str province:
+        :param str locality:
         :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.
@@ -496,7 +496,6 @@ def public_key_pem(self):
         ).decode()
 
     def update_from_zone_config(self, zone):
-        # TODO: check request against zone configuration
         """
         :param ZoneConfig zone:
         """
@@ -588,6 +587,8 @@ def __init__(self, user=None, password=None, access_token=None, refresh_token=No
 
 
 class CommonConnection:
+    def __init__(self):
+        pass
 
     def auth(self):
         """

vcert/connection_cloud.py

@@ -25,6 +25,7 @@
 import requests
 import six.moves.urllib.parse as urlparse
 from nacl.public import SealedBox
+from six import string_types
 
 from .common import (ZoneConfig, CertificateRequest, CommonConnection, Policy, get_ip_address, log_errors, MIME_JSON,
                      MIME_TEXT, MIME_ANY, CertField, KeyType, DEFAULT_TIMEOUT,
@@ -357,6 +358,9 @@ def request_cert(self, request, zone):
                 'identifier': ip_address
             }
         }
+        zone_config = self.read_zone_conf(zone)
+        request.update_from_zone_config(zone_config)
+
         if request.csr_origin != CSR_ORIGIN_SERVICE:
             if not request.csr:
                 request.build_csr()
@@ -540,12 +544,25 @@ def search_by_thumbprint(self, thumbprint, timeout=DEFAULT_TIMEOUT):
 
     def read_zone_conf(self, zone):
         policy = self._get_template_by_id(zone)
+        rs = policy.recommended_settings
+        org = CertField("")
+        org_unit = CertField("")
+        locality = CertField("")
+        state = CertField("")
+        country = CertField("")
+        if rs:
+            org = CertField(rs.subjectOValue)
+            org_unit = CertField(rs.subjectOUValue)
+            locality = CertField(rs.subjectLValue)
+            state = CertField(rs.subjectSTValue)
+            country = CertField(rs.subjectCValue)
+
         z = ZoneConfig(
-            organization=CertField(""),
-            organizational_unit=CertField(""),
-            country=CertField(""),
-            province=CertField(""),
-            locality=CertField(""),
+            organization=org,
+            organizational_unit=org_unit,
+            country=country,
+            province=state,
+            locality=locality,
             policy=policy,
             key_type=policy.key_types[0] if policy.key_types else None,
         )
@@ -736,9 +753,16 @@ def _get_service_generated_csr_attr(self, request, zone):
             csr_attr_map[CSR_ATTR_ORG] = ps.defaults.subject.org
 
         if request.organizational_unit:
+            if isinstance(request.organizational_unit, string_types):
+                org_units = [request.organizational_unit]
+            else:
+                org_units = request.organizational_unit
+
             if ps.policy and ps.policy.subject:
                 policy_ous = ps.policy.subject.org_units
-                valid = value_matches_regex(value=request.organizational_unit, pattern_list=policy_ous)
+                valid = all(
+                    value_matches_regex(value=ou, pattern_list=policy_ous) for ou in org_units
+                )
                 if not valid:
                     ou_str = "Organizational Unit"
                     log.error(MSG_VALUE_NOT_MATCH_POLICY % (ou_str, ou_str+"s", request.organizational_unit,

vcert/connection_tpp_abstract.py

@@ -79,12 +79,26 @@ def __init__(self):
 
 
 class AbstractTPPConnection(CommonConnection):
+    def __init__(self):
+        CommonConnection.__init__(self)
+
+    ARG_URL = 'url'
+    ARG_PARAMS = 'params'
+    ARG_CHECK_TOKEN = 'check_token'  # nosec
+    ARG_INCLUDE_TOKEN_HEADER = 'include_token_header'  # nosec
+    ARG_DATA = 'data'
+
+    def auth(self):
+        raise NotImplementedError
 
     def request_cert(self, request, zone):
-        request_data = {'PolicyDN': self._normalize_zone(zone),
-                        'ObjectName': request.friendly_name,
-                        'DisableAutomaticRenewal': "true"
-                        }
+        request_data = {
+            'PolicyDN': self._normalize_zone(zone),
+            'ObjectName': request.friendly_name,
+            'DisableAutomaticRenewal': "true"
+        }
+        zone_config = self.read_zone_conf(zone)
+        request.update_from_zone_config(zone_config)
 
         if request.csr_origin == CSR_ORIGIN_LOCAL:
             request.build_csr()
@@ -143,8 +157,11 @@ def request_cert(self, request, zone):
                 else:
                     request_data['CustomFields'] = [custom_field_json]
 
-
-        status, data = self._post(URLS.CERTIFICATE_REQUESTS, data=request_data)
+        args = {
+            self.ARG_URL: URLS.CERTIFICATE_REQUESTS,
+            self.ARG_DATA: request_data
+        }
+        status, data = self.post(args)
         if status == HTTPStatus.OK:
             request.id = data['CertificateDN']
             request.cert_guid = data['Guid']
@@ -321,6 +338,18 @@ def revoke_cert(self, request):
     def import_cert(self, request):
         raise NotImplementedError
 
+    def read_zone_conf(self, tag):
+        args = {
+            self.ARG_URL: URLS.ZONE_CONFIG,
+            self.ARG_DATA: {
+                'PolicyDN': self._normalize_zone(tag)
+            }
+        }
+        status, data = self.post(args=args)
+        if status != HTTPStatus.OK:
+            raise ServerUnexptedBehavior("Server returns %d status on reading zone configuration." % status)
+        return self._parse_zone_data_to_object(data)
+
     def get_policy(self, zone):
         # get policy spec from name
         policy_name = self._normalize_zone(zone)
@@ -630,12 +659,6 @@ def retrieve_ssh_config(self, ca_request):
             raise ServerUnexptedBehavior("Server returns %d status on requesting SSH CA Public Key Data for %s = %s."
                                          % (status, key, value))
 
-    ARG_URL = 'url'
-    ARG_PARAMS = 'params'
-    ARG_CHECK_TOKEN = 'check_token'  # nosec
-    ARG_INCLUDE_TOKEN_HEADER = 'include_token_header'  # nosec
-    ARG_DATA = 'data'
-
     def get(self, args):
         """
 
@@ -725,22 +748,13 @@ def _reset_policy_attr(self, zone, attr_name):
         return status, response
 
     def _reset_policy(self, zone):
-        self._reset_policy_attr(zone, SPA.TPP_DOMAIN_SUFFIX_WHITELIST)
-        self._reset_policy_attr(zone, SPA.TPP_PROHIBIT_WILDCARD)
-        self._reset_policy_attr(zone, SPA.TPP_CERT_AUTHORITY)
-        self._reset_policy_attr(zone, SPA.TPP_ORGANIZATION)
-        self._reset_policy_attr(zone, SPA.TPP_ORG_UNIT)
-        self._reset_policy_attr(zone, SPA.TPP_CITY)
-        self._reset_policy_attr(zone, SPA.TPP_STATE)
-        self._reset_policy_attr(zone, SPA.TPP_COUNTRY)
-        self._reset_policy_attr(zone, SPA.TPP_KEY_ALGORITHM)
-        self._reset_policy_attr(zone, SPA.TPP_KEY_BIT_STR)
-        self._reset_policy_attr(zone, SPA.TPP_ELLIPTIC_CURVE)
-        self._reset_policy_attr(zone, SPA.TPP_MANUAL_CSR)
-        self._reset_policy_attr(zone, SPA.TPP_PROHIBITED_SAN_TYPES)
-        self._reset_policy_attr(zone, SPA.TPP_ALLOWED_PRIVATE_KEY_REUSE)
-        self._reset_policy_attr(zone, SPA.TPP_WANT_RENEWAL)
-        self._reset_policy_attr(zone, SPA.TPP_MANAGEMENT_TYPE)
+        atrr_list = [SPA.TPP_DOMAIN_SUFFIX_WHITELIST, SPA.TPP_PROHIBIT_WILDCARD, SPA.TPP_CERT_AUTHORITY,
+                     SPA.TPP_ORGANIZATION, SPA.TPP_ORG_UNIT, SPA.TPP_CITY, SPA.TPP_STATE, SPA.TPP_COUNTRY,
+                     SPA.TPP_KEY_ALGORITHM, SPA.TPP_KEY_BIT_STR, SPA.TPP_ELLIPTIC_CURVE, SPA.TPP_MANUAL_CSR,
+                     SPA.TPP_PROHIBITED_SAN_TYPES, SPA.TPP_ALLOWED_PRIVATE_KEY_REUSE, SPA.TPP_WANT_RENEWAL,
+                     SPA.TPP_MANAGEMENT_TYPE]
+        for attr in atrr_list:
+            self._reset_policy_attr(zone, attr)
 
     @staticmethod
     def _parse_attr_response(response):
@@ -873,6 +887,7 @@ def _parse_zone_data_to_object(data):
             key_type = KeyType(KeyType.ECDSA, data['Policy']['KeyPair']['EllipticCurve']['Value'])
         else:
             key_type = None
+
         z = ZoneConfig(
             organization=CertField(s['Organization']['Value'], locked=s['Organization']['Locked']),
             organizational_unit=CertField(ou, locked=s['OrganizationalUnit']['Locked']),
@@ -884,12 +899,6 @@ def _parse_zone_data_to_object(data):
         )
         return z
 
-    def read_zone_conf(self, 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 _get_certificate_details(self, cert_guid):
         status, data = self._get(URLS.CERTIFICATE_SEARCH + cert_guid)
         if status != HTTPStatus.OK:

vcert/connection_tpp_token.py

@@ -24,16 +24,16 @@
 
 import requests
 
-from .common import MIME_JSON, TokenInfo, Authentication, KeyType, Policy, ZoneConfig, CertField
+from .common import MIME_JSON, TokenInfo, Authentication
 from .connection_tpp_abstract import AbstractTPPConnection, URLS
 from .errors import (ClientBadData, ServerUnexptedBehavior, AuthenticationError)
 from .http import HTTPStatus
 
-HEADER_AUTHORIZATION = "Authorization"  # type: str
+HEADER_AUTHORIZATION = 'Authorization'
 
-KEY_ACCESS_TOKEN = "access_token"  # type: str  # nosec
-KEY_REFRESH_TOKEN = "refresh_token"  # type: str  # nosec
-KEY_EXPIRATION_DATE = "expiration_date"  # type: str
+KEY_ACCESS_TOKEN = 'access_token'  # nosec
+KEY_REFRESH_TOKEN = 'refresh_token'  # nosec
+KEY_EXPIRATION_DATE = 'expiration_date'
 
 
 class TPPTokenConnection(AbstractTPPConnection):
@@ -46,6 +46,8 @@ def __init__(self, url, user=None, password=None, access_token=None, refresh_tok
         :param str refresh_token:
         :param dict[str,Any] http_request_kwargs:
         """
+        AbstractTPPConnection.__init__(self)
+
         self._base_url = url  # type: str
         self._auth = Authentication(user=user, password=password, access_token=access_token,
                                     refresh_token=refresh_token)  # type: Authentication

vcert/policy/__init__.py

@@ -33,7 +33,7 @@ def __init__(self):
 # tpp read policy attributes - RPA
 class RPA(CommonPA):
     def __init__(self):
-        pass
+        CommonPA.__init__(self)
 
     TPP_CERT_AUTHORITY = "CertificateAuthority"  # "Certificate Authority"
     TPP_WILDCARDS_ALLOWED = "WildcardsAllowed"
@@ -57,7 +57,7 @@ def __init__(self):
 # tpp set policy attributes - SPA
 class SPA(CommonPA):
     def __init__(self):
-        pass
+        CommonPA.__init__(self)
 
     TPP_CERT_AUTHORITY = "Certificate Authority"
     TPP_PROHIBIT_WILDCARD = "Prohibit Wildcard"

vcert/vaas_utils.py

@@ -92,7 +92,6 @@ def zip_to_pem(data, chain_option):
     """
     zip_data = zipfile.ZipFile(io.BytesIO(data))
     private_key = None
-    certs = []
     chain = []
     certificate = None
     for info in zip_data.infolist():
@@ -125,9 +124,3 @@ def value_matches_regex(value, pattern_list):
     :rtype: bool
     """
     return any((re.match(pattern, value) is not None) for pattern in pattern_list)
-    # for regex in regex_list:
-    #     result = re.match(regex, value)
-    #     if not result:
-    #         return False
-    #
-    # return True