Merge branch 'feat-cert-load'

Author: c00kiemon5ter

setup.cfg

@@ -48,7 +48,7 @@ scripts =
     tools/parse_xsd2.py
 python_requires = >=3.6, <4
 install_requires =
-    cryptography >= 1.4
+    cryptography >= 3.1
     defusedxml
     pyOpenSSL
     python-dateutil

src/saml2/cert.py

@@ -5,9 +5,10 @@
 import dateutil.parser
 import pytz
 import six
-from OpenSSL import crypto
-from os.path import join
 from os import remove
+from os.path import join
+
+from OpenSSL import crypto
 
 import saml2.cryptography.pki
 
@@ -323,8 +324,7 @@ def verify(self, signing_cert_str, cert_str):
                 cert_algorithm = cert_algorithm.decode('ascii')
                 cert_str = cert_str.encode('ascii')
 
-            cert_crypto = saml2.cryptography.pki.load_pem_x509_certificate(
-                    cert_str)
+            cert_crypto = saml2.cryptography.pki.load_pem_x509_certificate(cert_str)
 
             try:
                 crypto.verify(ca_cert, cert_crypto.signature,
@@ -335,3 +335,28 @@ def verify(self, signing_cert_str, cert_str):
                 return False, "Certificate is incorrectly signed."
         except Exception as e:
             return False, "Certificate is not valid for an unknown reason. %s" % str(e)
+
+
+def read_cert_from_file(cert_file, cert_type="pem"):
+    """Read a certificate from a file.
+
+    If there are multiple certificates in the file, the first is returned.
+
+    :param cert_file: The name of the file
+    :param cert_type: The certificate type
+    :return: A base64 encoded certificate as a string or the empty string
+    """
+    if not cert_file:
+        return ""
+
+    with open(cert_file, "rb") as fp:
+        data = fp.read()
+
+    try:
+        cert = saml2.cryptography.pki.load_x509_certificate(data, cert_type)
+        pem_data = saml2.cryptography.pki.get_public_bytes_from_cert(cert)
+    except Exception as e:
+        raise CertificateError(e)
+
+    pem_data_no_headers = "".join(pem_data.splitlines()[1:-1])
+    return pem_data_no_headers

src/saml2/cryptography/asymmetric.py

@@ -1,15 +1,13 @@
 """This module provides methods for asymmetric cryptography."""
 
-import cryptography.hazmat.backends as _backends
 import cryptography.hazmat.primitives.asymmetric as _asymmetric
 import cryptography.hazmat.primitives.hashes as _hashes
 import cryptography.hazmat.primitives.serialization as _serialization
 
 
-def load_pem_private_key(data, password):
+def load_pem_private_key(data, password=None):
     """Load RSA PEM certificate."""
-    key = _serialization.load_pem_private_key(
-        data, password, _backends.default_backend())
+    key = _serialization.load_pem_private_key(data, password)
     return key
 
 

src/saml2/cryptography/pki.py

@@ -1,9 +1,48 @@
 """This module provides methods for PKI operations."""
 
-import cryptography.hazmat.backends as _backends
+from logging import getLogger as get_logger
+
 import cryptography.x509 as _x509
+from cryptography.hazmat.primitives.serialization import Encoding as _cryptography_encoding
+
+
+logger = get_logger(__name__)
+
+DEFAULT_CERT_TYPE = "pem"
 
 
 def load_pem_x509_certificate(data):
     """Load X.509 PEM certificate."""
-    return _x509.load_pem_x509_certificate(data, _backends.default_backend())
+    return _x509.load_pem_x509_certificate(data)
+
+
+def load_der_x509_certificate(data):
+    """Load X.509 DER certificate."""
+    return _x509.load_der_x509_certificate(data)
+
+
+def load_x509_certificate(data, cert_type="pem"):
+    cert_reader = _x509_loaders.get(cert_type)
+
+    if not cert_reader:
+        cert_reader = _x509_loaders.get("pem")
+        context = {
+            "message": "Unknown cert_type, falling back to default",
+            "cert_type": cert_type,
+            "default": DEFAULT_CERT_TYPE,
+        }
+        logger.warning(context)
+
+    cert = cert_reader(data)
+    return cert
+
+
+def get_public_bytes_from_cert(cert):
+    data = cert.public_bytes(_cryptography_encoding.PEM).decode()
+    return data
+
+
+_x509_loaders = {
+    "pem": load_pem_x509_certificate,
+    "der": load_der_x509_certificate,
+}

src/saml2/cryptography/symmetric.py

@@ -10,7 +10,6 @@
 from warnings import warn as _warn
 
 import cryptography.fernet as _fernet
-import cryptography.hazmat.backends as _backends
 import cryptography.hazmat.primitives.ciphers as _ciphers
 
 from .errors import SymmetricCryptographyError
@@ -158,10 +157,7 @@ def build_cipher(self, alg='aes_128_cbc'):
         except KeyError:
             raise Exception('Unsupported chaining mode: {}'.format(cmode))
 
-        cipher = _ciphers.Cipher(
-                _ciphers.algorithms.AES(self.key),
-                mode(iv),
-                backend=_backends.default_backend())
+        cipher = _ciphers.Cipher(_ciphers.algorithms.AES(self.key), mode(iv))
 
         return cipher, iv
 

src/saml2/metadata.py

@@ -2,6 +2,7 @@
 from saml2.algsupport import algorithm_support_in_metadata
 from saml2.md import AttributeProfile
 from saml2.sigver import security_context
+from saml2.cert import read_cert_from_file
 from saml2.config import Config
 from saml2.validate import valid_instance
 from saml2.time_util import in_a_while
@@ -688,14 +689,14 @@ def entity_descriptor(confd):
     enc_cert = None
     if confd.cert_file is not None:
         mycert = []
-        mycert.append("".join(read_cert(confd.cert_file)))
+        mycert.append(read_cert_from_file(confd.cert_file))
         if confd.additional_cert_files is not None:
             for _cert_file in confd.additional_cert_files:
-                mycert.append("".join(read_cert(_cert_file)))
+                mycert.append(read_cert_from_file(_cert_file))
     if confd.encryption_keypairs is not None:
         enc_cert = []
         for _encryption in confd.encryption_keypairs:
-            enc_cert.append("".join(read_cert(_encryption["cert_file"])))
+            enc_cert.append(read_cert_from_file(_encryption["cert_file"]))
 
     entd = md.EntityDescriptor()
     entd.entity_id = confd.entityid
@@ -844,9 +845,3 @@ def sign_entity_descriptor(edesc, ident, secc, sign_alg=None, digest_alg=None):
     xmldoc = secc.sign_statement("%s" % edesc, class_name(edesc))
     edesc = md.entity_descriptor_from_string(xmldoc)
     return edesc, xmldoc
-
-
-def read_cert(path):
-    with open(path) as fp:
-        lines = fp.readlines()
-    return lines[1:-1]

src/saml2/sigver.py

@@ -13,6 +13,7 @@
 import six
 import sys
 from uuid import uuid4 as gen_random_key
+
 from time import mktime
 from tempfile import NamedTemporaryFile
 from subprocess import Popen
@@ -43,6 +44,8 @@
 from saml2 import saml
 from saml2 import ExtensionElement
 from saml2.cert import OpenSSLWrapper
+from saml2.cert import read_cert_from_file
+from saml2.cert import CertificateError
 from saml2.extension import pefim
 from saml2.extension.pefim import SPCertEnc
 from saml2.saml import EncryptedAssertion
@@ -108,10 +111,6 @@ class BadSignature(SigverError):
     pass
 
 
-class CertificateError(SigverError):
-    pass
-
-
 def get_pem_wrapped_unwrapped(cert):
     begin_cert = "-----BEGIN CERTIFICATE-----\n"
     end_cert = "\n-----END CERTIFICATE-----\n"
@@ -120,11 +119,6 @@ def get_pem_wrapped_unwrapped(cert):
     return wrapped_cert, unwrapped_cert
 
 
-def read_file(*args, **kwargs):
-    with open(*args, **kwargs) as handler:
-        return handler.read()
-
-
 def rm_xmltag(statement):
     XMLTAG = "<?xml version='1.0'?>"
     PREFIX1 = "<?xml version='1.0' encoding='UTF-8'?>"
@@ -489,8 +483,9 @@ def pem_format(key):
 
 
 def import_rsa_key_from_file(filename):
-    data = read_file(filename, 'rb')
-    key = saml2.cryptography.asymmetric.load_pem_private_key(data, None)
+    with open(filename, "rb") as fd:
+        data = fd.read()
+    key = saml2.cryptography.asymmetric.load_pem_private_key(data)
     return key
 
 
@@ -625,55 +620,6 @@ def verify_redirect_signature(saml_msg, crypto, cert=None, sigkey=None):
             return bool(signer.verify(string, _sign, _key))
 
 
-def make_str(txt):
-    if isinstance(txt, six.string_types):
-        return txt
-    else:
-        return txt.decode()
-
-
-def read_cert_from_file(cert_file, cert_type):
-    """ Reads a certificate from a file. The assumption is that there is
-    only one certificate in the file
-
-    :param cert_file: The name of the file
-    :param cert_type: The certificate type
-    :return: A base64 encoded certificate as a string or the empty string
-    """
-
-    if not cert_file:
-        return ''
-
-    if cert_type == 'pem':
-        _a = read_file(cert_file, 'rb').decode()
-        _b = _a.replace('\r\n', '\n')
-        lines = _b.split('\n')
-
-        for pattern in (
-                '-----BEGIN CERTIFICATE-----',
-                '-----BEGIN PUBLIC KEY-----'):
-            if pattern in lines:
-                lines = lines[lines.index(pattern) + 1:]
-                break
-        else:
-            raise CertificateError('Strange beginning of PEM file')
-
-        for pattern in (
-                '-----END CERTIFICATE-----',
-                '-----END PUBLIC KEY-----'):
-            if pattern in lines:
-                lines = lines[:lines.index(pattern)]
-                break
-        else:
-            raise CertificateError('Strange end of PEM file')
-        return make_str(''.join(lines).encode())
-
-    if cert_type in ['der', 'cer', 'crt']:
-        data = read_file(cert_file, 'rb')
-        _cert = base64.b64encode(data)
-        return make_str(_cert)
-
-
 class CryptoBackend(object):
     def version(self):
         raise NotImplementedError()

tests/extra_lines.crt

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIICITCCAYoCAQEwDQYJKoZIhvcNAQELBQAwWDELMAkGA1UEBhMCenoxCzAJBgNV
+BAgMAnp6MQ0wCwYDVQQHDAR6enp6MQ4wDAYDVQQKDAVaenp6ejEOMAwGA1UECwwF
+Wnp6enoxDTALBgNVBAMMBHRlc3QwIBcNMTkwNDEyMTk1MDM0WhgPMzAxODA4MTMx
+OTUwMzRaMFgxCzAJBgNVBAYTAnp6MQswCQYDVQQIDAJ6ejENMAsGA1UEBwwEenp6
+ejEOMAwGA1UECgwFWnp6enoxDjAMBgNVBAsMBVp6enp6MQ0wCwYDVQQDDAR0ZXN0
+MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDjW0kJM+4baWKtvO24ZsGXNvNK
+KkwTMz7OW5Z6BRqhSOq2WA0c5NCpMk6rD8Z2OTFEolPojEjf8dVyd/Ds/hrjFKQv
+8wQgbdXLN51YTIsgd6h+hBJO+vzhl0PT4aT7M0JKo5ALtS6qk4tsworW2BnwyvsG
+SAinwfeWt4t/b1J3kwIDAQABMA0GCSqGSIb3DQEBCwUAA4GBAFtj7WArQQBugmh/
+KQjjlfTQ5A052QeXfgTyO9vv1S6MRIi7qgiaEv49cGXnJv/TWbySkMKObPMUApjg
+6z8PqcxuShew5FCTkNvwhABFPiyu0fUj3e2FEPHfsBu76jz4ugtmhUqjqhzwFY9c
+tnWRkkl6J0AjM3LnHOSgjNIclDZG
+-----END CERTIFICATE-----
+
+
+
+
+

tests/malformed.crt

@@ -0,0 +1,12 @@
+-----BEGIN CERTIFICATE-----
+MIICITCCAYoCAQEwDQYJKoZIhvcNAQELBQAwWDELMAkGA1UEBhMCenoxCzAJBgNV
+BAgMAnp6MQ0wCwYDVQQHDAR6enp6MQ4wDAYDVQQKDAVaenp6ejEOMAwGA1UECwwF
+Wnp6enoxDTALBgNVBAMMBHRlc3QwIBcNMTkwNDEyMTk1MDM0WhgPMzAxODA4MTMx
+OTUwMzRaMFgxCzAJBgNVBAYTAnp6MQswCQYDVQQIDAJ6ejENMAsGA1UEBwwEenp6
+MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDjW0kJM+4baWKtvO24ZsGXNvNK
+KkwTMz7OW5Z6BRqhSOq2WA0c5NCpMk6rD8Z2OTFEolPojEjf8dVyd/Ds/hrjFKQv
+8wQgbdXLN51YTIsgd6h+hBJO+vzhl0PT4aT7M0JKo5ALtS6qk4tsworW2BnwyvsG
+SAinwfeWt4t/b1J3kwIDAQABMA0GCSqGSIb3DQEBCwUAA4GBAFtj7WArQQBugmh/
+KQjjlfTQ5A052QeXfgTyO9vv1S6MRIi7qgiaEv49cGXnJv/TWbySkMKObPMUApjg
+6z8PqcxuShew5FCTkNvwhABFPiyu0fUj3e2FEPHfsBu76jz4ugtmhUqjqhzwFY9c
+tnWRkkl6J0AjM3LnHOSgjNIclDZG