list and simple search of AWS Certificate Manager SSL certs using python and boto3

acm_search_certs.py

#!/usr/bin/env python3

"""
usage: acm_search_certs.py [-h] [--region REGION] [--san-names SAN_NAMES] [--identifiers IDENTIFIERS]
                    [--tags TAGS] [--issuers ISSUERS] [--cert-in-use {0,1}] [--json] [--and-filters]

Fetches ACM SSL certificates.

options:
  -h, --help            show this help message and exit
  --region REGION, -r REGION
                        AWS region (default "us-east-1")
  --san-names SAN_NAMES, -n SAN_NAMES
                        substring patterns to match in SAN hostnames - comma-separate multiple patterns
  --identifiers IDENTIFIERS, -d IDENTIFIERS
                        AWS identifier - exact match, comma-separate multiples
  --tags TAGS, -t TAGS  dotcms.client.name.short names - exact match, comma-separate multiples
  --issuers ISSUERS, -i ISSUERS
                        substring patterns to match in cert issuer - comma-separate multiple patterns
  --cert-in-use {0,1}, -u {0,1}
                        0 or 1 -- 'cert in use' filter
  --json, -j            print output as json
  --and-filters, -a     join multiple filters with AND rather than OR
"""

import argparse
import json

import boto3

# cli search of ACM-managed SSL certifictes


