Implement certificate information retrieval code in the ACME backends. (#736)

plugins/module_utils/acme/backend_cryptography.py

@@ -19,6 +19,7 @@ from ansible.module_utils.common.text.converters import to_bytes, to_native, to_
 from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
 
 from ansible_collections.community.crypto.plugins.module_utils.acme.backends import (
+    CertificateInformation,
     CryptoBackend,
 )
 
@@ -49,7 +50,9 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
 from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
     CRYPTOGRAPHY_TIMEZONE,
     cryptography_name_to_oid,
+    cryptography_serial_number_of_cert,
     get_not_valid_after,
+    get_not_valid_before,
 )
 
 from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
@@ -386,3 +389,44 @@ class CryptographyBackend(CryptoBackend):
         Given a Criterium object, creates a ChainMatcher object.
         '''
         return CryptographyChainMatcher(criterium, self.module)
+
+    def get_cert_information(self, cert_filename=None, cert_content=None):
+        '''
+        Return some information on a X.509 certificate as a CertificateInformation object.
+        '''
+        if cert_filename is not None:
+            cert_content = read_file(cert_filename)
+        else:
+            cert_content = to_bytes(cert_content)
+
+        # Make sure we have at most one PEM. Otherwise cryptography 36.0.0 will barf.
+        cert_content = to_bytes(extract_first_pem(to_text(cert_content)) or '')
+
+        try:
+            cert = cryptography.x509.load_pem_x509_certificate(cert_content, _cryptography_backend)
+        except Exception as e:
+            if cert_filename is None:
+                raise BackendException('Cannot parse certificate: {0}'.format(e))
+            raise BackendException('Cannot parse certificate {0}: {1}'.format(cert_filename, e))
+
+        ski = None
+        try:
+            ext = cert.extensions.get_extension_for_class(cryptography.x509.SubjectKeyIdentifier)
+            ski = ext.value.digest
+        except cryptography.x509.ExtensionNotFound:
+            pass
+
+        aki = None
+        try:
+            ext = cert.extensions.get_extension_for_class(cryptography.x509.AuthorityKeyIdentifier)
+            aki = ext.value.key_identifier
+        except cryptography.x509.ExtensionNotFound:
+            pass
+
+        return CertificateInformation(
+            not_valid_after=get_not_valid_after(cert),
+            not_valid_before=get_not_valid_before(cert),
+            serial_number=cryptography_serial_number_of_cert(cert),
+            subject_key_identifier=ski,
+            authority_key_identifier=aki,
+        )

plugins/module_utils/acme/backend_openssl_cli.py

@@ -20,6 +20,7 @@ import traceback
 from ansible.module_utils.common.text.converters import to_native, to_text, to_bytes
 
 from ansible_collections.community.crypto.plugins.module_utils.acme.backends import (
+    CertificateInformation,
     CryptoBackend,
 )
 
@@ -30,6 +31,8 @@ from ansible_collections.community.crypto.plugins.module_utils.acme.errors impor
 
 from ansible_collections.community.crypto.plugins.module_utils.acme.utils import nopad_b64
 
+from ansible_collections.community.crypto.plugins.module_utils.crypto.math import convert_bytes_to_int
+
 try:
     import ipaddress
 except ImportError:
@@ -39,6 +42,33 @@ except ImportError:
 _OPENSSL_ENVIRONMENT_UPDATE = dict(LANG='C', LC_ALL='C', LC_MESSAGES='C', LC_CTYPE='C')
 
 
+def _extract_date(out_text, name, cert_filename_suffix=""):
+    try:
+        date_str = re.search(r"\s+%s\s*:\s+(.*)" % name, out_text).group(1)
+        return datetime.datetime.strptime(date_str, '%b %d %H:%M:%S %Y %Z')
+    except AttributeError:
+        raise BackendException("No '{0}' date found{1}".format(name, cert_filename_suffix))
+    except ValueError as exc:
+        raise BackendException("Failed to parse '{0}' date{1}: {2}".format(name, cert_filename_suffix, exc))
+
+
+def _decode_octets(octets_text):
+    return binascii.unhexlify(re.sub(r"(\s|:)", "", octets_text).encode("utf-8"))
+
+
+def _extract_octets(out_text, name, required=True):
+    match = re.search(
+        r"\s+%s:\s*\n\s+([A-Fa-f0-9]{2}(?::[A-Fa-f0-9]{2})*)\s*\n" % name,
+        out_text,
+        re.MULTILINE | re.DOTALL,
+    )
+    if match is not None:
+        return _decode_octets(match.group(1))
+    if not required:
+        return None
+    raise BackendException("No '{0}' octet string found".format(name))
+
+
 class OpenSSLCLIBackend(CryptoBackend):
     def __init__(self, module, openssl_binary=None):
         super(OpenSSLCLIBackend, self).__init__(module)
@@ -89,10 +119,12 @@ class OpenSSLCLIBackend(CryptoBackend):
         dummy, out, dummy = self.module.run_command(
             openssl_keydump_cmd, check_rc=True, environ_update=_OPENSSL_ENVIRONMENT_UPDATE)
 
+        out_text = to_text(out, errors='surrogate_or_strict')
+
         if account_key_type == 'rsa':
-            pub_hex, pub_exp = re.search(
-                r"modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)",
-                to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL).groups()
+            pub_hex = re.search(r"modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent", out_text, re.MULTILINE | re.DOTALL).group(1)
+
+            pub_exp = re.search(r"\npublicExponent: ([0-9]+)", out_text, re.MULTILINE | re.DOTALL).group(1)
             pub_exp = "{0:x}".format(int(pub_exp))
             if len(pub_exp) % 2:
                 pub_exp = "0{0}".format(pub_exp)
@@ -104,17 +136,19 @@ class OpenSSLCLIBackend(CryptoBackend):
                 'jwk': {
                     "kty": "RSA",
                     "e": nopad_b64(binascii.unhexlify(pub_exp.encode("utf-8"))),
-                    "n": nopad_b64(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))),
+                    "n": nopad_b64(_decode_octets(pub_hex)),
                 },
                 'hash': 'sha256',
             }
         elif account_key_type == 'ec':
             pub_data = re.search(
                 r"pub:\s*\n\s+04:([a-f0-9\:\s]+?)\nASN1 OID: (\S+)(?:\nNIST CURVE: (\S+))?",
-                to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL)
+                out_text,
+                re.MULTILINE | re.DOTALL,
+            )
             if pub_data is None:
                 raise KeyParsingError('cannot parse elliptic curve key')
-            pub_hex = binascii.unhexlify(re.sub(r"(\s|:)", "", pub_data.group(1)).encode("utf-8"))
+            pub_hex = _decode_octets(pub_data.group(1))
             asn1_oid_curve = pub_data.group(2).lower()
             nist_curve = pub_data.group(3).lower() if pub_data.group(3) else None
             if asn1_oid_curve == 'prime256v1' or nist_curve == 'p-256':
@@ -303,13 +337,8 @@ class OpenSSLCLIBackend(CryptoBackend):
         openssl_cert_cmd = [self.openssl_binary, "x509", "-in", filename, "-noout", "-text"]
         dummy, out, dummy = self.module.run_command(
             openssl_cert_cmd, data=data, check_rc=True, binary_data=True, environ_update=_OPENSSL_ENVIRONMENT_UPDATE)
-        try:
-            not_after_str = re.search(r"\s+Not After\s*:\s+(.*)", to_text(out, errors='surrogate_or_strict')).group(1)
-            not_after = datetime.datetime.strptime(not_after_str, '%b %d %H:%M:%S %Y %Z')
-        except AttributeError:
-            raise BackendException("No 'Not after' date found{0}".format(cert_filename_suffix))
-        except ValueError:
-            raise BackendException("Failed to parse 'Not after' date{0}".format(cert_filename_suffix))
+        out_text = to_text(out, errors='surrogate_or_strict')
+        not_after = _extract_date(out_text, 'Not After', cert_filename_suffix=cert_filename_suffix)
         if now is None:
             now = datetime.datetime.now()
         return (not_after - now).days
@@ -319,3 +348,43 @@ class OpenSSLCLIBackend(CryptoBackend):
         Given a Criterium object, creates a ChainMatcher object.
         '''
         raise BackendException('Alternate chain matching can only be used with the "cryptography" backend.')
