optools/ssl_tls/certparser.py

#!/usr/bin/env python3

# stdlib
import argparse
import collections
import copy
import datetime
import hashlib
import importlib
import ipaddress
import json
import os
import pprint
import re
import shutil
import socket
import ssl
from urllib import parse
# PyPi/PIP
# These are handled automagically.
# If you'd rather install them via your distro's package manager (YOU SHOULD),
# then install them first then run this script.
# Otherwise you'll have to use pip to remove them.
thrd_prty = {'OpenSSL': 'pyOpenSSL',
             #'pyasn1': 'pyasn1',
             #'jinja2': 'Jinja2',
             'validators': 'validators'}

cols = shutil.get_terminal_size((80, 20)).columns

for mod in thrd_prty:
    try:
        globals()[mod] = importlib.import_module(mod)
    except ImportError:
        import pip
        pip.main(['install', '--quiet', '--quiet', '--quiet',
                  '--user', thrd_prty[mod]])
        globals()[mod] = importlib.import_module(mod)

class CertParse(object):
    def __init__(self, target, port = 443, force = None, cert_type = 'pem',
                 json_fmt = False, starttls = False, extensions = False,
                 alt_names = False):
        self.target = target
        self.port = port
        self.force_type = force
        self.cert_type = cert_type
        self.starttls = starttls
        self.json_fmt = json_fmt
        self.extensions = extensions
        self.alt_names = alt_names
        self.cert = None
        self.certinfo = None
        self.get_type()

    def getCert(self):
        if self.cert_type.lower() == 'pem':
            self.cert_type = OpenSSL.crypto.FILETYPE_PEM
        elif self.cert_type.lower() == 'asn1':
            self.cert_type = OpenSSL.crypto.FILETYPE_ASN1
        else:
            raise ValueError(('{0} is not a valid cert type; must be either ' +
                              '"pem" or "asn1"').format(self.cert_type))
        if not self.force_type in ('url', 'domain', 'ip'):
            with open(self.target, 'rb') as f:
                self.cert = OpenSSL.crypto.load_certificate(self.cert_type,
                                                            f.read())
        else:
            _cert = ssl.get_server_certificate((self.target, self.port))
            self.cert = OpenSSL.crypto.load_certificate(self.cert_type,
                                                        _cert)
        return()

    def parseCert(self):
        certinfo = collections.OrderedDict()
        timefmt = '%Y%m%d%H%M%SZ'
        certinfo['Subject'] = self.parse_name(self.cert.get_subject().\
                                                            get_components())
        certinfo['EXPIRED'] = self.cert.has_expired()
        certinfo['Issuer'] = self.parse_name(self.cert.get_issuer().\
                                                            get_components())
        certinfo['Issued'] = str(datetime.datetime.strptime(
                                    self.cert.get_notBefore().decode('utf-8'),
                                    timefmt))
        certinfo['Expires'] = str(datetime.datetime.strptime(
                                    self.cert.get_notAfter().decode('utf-8'),
                                    timefmt))
        if self.extensions:
            certinfo['Extensions'] = self.parse_ext()
        elif self.alt_names:
            certinfo['SANs'] = self.parse_ext_san_only()
        certinfo['Pubkey'] = self.get_pubkey()
        certinfo['Serial'] = int(self.cert.get_serial_number())
        certinfo['Signature Algorithm'] = self.cert.get_signature_algorithm().\
                                                                decode('utf-8')
        certinfo['Version'] = self.cert.get_version()
        certinfo['Subject Name Hash'] = self.cert.subject_name_hash()
        certinfo['Fingerprints'] = self.gen_hashes()
        self.certinfo = certinfo
        return()

    def print(self, json_fmt = None):
        if json_fmt is None:
            json_fmt = self.json_fmt
        if json_fmt:
            output = json.dumps(self.certinfo, indent = 4)
        else:
            output = self.certinfo
        if __name__ == '__main__':
            if not json_fmt:
                pprint.pprint(output, compact = False, width = cols)
            else:
                print(output)
            return()
        return(output)

    def get_pubkey(self):
        pubkey = {}
        key = self.cert.get_pubkey()
        pubkey['Bit Length'] = key.bits()
        # I wish there was a more comfortable way of comparing these.
        if key.type() == OpenSSL.crypto.TYPE_RSA:
            pubkey['Algorithm'] = 'RSA'
        elif key.type() == OpenSSL.crypto.TYPE_DSA:
            pubkey['Algorithm'] = 'DSA'
        return(pubkey)

    def gen_hashes(self):
        hashes = {}
        # Note: MD2 is *so old* that they aren't even
        # *supported in python 3*.
        # If we NEED to implement, https://urchin.earth.li/~twic/md2.py
        fpt_types = sorted([i.lower() for i in ['md2', 'md5', 'sha1', 'mdc2',
                                                'ripemd160', 'blake2b512',
                                                'blake2s256', 'sha224',
                                                'sha256', 'sha384', 'sha512']])
        supported_types = sorted([i.lower() for i in \
                                  list(hashlib.algorithms_available)])
        cert_hash_types = [i for i in fpt_types if i in supported_types]
        for h in cert_hash_types:
            hashes[h.upper()] = self.cert.digest(h).decode('utf-8')
        return(hashes)

    def parse_name(self, item):
        component_map = {'C': 'Country',
                         'countryName': 'Country',
                         'ST': 'State/Province',
                         'stateOrProvinceName': 'State/Province',
                         'L': 'Locality/City/Town/Region',
                         'localityName': 'Locality/City/Town/Region',
                         'O': 'Organization',
                         'organizationName': 'Organization',
                         'OU': 'Department/Team/Organization Unit',
                         'organizationalUnitName': ('Department/Team/' +
                                                    'Organization Unit'),
                         'CN': 'Common name',
                         'commonName': 'Common name',
                         'emailAddress': 'eMail Address'}
        info = {}
        for c in item:
            item = c[0].decode('utf-8')
            value = c[1].decode('utf-8')
            if item in component_map.keys():
                info[component_map[item]] = value
            else:
                info[item] = value
        return(info)

    def parse_ext_san_only(self):
        SANs = []
        for idx in range(0, self.cert.get_extension_count()):
            ext = self.cert.get_extension(idx)
            name = ext.get_short_name().decode('utf-8').lower()
            x = str(ext).strip()
            if name == 'subjectaltname':
                val_lst = [i.strip() for i in x.split(',')]
                for v in val_lst:
                    parsed_val = re.sub('^\s*DNS:\s*(.*)', '\g<1>', v)
                    if parsed_val not in ('\n', ''):
                        SANs.append(parsed_val.lower())
        return(SANs)

    def parse_ext(self):
        exts = {}
        for idx in range(0, self.cert.get_extension_count()):
            ext = self.cert.get_extension(idx)
            keyname = ext.get_short_name().decode('utf-8')
            value_str = str(ext).strip()
            # These should be split into lists by commas.
            if keyname in ('subjectAltName', 'keyUsage', 'extendedKeyUsage',
                           'basicConstraints'):
                val_lst = [i.strip() for i in value_str.split(',')]
                value_str = []
                for v in val_lst:
                    parsed_val = re.sub('^\s*DNS:\s*(.*)', '\g<1>', v)
                    if parsed_val not in ('\n', ''):
                        value_str.append(parsed_val)
            # These should be split into lists by lines.
            elif keyname in ('certificatePolicies', 'ct_precert_scts',
                             'authorityInfoAccess'):
                val_lst = [i.strip() for i in value_str.splitlines()]
                value_str = []
                for v in val_lst:
                    value_str.append(v)
            exts[keyname] = value_str
        # These are split FURTHER into dicts but require unique... massaging.
        # authorityInfoAccess
        if 'authorityInfoAccess' in exts.keys():
            _tmp = copy.deepcopy(exts['authorityInfoAccess'])
            exts['authorityInfoAccess'] = {}
            for i in _tmp:
                x = [n.strip() for n in i.split('-', 1)]
                y = [n.strip() for n in x[1].split(':', 1)]
                exts['authorityInfoAccess'][x[0]] = {y[0]: y[1]}
        # authorityKeyIdentifier
        if 'authorityKeyIdentifier' in exts.keys():
            _tmp = copy.deepcopy(exts['authorityKeyIdentifier'])
            exts['authorityKeyIdentifier'] = {_tmp.split(':', 1)[0]:
                                                        _tmp.split(':', 1)[1]}
        # basicConstraints
        if 'basicConstraints' in exts.keys():
            _tmp = copy.deepcopy(exts['basicConstraints'])
            exts['basicConstraints'] = {}
            for i in _tmp:
                x = [n.strip() for n in i.split(':', 1)]
                if len(x) >= 1:
                    if x[1].lower() in ('true', 'false'):
                        x[1] = (x[1].lower() == 'true')
                    exts['basicConstraints'][x[0]] = x[1]
                else:
                    exts['basicConstraints'][x[0]] = True
        # certificatePolicies
        # What a mess.
        if 'certificatePolicies' in exts.keys():
            _tmp = copy.deepcopy(exts['certificatePolicies'])
            exts['certificatePolicies'] = {}
            last_key = None
            for i in [n.strip() for n in _tmp]:
                l = [y for y in i.split(':', 1) if y not in ('', None)]
                if len(l) > 1:
                    # It MAY be a key:value.
                    if re.search('^\s+', l[1]):
                        val = l[1].strip()
                        if last_key == 'Policy':
                            if not isinstance(exts['certificatePolicies']\
                                                                    [last_key],
                                              list):
                                exts['certificatePolicies'][last_key] = [
                                        exts['certificatePolicies'][last_key]]
                            exts['certificatePolicies'][last_key].append(val)
                        # I can't seem to get CPS as a separate dict.
                        # Patches welcome.
                        # Also, are CPS and User Notice *subitems* of Policy
                        # items?
                        elif last_key not in ('User Notice', 'CPS'):
                            # It's a value.
                            last_key = l[0].strip()
                            exts['certificatePolicies'][last_key] = val
                        else:
                            k = l[0].strip()
                            exts['certificatePolicies'][last_key][k] = val
                else:
                    # Standalone key line
                    last_key = l[0].strip()
                    exts['certificatePolicies'][last_key] = {}
        # ct_precert_scts
        # another mess. a much. much, bigger mess.
        if 'ct_precert_scts' in exts.keys():
            _tmp = copy.deepcopy(exts['ct_precert_scts'])
            exts['ct_precert_scts'] = {}
            last_key = None
            last_sub_key = None
            cnt = 0
            for i in [n.strip() for n in _tmp]:
                l = [y for y in i.split(':', 1) if y not in ('', None)]
                if len(l) > 1:
                    # Is it a line continuation (of a hex value)?
                    if ((re.search('^[0-9A-Z]{2}$', l[0])) and
                                (re.search('^[0-9A-Z:]*:?$', ':'.join(l)))):
                        exts['ct_precert_scts'][last_key][cnt]\
                                                [last_sub_key] += ':'.join(l)
                        continue
                    # It MAY be a key:value.
                    if re.search('^\s+', l[1]) and (
                                            last_key !=
                                            'Signed Certificate Timestamp'):
                        # It's a value.
                        last_key = l[0].strip()
                        val = l[1].strip()
                        if val.lower() == 'none':
                            val = None
                        exts['ct_precert_scts'][last_key] = val
                    elif re.search('^\s+', l[1]):
                        last_sub_key = l[0].strip()
                        val = l[1].strip()
                        if val.lower() == 'none':
                            val = None
                        if last_sub_key == 'Signature':
                            val += ' '
                        exts['ct_precert_scts'][last_key][cnt]\
                                                        [last_sub_key] = val
                else:
                    # Standalone key line
                    last_key = l[0].strip()
                    if last_key == 'Signed Certificate Timestamp':
                        if last_key not in exts['ct_precert_scts'].keys():
                            exts['ct_precert_scts'][last_key] = [{}]
                        else:
                            exts['ct_precert_scts'][last_key].append({})
                            cnt += 1
            # some laaaast bit of cleanup...
            if 'Signed Certificate Timestamp' in exts['ct_precert_scts']:
                for i in exts['ct_precert_scts']\
                                            ['Signed Certificate Timestamp']:
                    if 'Signature' in i.keys():
                        d = i['Signature'].split()
                        i['Signature'] = {d[0]: d[1]}
        return(exts)

    def get_domain_from_url(self, url):
        orig_url = url
        # Needed in case a URL is passed with no http:// or https://, etc.
        url = re.sub('^((ht|f)tps?://)*',
                     'https://',
                     url,
                     re.IGNORECASE).lower()
        if not self.validURL(url):
            raise ValueError(('{0} is not a valid URL').format(orig_url))
        domain = parse.urlparse(url).netloc
        return(domain)

    def validIP(self, ip):
        is_valid = False
        try:
            ipaddress.ip_address(self.target)
            is_valid = True
        except ValueError:
            pass
        return(is_valid)

    def validDomain(self, domain):
        is_valid = False
        if not isinstance(validators.domain(domain),
                          validators.utils.ValidationFailure):
            is_valid = True
        return(is_valid)

    def validURL(self, url):
        is_valid = False
        if not isinstance(validators.url(url),
                          validators.utils.ValidationFailure):
            is_valid = True
        return(is_valid)

    def validPath(self, path):
        is_valid = False
        if os.path.isfile(path):
            is_valid = True
        return(is_valid)

    def get_type(self):
        if self.force_type:
            # Just run the validator and some cleanup.
            if self.force_type == 'url':
                self.target = self.get_domain_from_url(self.target)
                chk = self.validURL(self.target)
                if chk:
                    self.force_type = 'domain'
            elif self.force_type == 'ip':
                chk = self.validIP(self.target)
            elif self.force_type == 'domain':
                chk = self.validDomain(self.target)
            elif self.force_type == 'file':
                self.target = os.path.abspath(os.path.expanduser(self.target))
                chk = self.validPath(self.target)
            if not chk:
                raise TypeError(('{0} does not appear to be a valid ' +
                                 'instance of type {1}'.format(self.target,
                                                               self.force_type)
                                ))
            if self.force_type in ('url', 'domain', 'ip'):
                self.remote = True
            else:
                self.remote = False
            return()
        # Is it an IP address?
        if self.validIP(self.target):
            self.force_type = 'ip'
            return()
        # Is it a filepath?
        fpath = os.path.abspath(os.path.expanduser(self.target))
        if self.validPath(fpath):
            self.target = fpath
            self.force_type = 'file'
            return()
        # Is it a domain?
        if self.validDomain(self.target):
            self.force_type = 'domain'
            return()
        # Lastly, is it a URL?
        if self.validURL(self.target):
            domain = self.get_domain_from_url(self.target)
            if self.validDomain(domain):
                self.target = domain
                self.force_type = 'domain'
        if not self.force_type:  # We couldn't detect it
            raise RuntimeError(('Automatic type detection of {0} requested ' +
                                'but we could not determine what type of ' +
                                'resource it is'))
        return()

