diff --git a/acertmgr/authority/v2.py b/acertmgr/authority/v2.py index 4031887..b23ecc7 100644 --- a/acertmgr/authority/v2.py +++ b/acertmgr/authority/v2.py @@ -260,8 +260,8 @@ class ACMEAuthority(AbstractACMEAuthority): if code >= 400: raise ValueError("Error downloading certificate chain: {0} {1}".format(code, certificate)) - cert_dict = re.match((r'(?P-----BEGIN CERTIFICATE-----[^\-]+-----END CERTIFICATE-----)\n\n' - r'(?P-----BEGIN CERTIFICATE-----[^\-]+-----END CERTIFICATE-----)?'), + cert_dict = re.match((r'(?P^-----BEGIN CERTIFICATE-----\n[^\-]+\n-----END CERTIFICATE-----)\n*' + r'(?P-----BEGIN CERTIFICATE-----\n.+\n-----END CERTIFICATE-----)?$'), certificate, re.DOTALL).groupdict() cert = tools.convert_pem_str_to_cert(cert_dict['cert']) if cert_dict['ca'] is None: diff --git a/acertmgr/tools.py b/acertmgr/tools.py index 109e5d0..356551b 100644 --- a/acertmgr/tools.py +++ b/acertmgr/tools.py @@ -10,6 +10,7 @@ import base64 import datetime import io import os +import re import stat import sys import traceback @@ -257,15 +258,26 @@ def get_cert_valid_until(cert): # @brief convert certificate to PEM format -# @param cert certificate object in pyopenssl format +# @param cert certificate object or a list thereof # @return the certificate in PEM format def convert_cert_to_pem_str(cert): - return cert.public_bytes(serialization.Encoding.PEM).decode('utf8') + if not isinstance(cert, list): + cert = [cert] + result = list() + for data in cert: + result.append(data.public_bytes(serialization.Encoding.PEM).decode('utf8')) + return '\n'.join(result) # @brief load a PEM certificate from str +# @return a certificate object or a list of objects if multiple are in the string def convert_pem_str_to_cert(certdata): - return x509.load_pem_x509_certificate(certdata.encode('utf8'), default_backend()) + certs = re.findall(r'(-----BEGIN CERTIFICATE-----\n[^\-]+\n-----END CERTIFICATE-----)', + certdata, re.DOTALL) + result = list() + for data in certs: + result.append(x509.load_pem_x509_certificate(data.encode('utf8'), default_backend())) + return result[0] if len(result) == 1 else result # @brief serialize cert/csr to DER bytes @@ -411,6 +423,9 @@ def is_ocsp_valid(cert, issuer, hash_algo): log("Invalid hash algorithm '{}' used for OCSP validation. Validation ignored.".format(hash_algo), warning=True) return True + if isinstance(issuer, list): + issuer = issuer[0] # First certificate in the CA chain is the immediate issuer + try: ocsp_urls = [] aia = cert.extensions.get_extension_for_oid(ExtensionOID.AUTHORITY_INFORMATION_ACCESS)