class ACM:
    """if only AWS ACM were searchable"""

    def __init__(self, region="us-east-1", debug=False):
        """initialize self.all_certs with all ACM certs in this region"""
        self.region = region
        self.debug = debug
        self.client = boto3.client("acm", region_name=region)
        self.tag_key = "dotcms.client.name.short"
        # use arn as key in self.all_certs
        self.all_certs = {}
        for cert in self.client.list_certificates()["CertificateSummaryList"]:
            arn = cert["CertificateArn"]
            self.all_certs[arn] = self.client.describe_certificate(CertificateArn=arn)[
                "Certificate"
            ]

    def filter_certs(
        self,
        san_names=(),
        identifiers=(),
        tags=(),
        issuers=(),
        cert_in_use=None,
        and_filters=False,
    ):
        """
        filters self.all_certs if filter patterns are provided
        multiple filters are OR'd
        """
        filtered_results = []
        if san_names:
            filtered_results.append(self._filter_san_names(san_names))
        if identifiers:
            filtered_results.append(self._filter_identifier(identifiers))
        if tags:
            filtered_results.append(self._filter_tags(tags))
        if issuers:
            filtered_results.append(self._filter_issuers(issuers))
        if cert_in_use is not None:
            filter_results.append(self._filter_in_use(cert_in_use))

        if filtered_results:
            if and_filters:
                self._join_filters_and(filtered_results)
            else:
                self._join_filters_or(filtered_results)

    def print_certs(self, json_output=False):
        if json_output:
            print(json.dumps([cert for cert in self.all_certs.values()], default=str))
        else:
            for cert in self.all_certs.values():
                try:
                    print(
                        f"""
DomainName: {cert['DomainName']}
SubjectAlternativeNames: { ','.join(sorted(cert['SubjectAlternativeNames'])) }
Issuer: {cert['Issuer']}
Status: {cert['Status']}
ValidationStatus: {cert['ValidationStatus'] if 'ValidationStatus' in cert else '----'}
NotAfter: {cert['NotAfter'] if 'NotAfter' in cert else '----'}
InUse: {','.join(cert['InUseBy']) if cert['InUseBy'] else 'Not in use' }
CertificateArn: {cert['CertificateArn']}
Region: {self.region}
=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- """
                    )
                except KeyError as e:
                    print("Oops \n")
                    print(cert)
                    raise e

    def _filter_identifier(self, identifiers=(), search_certs=None):
        """
        identifier: list
        search_certs: optionally pass in a dict of certs already filtered elsewhere, else filter all certs
        returns list of certs matching identifiers
        """
        filtered_certs = {}
        if search_certs == None:
            search_certs = self.all_certs
        for identifier in identifiers:
            for cert in search_certs.values():
                if cert["CertificateArn"].endswith(f"certificate/{identifier}"):
                    filtered_certs[cert["CertificateArn"]] = cert
        return filtered_certs

    def _filter_san_names(self, san_names=(), search_certs=None):
        """
        san_names: list
        search_certs: optionally pass in a dict of certs already filtered elsewhere, else filter all certs
        returns list of certs that have a SAN name that substring matches any patten
        """
        filtered_certs = {}
        if search_certs == None:
            search_certs = self.all_certs
        for san_name in san_names:
            san_name = san_name.lower()
            for cert in search_certs.values():
                match = False
                for san in cert["SubjectAlternativeNames"]:
                    if san_name in san:
                        match = True
                if match:
                    filtered_certs[cert["CertificateArn"]] = cert
        return filtered_certs

    def _filter_tags(self, tags=(), search_certs=None):
        """
        tags: list of tags
        search_certs: optionally pass in a list of certs already filtered elsewhere, else filter all certs
        returns list of certs matching the AWS tag
        """
        filtered_certs = {}
        if search_certs == None:
            search_certs = self.all_certs
        else:
            search_certs = self.all_certs
        for arn, cert in search_certs.items():
            cert_tags = self.client.list_tags_for_certificate(CertificateArn=arn)
            for tag in tags:
                try:
                    for cert_tag in cert_tags["Tags"]:
                        if cert_tag["Key"] == self.tag_key and cert_tag["Value"] == tag:
                            filtered_certs[cert["CertificateArn"]] = cert
                except (TypeError, KeyError):
                    pass
        return filtered_certs

    def _filter_issuers(self, issuers=(), search_certs=None):
        """
        issuers: list of substring patterns to match
        search_certs: optionally pass in a list of certs already filtered elsewhere, else filter all certs
        returns list of certs matching issuers
        """
        filtered_certs = {}
        if search_certs == None:
            search_certs = self.all_certs
        for issuer in issuers:
            issuer = issuer.lower()
            for cert in search_certs.values():
                if issuer in cert["Issuer"].lower():
                    filtered_certs[cert["CertificateArn"]] = cert
        return filtered_certs

    def _filter_in_use(self, cert_in_use, search_certs=None):
        """
        cert_in_use: bool
        search_certs: optionally pass in a list of certs already filtered elsewhere, else filter all certs
        returns list of certs that are/are not in use
        """
        filtered_certs = {}
        if search_certs == None:
            search_certs = self.all_certs
        if cert_in_use:
            filtered_certs = {
                arn: cert for (arn, cert) in search_certs.items() if cert["InUseBy"]
            }
        else:
            filtered_certs = {
                arn: cert for (arn, cert) in search_certs.items() if not cert["InUseBy"]
            }
        return filtered_certs

    def _join_filters_and(self, filtered_results):
        """
        filtered_results is a list of dicts arn: cert
        """
        self.all_certs = {}
        arns = set.intersection(*[set(r.keys()) for r in filtered_results])
        for cert_list in filtered_results:
            for arn, cert in cert_list.items():
                if arn in arns:
                    self.all_certs[arn] = cert

    def _join_filters_or(self, filtered_results):
        """
        filtered_results is a list of dicts arn: cert
        """
        self.all_certs = {}
        for cert_list in filtered_results:
            for arn, cert in cert_list.items():
                self.all_certs[arn] = cert


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Fetches ACM SSL certificates.")
    parser.add_argument(
        "--region", "-r", default="us-east-1", help='AWS region (default "us-east-1")'
    )
    parser.add_argument(
        "--san-names",
        "-n",
        help="substring patterns to match in SAN hostnames - comma-separate multiple patterns",
    )
    parser.add_argument(
        "--identifiers",
        "-d",
        help="AWS identifier - exact match, comma-separate multiples",
    )
    parser.add_argument(
        "--tags",
        "-t",
        help="dotcms.client.name.short names - exact match, comma-separate multiples",
    )
    parser.add_argument(
        "--issuers",
        "-i",
        help="substring patterns to match in cert issuer - comma-separate multiple patterns",
    )
    parser.add_argument(
        "--cert-in-use",
        "-u",
        type=int,
        choices={0, 1},
        help="0 or 1 -- 'cert in use' filter",
    )
    parser.add_argument(
        "--json", "-j", action="store_true", help="print output as json"
    )
    parser.add_argument(
        "--and-filters",
        "-a",
        action="store_true",
        help="join multiple filters with AND rather than OR",
    )

    parser.add_argument("--debug", action="store_true")

    args = parser.parse_args()

    san_names = []
    if args.san_names:
        san_names = args.san_names.split(",")

    identifiers = []
    if args.identifiers:
        identifiers = args.identifiers.split(",")

    tags = []
    if args.tags:
        tags = args.tags.split(",")

    issuers = []
    if args.issuers:
        issuers = args.issuers.split(",")

    acm = ACM(args.region, args.debug)
    acm.filter_certs(
        san_names=san_names,
        identifiers=identifiers,
        tags=tags,
        issuers=issuers,
        cert_in_use=args.cert_in_use,
        and_filters=args.and_filters,
    )
    acm.print_certs(json_output=args.json)