+
+    def get_cert_information(self, cert_filename=None, cert_content=None):
+        '''
+        Return some information on a X.509 certificate as a CertificateInformation object.
+        '''
+        filename = cert_filename
+        data = None
+        if cert_filename is not None:
+            cert_filename_suffix = ' in {0}'.format(cert_filename)
+        else:
+            filename = '/dev/stdin'
+            data = to_bytes(cert_content)
+            cert_filename_suffix = ''
+
+        openssl_cert_cmd = [self.openssl_binary, "x509", "-in", filename, "-noout", "-text"]
+        dummy, out, dummy = self.module.run_command(
+            openssl_cert_cmd, data=data, check_rc=True, binary_data=True, environ_update=_OPENSSL_ENVIRONMENT_UPDATE)
+        out_text = to_text(out, errors='surrogate_or_strict')
+
+        not_after = _extract_date(out_text, 'Not After', cert_filename_suffix=cert_filename_suffix)
+        not_before = _extract_date(out_text, 'Not Before', cert_filename_suffix=cert_filename_suffix)
+
+        sn = re.search(
+            r" Serial Number: ([0-9]+)",
+            to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL)
+        if sn:
+            serial = int(sn.group(1))
+        else:
+            serial = convert_bytes_to_int(_extract_octets(out_text, 'Serial Number', required=True))
+
+        ski = _extract_octets(out_text, 'X509v3 Subject Key Identifier', required=False)
+        aki = _extract_octets(out_text, 'X509v3 Authority Key Identifier', required=False)
+
+        return CertificateInformation(
+            not_valid_after=not_after,
+            not_valid_before=not_before,
+            serial_number=serial,
+            subject_key_identifier=ski,
+            authority_key_identifier=aki,
+        )

