Refactor certificate loading

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

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.
+
+    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 ""
+
+    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

@@ -5,7 +5,7 @@
 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)
     return key

src/saml2/cryptography/pki.py

@@ -1,8 +1,48 @@
 """This module provides methods for PKI operations."""
 
+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)
+
+
+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/metadata.py

@@ -1,10 +1,8 @@
 #!/usr/bin/env python
-from cryptography import x509
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives.serialization import Encoding
 from saml2.algsupport import algorithm_support_in_metadata
 from saml2.md import AttributeProfile
-from saml2.sigver import security_context, read_cert_from_file
+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

src/saml2/sigver.py

@@ -28,10 +28,6 @@
 
 from OpenSSL import crypto
 
-from cryptography import x509
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives.serialization import Encoding
-
 import pytz
 
 from six.moves.urllib import parse
@@ -48,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
@@ -113,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"
@@ -125,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'?>"
@@ -494,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
 
 
@@ -630,40 +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="pem"):
-    """ 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 ""
-
-    with open(cert_file, "rb") as fp:
-        data = fp.read()
-
-    cert_reader = (x509.load_der_x509_certificate
-                   if cert_type == 'der'
-                   else x509.load_pem_x509_certificate)
-
-    try:
-        cert = cert_reader(data, default_backend())
-        pem_data = cert.public_bytes(Encoding.PEM)
-    except Exception as e:
-        raise CertificateError(e)
-    else:
-        return "".join(pem_data.decode().splitlines()[1:-1])
-
-
 class CryptoBackend(object):
     def version(self):
         raise NotImplementedError()

tests/test_39_metadata.py

@@ -2,13 +2,15 @@
 from saml2.config import SPConfig
 from saml2.metadata import create_metadata_string
 from saml2.metadata import entity_descriptor
-from saml2.metadata import read_cert
+from saml2.cert import read_cert_from_file as read_cert
+from saml2.cert import CertificateError
 from saml2.saml import NAME_FORMAT_URI, NAME_FORMAT_BASIC
 from saml2 import sigver
 
 from pathutils import full_path
 
-__author__ = 'roland'
+from pytest import raises
+
 
 sp_conf = {
     "entityid": "urn:mace:umu.se:saml:roland:sp",
@@ -70,7 +72,7 @@ def test_cert_trailing_newlines_ignored():
 
 
 def test_invalid_cert_raises_error():
-    with pytest.raises(ValueError):
+    with raises(CertificateError):
         read_cert(full_path("malformed.crt"))
 
 

tests/test_40_sigver.py

@@ -9,10 +9,11 @@
 from saml2 import time_util
 from saml2 import saml, samlp
 from saml2 import config
-from saml2.sigver import pre_encryption_part, read_cert_from_file
+from saml2.cert import read_cert_from_file
+from saml2.cert import CertificateError
+from saml2.sigver import pre_encryption_part
 from saml2.sigver import make_temp
 from saml2.sigver import XmlsecError
-from saml2.sigver import CertificateError
 from saml2.mdstore import MetadataStore
 from saml2.saml import assertion_from_string
 from saml2.saml import EncryptedAssertion
@@ -97,8 +98,7 @@ def test_cert_from_instance_1():
     assert certs[0] == CERT1
 
 
-@pytest.mark.skipif(not decoder,
-                    reason="pyasn1 is not installed")
+@pytest.mark.skipif(not decoder, reason="pyasn1 is not installed")
 def test_cert_from_instance_ssp():
     with open(SIMPLE_SAML_PHP_RESPONSE) as fp:
         xml_response = fp.read()
@@ -1120,7 +1120,7 @@ def test_cert_trailing_newlines_ignored():
 
 
 def test_invalid_cert_raises_error():
-    with pytest.raises(CertificateError):
+    with raises(CertificateError):
         read_cert_from_file(full_path("malformed.crt"))