From 9c355e5c52c77c0d68a1911e410665798b59e55d Mon Sep 17 00:00:00 2001 From: Felix Fontein Date: Mon, 25 Mar 2019 14:20:05 +0100 Subject: [PATCH] openssl_certificate, openssl_csr: refactoring / cleanup (#54287) * Moving common cryptography-related code to module_utils/crypto.py. * Fix typo / linting. --- lib/ansible/module_utils/crypto.py | 191 +++++++++++++++++- .../modules/crypto/openssl_certificate.py | 165 +-------------- lib/ansible/modules/crypto/openssl_csr.py | 172 +--------------- 3 files changed, 205 insertions(+), 323 deletions(-) diff --git a/lib/ansible/module_utils/crypto.py b/lib/ansible/module_utils/crypto.py index f698226ed28..c541e2cc970 100644 --- a/lib/ansible/module_utils/crypto.py +++ b/lib/ansible/module_utils/crypto.py @@ -28,6 +28,7 @@ try: from cryptography.hazmat.backends import default_backend as cryptography_backend from cryptography.hazmat.primitives.serialization import load_pem_private_key from cryptography.hazmat.primitives import hashes + import ipaddress except ImportError: # Error handled in the calling module. pass @@ -41,7 +42,7 @@ import os import re from ansible.module_utils import six -from ansible.module_utils._text import to_bytes +from ansible.module_utils._text import to_bytes, to_text class OpenSSLObjectError(Exception): @@ -128,7 +129,7 @@ def load_privatekey(path, passphrase=None, check_passphrase=True, backend='pyope # Since we can load the key without an exception, the # key isn't password-protected raise OpenSSLBadPassphraseError('Passphrase provided, but private key is not password-protected!') - except crypto.Error: + except crypto.Error as e: if passphrase is None and len(e.args) > 0 and len(e.args[0]) > 0 and e.args[0][0][2] == 'bad decrypt': # The key is obviously protected by the empty string. # Don't do this at home (if it's possible at all)... @@ -138,9 +139,9 @@ def load_privatekey(path, passphrase=None, check_passphrase=True, backend='pyope result = load_pem_private_key(priv_key_detail, passphrase, cryptography_backend()) - except TypeError as e: + except TypeError as dummy: raise OpenSSLBadPassphraseError('Wrong or empty passphrase provided for private key') - except ValueError as e: + except ValueError as dummy: raise OpenSSLBadPassphraseError('Wrong passphrase provided for private key') return result @@ -283,3 +284,185 @@ class OpenSSLObject(object): raise OpenSSLObjectError(exc) else: pass + + +def cryptography_get_name_oid(id): + ''' + Given a symbolic ID, finds the appropriate OID for use with cryptography. + Raises an OpenSSLObjectError if the ID is unknown. + ''' + if id in ('CN', 'commonName'): + return x509.oid.NameOID.COMMON_NAME + if id in ('C', 'countryName'): + return x509.oid.NameOID.COUNTRY_NAME + if id in ('L', 'localityName'): + return x509.oid.NameOID.LOCALITY_NAME + if id in ('ST', 'stateOrProvinceName'): + return x509.oid.NameOID.STATE_OR_PROVINCE_NAME + if id in ('street', 'streetAddress'): + return x509.oid.NameOID.STREET_ADDRESS + if id in ('O', 'organizationName'): + return x509.oid.NameOID.ORGANIZATION_NAME + if id in ('OU', 'organizationalUnitName'): + return x509.oid.NameOID.ORGANIZATIONAL_UNIT_NAME + if id in ('serialNumber', ): + return x509.oid.NameOID.SERIAL_NUMBER + if id in ('SN', 'surname'): + return x509.oid.NameOID.SURNAME + if id in ('GN', 'givenName'): + return x509.oid.NameOID.GIVEN_NAME + if id in ('title', ): + return x509.oid.NameOID.TITLE + if id in ('generationQualifier', ): + return x509.oid.NameOID.GENERATION_QUALIFIER + if id in ('x500UniqueIdentifier', ): + return x509.oid.NameOID.X500_UNIQUE_IDENTIFIER + if id in ('dnQualifier', ): + return x509.oid.NameOID.DN_QUALIFIER + if id in ('pseudonym', ): + return x509.oid.NameOID.PSEUDONYM + if id in ('UID', 'userId'): + return x509.oid.NameOID.USER_ID + if id in ('DC', 'domainComponent'): + return x509.oid.NameOID.DOMAIN_COMPONENT + if id in ('emailAddress', ): + return x509.oid.NameOID.EMAIL_ADDRESS + if id in ('jurisdictionC', 'jurisdictionCountryName'): + return x509.oid.NameOID.JURISDICTION_COUNTRY_NAME + if id in ('jurisdictionL', 'jurisdictionLocalityName'): + return x509.oid.NameOID.JURISDICTION_LOCALITY_NAME + if id in ('jurisdictionST', 'jurisdictionStateOrProvinceName'): + return x509.oid.NameOID.JURISDICTION_STATE_OR_PROVINCE_NAME + if id in ('businessCategory', ): + return x509.oid.NameOID.BUSINESS_CATEGORY + if id in ('postalAddress', ): + return x509.oid.NameOID.POSTAL_ADDRESS + if id in ('postalCode', ): + return x509.oid.NameOID.POSTAL_CODE + raise OpenSSLObjectError('Unknown subject field identifier "{0}"'.format(id)) + + +def cryptography_get_name(name): + ''' + Given a name string, returns a cryptography x509.Name object. + Raises an OpenSSLObjectError if the name is unknown or cannot be parsed. + ''' + try: + if name.startswith('DNS:'): + return x509.DNSName(to_text(name[4:])) + if name.startswith('IP:'): + return x509.IPAddress(ipaddress.ip_address(to_text(name[3:]))) + if name.startswith('email:'): + return x509.RFC822Name(to_text(name[6:])) + if name.startswith('URI:'): + return x509.UniformResourceIdentifier(to_text(name[4:])) + except Exception as e: + raise OpenSSLObjectError('Cannot parse Subject Alternative Name "{0}": {1}'.format(name, e)) + if ':' not in name: + raise OpenSSLObjectError('Cannot parse Subject Alternative Name "{0}" (forgot "DNS:" prefix?)'.format(name)) + raise OpenSSLObjectError('Cannot parse Subject Alternative Name "{0}" (potentially unsupported by cryptography backend)'.format(name)) + + +def _cryptography_get_keyusage(usage): + ''' + Given a key usage identifier string, returns the parameter name used by cryptography's x509.KeyUsage(). + Raises an OpenSSLObjectError if the identifier is unknown. + ''' + if usage in ('Digital Signature', 'digitalSignature'): + return 'digital_signature' + if usage in ('Non Repudiation', 'nonRepudiation'): + return 'content_commitment' + if usage in ('Key Encipherment', 'keyEncipherment'): + return 'key_encipherment' + if usage in ('Data Encipherment', 'dataEncipherment'): + return 'data_encipherment' + if usage in ('Key Agreement', 'keyAgreement'): + return 'key_agreement' + if usage in ('Certificate Sign', 'keyCertSign'): + return 'key_cert_sign' + if usage in ('CRL Sign', 'cRLSign'): + return 'crl_sign' + if usage in ('Encipher Only', 'encipherOnly'): + return 'encipher_only' + if usage in ('Decipher Only', 'decipherOnly'): + return 'decipher_only' + raise OpenSSLObjectError('Unknown key usage "{0}"'.format(usage)) + + +def cryptography_parse_key_usage_params(usages): + ''' + Given a list of key usage identifier strings, returns the parameters for cryptography's x509.KeyUsage(). + Raises an OpenSSLObjectError if an identifier is unknown. + ''' + params = dict( + digital_signature=False, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=False, + crl_sign=False, + encipher_only=False, + decipher_only=False, + ) + for usage in usages: + params[_cryptography_get_keyusage(usage)] = True + return params + + +def cryptography_get_ext_keyusage(usage): + ''' + Given an extended key usage identifier string, returns the OID used by cryptography. + Raises an OpenSSLObjectError if the identifier is unknown. + ''' + if usage in ('serverAuth', 'TLS Web Server Authentication'): + return x509.oid.ExtendedKeyUsageOID.SERVER_AUTH + if usage in ('clientAuth', 'TLS Web Client Authentication'): + return x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH + if usage in ('codeSigning', 'Code Signing'): + return x509.oid.ExtendedKeyUsageOID.CODE_SIGNING + if usage in ('emailProtection', 'E-mail Protection'): + return x509.oid.ExtendedKeyUsageOID.EMAIL_PROTECTION + if usage in ('timeStamping', 'Time Stamping'): + return x509.oid.ExtendedKeyUsageOID.TIME_STAMPING + if usage in ('OCSPSigning', 'OCSP Signing'): + return x509.oid.ExtendedKeyUsageOID.OCSP_SIGNING + if usage in ('anyExtendedKeyUsage', 'Any Extended Key Usage'): + return x509.oid.ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE + if usage in ('qcStatements', ): + return x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.3") + if usage in ('DVCS', ): + return x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.3.10") + if usage in ('IPSec User', 'ipsecUser'): + return x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.3.7") + if usage in ('Biometric Info', 'biometricInfo'): + return x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.2") + # FIXME need some more, probably all from https://www.iana.org/assignments/smi-numbers/smi-numbers.xhtml#smi-numbers-1.3.6.1.5.5.7.3 + raise OpenSSLObjectError('Unknown extended key usage "{0}"'.format(usage)) + + +def cryptography_get_basic_constraints(constraints): + ''' + Given a list of constraints, returns a tuple (ca, path_length). + Raises an OpenSSLObjectError if a constraint is unknown or cannot be parsed. + ''' + ca = False + path_length = None + if constraints: + for constraint in constraints: + if constraint.startswith('CA:'): + if constraint == 'CA:TRUE': + ca = True + elif constraint == 'CA:FALSE': + ca = False + else: + raise OpenSSLObjectError('Unknown basic constraint value "{0}" for CA'.format(constraint[3:])) + elif constraint.startswith('pathlen:'): + v = constraint[len('pathlen:'):] + try: + path_length = int(v) + except Exception as e: + raise OpenSSLObjectError('Cannot parse path length constraint "{0}" ({1})'.format(v, e)) + else: + raise OpenSSLObjectError('Unknown basic constraint "{0}"'.format(constraint)) + return ca, path_length diff --git a/lib/ansible/modules/crypto/openssl_certificate.py b/lib/ansible/modules/crypto/openssl_certificate.py index aa84bc1716e..916d1596f0c 100644 --- a/lib/ansible/modules/crypto/openssl_certificate.py +++ b/lib/ansible/modules/crypto/openssl_certificate.py @@ -659,7 +659,7 @@ class Certificate(crypto_utils.OpenSSLObject): csr_ext = csr_exts.get_extension_for_oid(cert_ext.oid) if cert_ext != csr_ext: return False - except cryptography.x509.ExtensionNotFound as e: + except cryptography.x509.ExtensionNotFound as dummy: return False return True @@ -1097,156 +1097,6 @@ class AssertOnlyCertificateCryptography(Certificate): self.valid_in = module.params['valid_in'], self.message = [] - def _get_name_oid(self, id): - if id in ('CN', 'commonName'): - return cryptography.x509.oid.NameOID.COMMON_NAME - if id in ('C', 'countryName'): - return cryptography.x509.oid.NameOID.COUNTRY_NAME - if id in ('L', 'localityName'): - return cryptography.x509.oid.NameOID.LOCALITY_NAME - if id in ('ST', 'stateOrProvinceName'): - return cryptography.x509.oid.NameOID.STATE_OR_PROVINCE_NAME - if id in ('street', 'streetAddress'): - return cryptography.x509.oid.NameOID.STREET_ADDRESS - if id in ('O', 'organizationName'): - return cryptography.x509.oid.NameOID.ORGANIZATION_NAME - if id in ('OU', 'organizationalUnitName'): - return cryptography.x509.oid.NameOID.ORGANIZATIONAL_UNIT_NAME - if id in ('serialNumber', ): - return cryptography.x509.oid.NameOID.SERIAL_NUMBER - if id in ('SN', 'surname'): - return cryptography.x509.oid.NameOID.SURNAME - if id in ('GN', 'givenName'): - return cryptography.x509.oid.NameOID.GIVEN_NAME - if id in ('title', ): - return cryptography.x509.oid.NameOID.TITLE - if id in ('generationQualifier', ): - return cryptography.x509.oid.NameOID.GENERATION_QUALIFIER - if id in ('x500UniqueIdentifier', ): - return cryptography.x509.oid.NameOID.X500_UNIQUE_IDENTIFIER - if id in ('dnQualifier', ): - return cryptography.x509.oid.NameOID.DN_QUALIFIER - if id in ('pseudonym', ): - return cryptography.x509.oid.NameOID.PSEUDONYM - if id in ('UID', 'userId'): - return cryptography.x509.oid.NameOID.USER_ID - if id in ('DC', 'domainComponent'): - return cryptography.x509.oid.NameOID.DOMAIN_COMPONENT - if id in ('emailAddress', ): - return cryptography.x509.oid.NameOID.EMAIL_ADDRESS - if id in ('jurisdictionC', 'jurisdictionCountryName'): - return cryptography.x509.oid.NameOID.JURISDICTION_COUNTRY_NAME - if id in ('jurisdictionL', 'jurisdictionLocalityName'): - return cryptography.x509.oid.NameOID.JURISDICTION_LOCALITY_NAME - if id in ('jurisdictionST', 'jurisdictionStateOrProvinceName'): - return cryptography.x509.oid.NameOID.JURISDICTION_STATE_OR_PROVINCE_NAME - if id in ('businessCategory', ): - return cryptography.x509.oid.NameOID.BUSINESS_CATEGORY - if id in ('postalAddress', ): - return cryptography.x509.oid.NameOID.POSTAL_ADDRESS - if id in ('postalCode', ): - return cryptography.x509.oid.NameOID.POSTAL_CODE - - def _get_san(self, name): - if name.startswith('DNS:'): - return cryptography.x509.DNSName(to_native(name[4:])) - if name.startswith('IP:'): - return cryptography.x509.IPAddress(to_native(name[3:])) - if name.startswith('email:'): - return cryptography.x509.RFC822Name(to_native(name[6:])) - if name.startswith('URI:'): - return cryptography.x509.UniformResourceIdentifier(to_native(name[4:])) - if name.startswith('DirName:'): - return cryptography.x509.DirectoryName(to_native(name[8:])) - if ':' not in name: - raise CertificateError('Cannot parse Subject Alternative Name "{0}" (forgot "DNS:" prefix?)'.format(name)) - raise CertificateError('Cannot parse Subject Alternative Name "{0}" (potentially unsupported by cryptography backend)'.format(name)) - - def _get_keyusage(self, usage): - if usage in ('Digital Signature', 'digitalSignature'): - return 'digital_signature' - if usage in ('Non Repudiation', 'nonRepudiation'): - return 'content_commitment' - if usage in ('Key Encipherment', 'keyEncipherment'): - return 'key_encipherment' - if usage in ('Data Encipherment', 'dataEncipherment'): - return 'data_encipherment' - if usage in ('Key Agreement', 'keyAgreement'): - return 'key_agreement' - if usage in ('Certificate Sign', 'keyCertSign'): - return 'key_cert_sign' - if usage in ('CRL Sign', 'cRLSign'): - return 'crl_sign' - if usage in ('Encipher Only', 'encipherOnly'): - return 'encipher_only' - if usage in ('Decipher Only', 'decipherOnly'): - return 'decipher_only' - raise CertificateError('Unknown key usage "{0}"'.format(usage)) - - def _get_ext_keyusage(self, usage): - if usage in ('serverAuth', 'TLS Web Server Authentication'): - return cryptography.x509.oid.ExtendedKeyUsageOID.SERVER_AUTH - if usage in ('clientAuth', 'TLS Web Client Authentication'): - return cryptography.x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH - if usage in ('codeSigning', 'Code Signing'): - return cryptography.x509.oid.ExtendedKeyUsageOID.CODE_SIGNING - if usage in ('emailProtection', 'E-mail Protection'): - return cryptography.x509.oid.ExtendedKeyUsageOID.EMAIL_PROTECTION - if usage in ('timeStamping', 'Time Stamping'): - return cryptography.x509.oid.ExtendedKeyUsageOID.TIME_STAMPING - if usage in ('OCSPSigning', 'OCSP Signing'): - return cryptography.x509.oid.ExtendedKeyUsageOID.OCSP_SIGNING - if usage in ('anyExtendedKeyUsage', 'Any Extended Key Usage'): - return cryptography.x509.oid.ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE - if usage in ('qcStatements', ): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.3") - if usage in ('DVCS', ): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.3.10") - if usage in ('IPSec User', 'ipsecUser'): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.3.7") - if usage in ('Biometric Info', 'biometricInfo'): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.2") - # FIXME need some more, probably all from https://www.iana.org/assignments/smi-numbers/smi-numbers.xhtml#smi-numbers-1.3.6.1.5.5.7.3 - raise CertificateError('Unknown extended key usage "{0}"'.format(usage)) - - def _get_basic_constraints(self, constraints): - ca = False - path_length = None - if constraints: - for constraint in constraints: - if constraint.startswith('CA:'): - if constraint == 'CA:TRUE': - ca = True - elif constraint == 'CA:FALSE': - ca = False - else: - raise CertificateError('Unknown basic constraint value "{0}" for CA'.format(constraint[3:])) - elif constraint.startswith('pathlen:'): - v = constraint[len('pathlen:'):] - try: - path_length = int(v) - except Exception as e: - raise CertificateError('Cannot parse path length constraint "{0}" ({1})'.format(v, e)) - else: - raise CertificateError('Unknown basic constraint "{0}"'.format(constraint)) - return ca, path_length - - def _parse_key_usage(self): - params = dict( - digital_signature=False, - content_commitment=False, - key_encipherment=False, - data_encipherment=False, - key_agreement=False, - key_cert_sign=False, - crl_sign=False, - encipher_only=False, - decipher_only=False, - ) - for usage in self.keyUsage: - params[self._get_keyusage(usage)] = True - return params - def assertonly(self): self.cert = crypto_utils.load_certificate(self.path, backend=self.backend) @@ -1260,7 +1110,7 @@ class AssertOnlyCertificateCryptography(Certificate): def _validate_subject(): if self.subject: - expected_subject = Name([NameAttribute(oid=self._get_name_oid(sub[0]), value=to_text(sub[1])) + expected_subject = Name([NameAttribute(oid=crypto_utils.cryptography_get_name_oid(sub[0]), value=to_text(sub[1])) for sub in self.subject]) cert_subject = self.cert.subject if (not self.subject_strict and not all(x in cert_subject for x in expected_subject)) or \ @@ -1272,7 +1122,7 @@ class AssertOnlyCertificateCryptography(Certificate): def _validate_issuer(): if self.issuer: - expected_issuer = Name([NameAttribute(oid=self._get_name_oid(iss[0]), value=to_text(iss[1])) + expected_issuer = Name([NameAttribute(oid=crypto_utils.cryptography_get_name_oid(iss[0]), value=to_text(iss[1])) for iss in self.issuer]) cert_issuer = self.cert.issuer if (not self.issuer_strict and not all(x in cert_issuer for x in expected_issuer)) or \ @@ -1303,7 +1153,7 @@ class AssertOnlyCertificateCryptography(Certificate): if self.keyUsage: try: current_keyusage = self.cert.extensions.get_extension_for_class(x509.KeyUsage).value - expected_keyusage = x509.KeyUsage(**self._parse_key_usage()) + expected_keyusage = x509.KeyUsage(**crypto_utils.cryptography_parse_key_usage_params(self.keyUsage)) test_keyusage = dict( digital_signature=current_keyusage.digital_signature, content_commitment=current_keyusage.content_commitment, @@ -1324,7 +1174,8 @@ class AssertOnlyCertificateCryptography(Certificate): decipher_only=False )) - if (not self.keyUsage_strict and not all(self._parse_key_usage()[x] == test_keyusage[x] for x in self._parse_key_usage())) or \ + key_usages = crypto_utils.cryptography_parse_key_usage_params(self.keyUsage) + if (not self.keyUsage_strict and not all(key_usages[x] == test_keyusage[x] for x in key_usages)) or \ (self.keyUsage_strict and current_keyusage != expected_keyusage): self.message.append( 'Invalid keyUsage components (got %s, expected all of %s to be present)' % @@ -1338,7 +1189,7 @@ class AssertOnlyCertificateCryptography(Certificate): if self.extendedKeyUsage: try: current_ext_keyusage = self.cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value - usages = [self._get_ext_keyusage(usage) for usage in self.extendedKeyUsage] + usages = [crypto_utils.cryptography_get_ext_keyusage(usage) for usage in self.extendedKeyUsage] expected_ext_keyusage = x509.ExtendedKeyUsage(usages) if (not self.extendedKeyUsage_strict and not all(x in expected_ext_keyusage for x in current_ext_keyusage)) or \ (self.extendedKeyUsage_strict and not current_ext_keyusage == expected_ext_keyusage): @@ -1354,7 +1205,7 @@ class AssertOnlyCertificateCryptography(Certificate): if self.subjectAltName: try: current_san = self.cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value - expected_san = [self._get_san(san) for san in self.subjectAltName] + expected_san = [crypto_utils.cryptography_get_name(san) for san in self.subjectAltName] if (not self.subjectAltName_strict and not all(x in current_san for x in expected_san)) or \ (self.subjectAltName_strict and not set(current_san) == set(expected_san)): self.message.append( diff --git a/lib/ansible/modules/crypto/openssl_csr.py b/lib/ansible/modules/crypto/openssl_csr.py index bb8bfb0ac3f..edc2591c5a4 100644 --- a/lib/ansible/modules/crypto/openssl_csr.py +++ b/lib/ansible/modules/crypto/openssl_csr.py @@ -622,183 +622,31 @@ class CertificateSigningRequestCryptography(CertificateSigningRequestBase): super(CertificateSigningRequestCryptography, self).__init__(module) self.cryptography_backend = cryptography.hazmat.backends.default_backend() - def _get_name_oid(self, id): - if id in ('CN', 'commonName'): - return cryptography.x509.oid.NameOID.COMMON_NAME - if id in ('C', 'countryName'): - return cryptography.x509.oid.NameOID.COUNTRY_NAME - if id in ('L', 'localityName'): - return cryptography.x509.oid.NameOID.LOCALITY_NAME - if id in ('ST', 'stateOrProvinceName'): - return cryptography.x509.oid.NameOID.STATE_OR_PROVINCE_NAME - if id in ('street', 'streetAddress'): - return cryptography.x509.oid.NameOID.STREET_ADDRESS - if id in ('O', 'organizationName'): - return cryptography.x509.oid.NameOID.ORGANIZATION_NAME - if id in ('OU', 'organizationalUnitName'): - return cryptography.x509.oid.NameOID.ORGANIZATIONAL_UNIT_NAME - if id in ('serialNumber', ): - return cryptography.x509.oid.NameOID.SERIAL_NUMBER - if id in ('SN', 'surname'): - return cryptography.x509.oid.NameOID.SURNAME - if id in ('GN', 'givenName'): - return cryptography.x509.oid.NameOID.GIVEN_NAME - if id in ('title', ): - return cryptography.x509.oid.NameOID.TITLE - if id in ('generationQualifier', ): - return cryptography.x509.oid.NameOID.GENERATION_QUALIFIER - if id in ('x500UniqueIdentifier', ): - return cryptography.x509.oid.NameOID.X500_UNIQUE_IDENTIFIER - if id in ('dnQualifier', ): - return cryptography.x509.oid.NameOID.DN_QUALIFIER - if id in ('pseudonym', ): - return cryptography.x509.oid.NameOID.PSEUDONYM - if id in ('UID', 'userId'): - return cryptography.x509.oid.NameOID.USER_ID - if id in ('DC', 'domainComponent'): - return cryptography.x509.oid.NameOID.DOMAIN_COMPONENT - if id in ('emailAddress', ): - return cryptography.x509.oid.NameOID.EMAIL_ADDRESS - if id in ('jurisdictionC', 'jurisdictionCountryName'): - return cryptography.x509.oid.NameOID.JURISDICTION_COUNTRY_NAME - if id in ('jurisdictionL', 'jurisdictionLocalityName'): - return cryptography.x509.oid.NameOID.JURISDICTION_LOCALITY_NAME - if id in ('jurisdictionST', 'jurisdictionStateOrProvinceName'): - return cryptography.x509.oid.NameOID.JURISDICTION_STATE_OR_PROVINCE_NAME - if id in ('businessCategory', ): - return cryptography.x509.oid.NameOID.BUSINESS_CATEGORY - if id in ('postalAddress', ): - return cryptography.x509.oid.NameOID.POSTAL_ADDRESS - if id in ('postalCode', ): - return cryptography.x509.oid.NameOID.POSTAL_CODE - raise CertificateSigningRequestError('Unknown subject field identifier "{0}"'.format(id)) - - def _get_san(self, name): - try: - if name.startswith('DNS:'): - return cryptography.x509.DNSName(to_text(name[4:])) - if name.startswith('IP:'): - return cryptography.x509.IPAddress(ipaddress.ip_address(to_text(name[3:]))) - if name.startswith('email:'): - return cryptography.x509.RFC822Name(to_text(name[6:])) - if name.startswith('URI:'): - return cryptography.x509.UniformResourceIdentifier(to_text(name[4:])) - except Exception as e: - raise CertificateSigningRequestError('Cannot parse Subject Alternative Name "{0}": {1}'.format(name, e)) - if ':' not in name: - raise CertificateSigningRequestError('Cannot parse Subject Alternative Name "{0}" (forgot "DNS:" prefix?)'.format(name)) - raise CertificateSigningRequestError('Cannot parse Subject Alternative Name "{0}" (potentially unsupported by cryptography backend)'.format(name)) - - def _get_keyusage(self, usage): - if usage in ('Digital Signature', 'digitalSignature'): - return 'digital_signature' - if usage in ('Non Repudiation', 'nonRepudiation'): - return 'content_commitment' - if usage in ('Key Encipherment', 'keyEncipherment'): - return 'key_encipherment' - if usage in ('Data Encipherment', 'dataEncipherment'): - return 'data_encipherment' - if usage in ('Key Agreement', 'keyAgreement'): - return 'key_agreement' - if usage in ('Certificate Sign', 'keyCertSign'): - return 'key_cert_sign' - if usage in ('CRL Sign', 'cRLSign'): - return 'crl_sign' - if usage in ('Encipher Only', 'encipherOnly'): - return 'encipher_only' - if usage in ('Decipher Only', 'decipherOnly'): - return 'decipher_only' - raise CertificateSigningRequestError('Unknown key usage "{0}"'.format(usage)) - - def _get_ext_keyusage(self, usage): - if usage in ('serverAuth', 'TLS Web Server Authentication'): - return cryptography.x509.oid.ExtendedKeyUsageOID.SERVER_AUTH - if usage in ('clientAuth', 'TLS Web Client Authentication'): - return cryptography.x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH - if usage in ('codeSigning', 'Code Signing'): - return cryptography.x509.oid.ExtendedKeyUsageOID.CODE_SIGNING - if usage in ('emailProtection', 'E-mail Protection'): - return cryptography.x509.oid.ExtendedKeyUsageOID.EMAIL_PROTECTION - if usage in ('timeStamping', 'Time Stamping'): - return cryptography.x509.oid.ExtendedKeyUsageOID.TIME_STAMPING - if usage in ('OCSPSigning', 'OCSP Signing'): - return cryptography.x509.oid.ExtendedKeyUsageOID.OCSP_SIGNING - if usage in ('anyExtendedKeyUsage', 'Any Extended Key Usage'): - return cryptography.x509.oid.ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE - if usage in ('qcStatements', ): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.3") - if usage in ('DVCS', ): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.3.10") - if usage in ('IPSec User', 'ipsecUser'): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.3.7") - if usage in ('Biometric Info', 'biometricInfo'): - return cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.2") - # FIXME need some more, probably all from https://www.iana.org/assignments/smi-numbers/smi-numbers.xhtml#smi-numbers-1.3.6.1.5.5.7.3 - raise CertificateSigningRequestError('Unknown extended key usage "{0}"'.format(usage)) - - def _get_basic_constraints(self, constraints): - ca = False - path_length = None - if constraints: - for constraint in constraints: - if constraint.startswith('CA:'): - if constraint == 'CA:TRUE': - ca = True - elif constraint == 'CA:FALSE': - ca = False - else: - raise CertificateSigningRequestError('Unknown basic constraint value "{0}" for CA'.format(constraint[3:])) - elif constraint.startswith('pathlen:'): - v = constraint[len('pathlen:'):] - try: - path_length = int(v) - except Exception as e: - raise CertificateSigningRequestError('Cannot parse path length constraint "{0}" ({1})'.format(v, e)) - else: - raise CertificateSigningRequestError('Unknown basic constraint "{0}"'.format(constraint)) - return ca, path_length - - def _parse_key_usage(self): - params = dict( - digital_signature=False, - content_commitment=False, - key_encipherment=False, - data_encipherment=False, - key_agreement=False, - key_cert_sign=False, - crl_sign=False, - encipher_only=False, - decipher_only=False, - ) - for usage in self.keyUsage: - params[self._get_keyusage(usage)] = True - return params - def _generate_csr(self): csr = cryptography.x509.CertificateSigningRequestBuilder() try: csr = csr.subject_name(cryptography.x509.Name([ - cryptography.x509.NameAttribute(self._get_name_oid(entry[0]), to_text(entry[1])) for entry in self.subject + cryptography.x509.NameAttribute(crypto_utils.cryptography_get_name_oid(entry[0]), to_text(entry[1])) for entry in self.subject ])) except ValueError as e: raise CertificateSigningRequestError(e) if self.subjectAltName: csr = csr.add_extension(cryptography.x509.SubjectAlternativeName([ - self._get_san(name) for name in self.subjectAltName + crypto_utils.cryptography_get_name(name) for name in self.subjectAltName ]), critical=self.subjectAltName_critical) if self.keyUsage: - params = self._parse_key_usage() + params = crypto_utils.cryptography_parse_key_usage_params(self.keyUsage) csr = csr.add_extension(cryptography.x509.KeyUsage(**params), critical=self.keyUsage_critical) if self.extendedKeyUsage: - usages = [self._get_ext_keyusage(usage) for usage in self.extendedKeyUsage] + usages = [crypto_utils.cryptography_get_ext_keyusage(usage) for usage in self.extendedKeyUsage] csr = csr.add_extension(cryptography.x509.ExtendedKeyUsage(usages), critical=self.extendedKeyUsage_critical) if self.basicConstraints: params = {} - ca, path_length = self._get_basic_constraints(self.basicConstraints) + ca, path_length = crypto_utils.cryptography_get_basic_constraints(self.basicConstraints) csr = csr.add_extension(cryptography.x509.BasicConstraints(ca, path_length), critical=self.basicConstraints_critical) if self.ocspMustStaple: @@ -842,7 +690,7 @@ class CertificateSigningRequestCryptography(CertificateSigningRequestBase): def _check_csr(self): def _check_subject(csr): - subject = [(self._get_name_oid(entry[0]), entry[1]) for entry in self.subject] + subject = [(crypto_utils.cryptography_get_name_oid(entry[0]), entry[1]) for entry in self.subject] current_subject = [(sub.oid, sub.value) for sub in csr.subject] return set(subject) == set(current_subject) @@ -855,7 +703,7 @@ class CertificateSigningRequestCryptography(CertificateSigningRequestBase): def _check_subjectAltName(extensions): current_altnames_ext = _find_extension(extensions, cryptography.x509.SubjectAlternativeName) current_altnames = [str(altname) for altname in current_altnames_ext.value] if current_altnames_ext else [] - altnames = [str(self._get_san(altname)) for altname in self.subjectAltName] if self.subjectAltName else [] + altnames = [str(crypto_utils.cryptography_get_name(altname)) for altname in self.subjectAltName] if self.subjectAltName else [] if set(altnames) != set(current_altnames): return False if altnames: @@ -869,7 +717,7 @@ class CertificateSigningRequestCryptography(CertificateSigningRequestBase): return current_keyusage_ext is None elif current_keyusage_ext is None: return False - params = self._parse_key_usage() + params = crypto_utils.cryptography_parse_key_usage_params(self.keyUsage) for param in params: if getattr(current_keyusage_ext.value, '_' + param) != params[param]: return False @@ -880,7 +728,7 @@ class CertificateSigningRequestCryptography(CertificateSigningRequestBase): def _check_extenededKeyUsage(extensions): current_usages_ext = _find_extension(extensions, cryptography.x509.ExtendedKeyUsage) current_usages = [str(usage) for usage in current_usages_ext.value] if current_usages_ext else [] - usages = [str(self._get_ext_keyusage(usage)) for usage in self.extendedKeyUsage] if self.extendedKeyUsage else [] + usages = [str(crypto_utils.cryptography_get_ext_keyusage(usage)) for usage in self.extendedKeyUsage] if self.extendedKeyUsage else [] if set(current_usages) != set(usages): return False if usages: @@ -892,7 +740,7 @@ class CertificateSigningRequestCryptography(CertificateSigningRequestBase): bc_ext = _find_extension(extensions, cryptography.x509.BasicConstraints) current_ca = bc_ext.ca if bc_ext else False current_path_length = bc_ext.path_length if bc_ext else None - ca, path_length = self._get_basic_constraints(self.basicConstraints) + ca, path_length = crypto_utils.cryptography_get_basic_constraints(self.basicConstraints) # Check CA flag if ca != current_ca: return False