plugins/module_utils/acme/backends.py

@@ -9,10 +9,27 @@ from __future__ import absolute_import, division, print_function
 __metaclass__ = type
 
 
+from collections import namedtuple
 import abc
 
 from ansible.module_utils import six
 
+from ansible_collections.community.crypto.plugins.module_utils.acme.errors import (
+    BackendException,
+)
+
+
+CertificateInformation = namedtuple(
+    'CertificateInformation',
+    (
+        'not_valid_after',
+        'not_valid_before',
+        'serial_number',
+        'subject_key_identifier',
+        'authority_key_identifier',
+    ),
+)
+
 
 @six.add_metaclass(abc.ABCMeta)
 class CryptoBackend(object):
@@ -74,3 +91,12 @@ class CryptoBackend(object):
         '''
         Given a Criterium object, creates a ChainMatcher object.
         '''
+
+    def get_cert_information(self, cert_filename=None, cert_content=None):
+        '''
+        Return some information on a X.509 certificate as a CertificateInformation object.
+        '''
+        # Not implementing this method in a backend is DEPRECATED and will be
+        # disallowed in community.crypto 3.0.0. This method will be marked as
+        # @abstractmethod by then.
+        raise BackendException('This backend does not support get_cert_information()')

plugins/module_utils/crypto/math.py

@@ -110,6 +110,9 @@ if sys.version_info[0] >= 3:
     def _convert_int_to_bytes(count, no):
         return no.to_bytes(count, byteorder='big')
 
+    def _convert_bytes_to_int(data):
+        return int.from_bytes(data, byteorder='big', signed=False)
+
     def _to_hex(no):
         return hex(no)[2:]
 else:
@@ -122,6 +125,12 @@ else:
             raise Exception('Number {1} needs more than {0} bytes!'.format(count, n))
         return ('0' * (2 * count - len(h)) + h).decode('hex')
 
+    def _convert_bytes_to_int(data):
+        v = 0
+        for x in data:
+            v = (v << 8) | ord(x)
+        return v
+
     def _to_hex(no):
         return '%x' % no
 
@@ -155,3 +164,10 @@ def convert_int_to_hex(no, digits=None):
     if digits is not None and len(value) < digits:
         value = '0' * (digits - len(value)) + value
     return value
+
+
+def convert_bytes_to_int(data):
+    """
+    Convert a byte string to an unsigned integer in network byte order.
+    """
+    return _convert_bytes_to_int(data)