Load certificates using cryptography

- Use cryptography.x509 load_pem_x509_certificate or load_der_x509_certificate depending
  on the cert type. This ensures
  1) the certificate is a valid certificate
  2) trailing newlines and whitespaces will be ignored

- Ignore cer/crt as certificate type since these are file extensions and do not
  guarrantee the certificate encoding. Uses "pem" as default type for backwards
  compatibility. Only other valid option is "der". Everything else falls back to "pem".

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

Author: ioparaskev

src/saml2/metadata.py

@@ -1,7 +1,10 @@
 #!/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
+from saml2.sigver import security_context, 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 +691,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 +847,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
@@ -27,6 +28,10 @@
 
 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
@@ -632,46 +637,31 @@ def make_str(txt):
         return txt.decode()
 
 
-def read_cert_from_file(cert_file, cert_type):
+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 ''
-
-    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())
+        return ""
 
-    if cert_type in ['der', 'cer', 'crt']:
-        data = read_file(cert_file, 'rb')
-        _cert = base64.b64encode(data)
-        return make_str(_cert)
+    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):

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

tests/test_1.der

@@ -1,6 +1,8 @@
 import copy
 from saml2.config import SPConfig
-from saml2.metadata import create_metadata_string, entity_descriptor
+from saml2.metadata import create_metadata_string
+from saml2.metadata import entity_descriptor
+from saml2.metadata import read_cert
 from saml2.saml import NAME_FORMAT_URI, NAME_FORMAT_BASIC
 from saml2 import sigver
 
@@ -62,5 +64,18 @@ def test_signed_metadata_proper_str_bytes_handling():
     sp_metadata = create_metadata_string('', config=cnf, sign=True)
 
 
+def test_cert_trailing_newlines_ignored():
+    assert "".join(read_cert(full_path("extra_lines.crt"))) \
+           == "".join(read_cert(full_path("test_2.crt")))
+
+
+def test_invalid_cert_raises_error():
+    with pytest.raises(ValueError):
+        read_cert(full_path("malformed.crt"))
+
+
 if __name__ == '__main__':
     test_requested_attribute_name_format()
+    test_cert_trailing_newlines_ignored()
+    test_invalid_cert_raises_error()
+    test_signed_metadata_proper_str_bytes_handling()

tests/test_39_metadata.py

@@ -9,10 +9,10 @@
 from saml2 import time_util
 from saml2 import saml, samlp
 from saml2 import config
-from saml2.sigver import pre_encryption_part
+from saml2.sigver import pre_encryption_part, read_cert_from_file
 from saml2.sigver import make_temp
 from saml2.sigver import XmlsecError
-from saml2.sigver import SigverError
+from saml2.sigver import CertificateError
 from saml2.mdstore import MetadataStore
 from saml2.saml import assertion_from_string
 from saml2.saml import EncryptedAssertion
@@ -1114,6 +1114,21 @@ def test_xmlsec_output_line_parsing():
         sigver.parse_xmlsec_output(output4)
 
 
+def test_cert_trailing_newlines_ignored():
+    assert read_cert_from_file(full_path("extra_lines.crt")) \
+           == read_cert_from_file(full_path("test_2.crt"))
+
+
+def test_invalid_cert_raises_error():
+    with pytest.raises(CertificateError):
+        read_cert_from_file(full_path("malformed.crt"))
+
+
+def test_der_certificate_loading():
+    assert read_cert_from_file(full_path("test_1.der"), "der") == \
+           read_cert_from_file(full_path("test_1.crt"))
+
+
 if __name__ == "__main__":
     # t = TestSecurity()
     # t.setup_class()

tests/test_40_sigver.py

@@ -18,7 +18,7 @@
 client = Saml2Client(conf)
 
 # place a certificate in an authn request
-cert = read_cert_from_file(full_path("test.pem"), "pem")
+cert = read_cert_from_file(full_path("test.pem"))
 
 spcertenc = SPCertEnc(
     x509_data=ds.X509Data(