def parseArgs():
    args = argparse.ArgumentParser()
    args.add_argument('-e', '--extensions',
                      dest = 'extensions',
                      action = 'store_true',
                      help = ('If specified, include ALL extension info ' +
                              '(this DRASTICALLY increases the output. You ' +
                              'have been warned)'))
    args.add_argument('-a', '--alt-names',
                      dest = 'alt_names',
                      action = 'store_true',
                      help = ('If specified, ONLY include the SAN (Subject ' +
                              'Alt Name) extension. This is highly ' +
                              'recommended over -e/--extensions. Ignored if ' +
                              '-e/--extensions is set (as the SANs are ' +
                              'included in that)'))
    args.add_argument('-j','--json',
                      dest = 'json_fmt',
                      action = 'store_true',
                      help = ('If specified, return the results in JSON'))
    args.add_argument('-f', '--force',
                      choices = ['url', 'domain', 'ip', 'file'],
                      default = None,
                      help = ('If specified, force the TARGET to be parsed ' +
                              'as the given type'))
    args.add_argument('-p', '--port',
                      dest = 'port',
                      type = int,
                      default = 443,
                      help = ('Use a port other than 443 (only used for ' +
                              'URL/domain/IP address targets)'))
    args.add_argument('-t', '--cert-type',
                      dest = 'cert_type',
                      default = 'pem',
                      choices = ['pem', 'asn1'],
                      help = ('The type of certificate (only used for '
                              'file targets). Note that "DER"-encoded ' +
                              'certificates should use "asn1". The default ' +
                              'is pem'))
#    TODO: I think the starttls process depends on the protocol? If so, this...
#          won't be feasible.
#    args.add_argument('-s', '--starttls',
#                      dest = 'starttls',
#                      action = 'store_true',
#                      help = ('If specified, initiate STARTTLS on the ' +
#                              'target instead of pure SSL/TLS'))
    args.add_argument('TARGET',
                      help = ('The target to gather cert info for. Can be a ' +
                              'filepath (to the certificate, not key etc.), ' +
                              'a URL/domain, or IP address'))
    return(args)

def main():
    args = vars(parseArgs().parse_args())
    args['target'] = copy.deepcopy(args['TARGET'])
    del(args['TARGET'])
    p = CertParse(**args)
    p.getCert()
    p.parseCert()
    p.print()

if __name__ == '__main__':
    main()