Certificate Expiration Tracker

check-certificates.py

#!/usr/bin/python3
# TLS Certificate Checker
# Ingest a list of hosts, and return a report (YAML) with expiration dates
# Nicholas Schmidt

# Imports

# Plaintext
from ruamel.yaml import YAML
import json

# Arguments
import argparse

# System
import sys
from datetime import datetime

# TLS
import ssl
import OpenSSL

# DNS
from fqdn import FQDN


# Function: Get the expiration date on common ports
def get_tls_expiry(get_tls_expiry_hostname, port=None):
    if FQDN(get_tls_expiry_hostname).is_valid:
        tls_cert = ssl.get_server_certificate((get_tls_expiry_hostname, port))
        x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, tls_cert)
        datetime.strptime(x509.get_notAfter().decode("ascii"), "%Y%m%d%H%M%SZ")
        return (
            datetime.now(),
            datetime.strptime(x509.get_notBefore().decode("ascii"), "%Y%m%d%H%M%SZ"),
            datetime.strptime(x509.get_notAfter().decode("ascii"), "%Y%m%d%H%M%SZ"),
            str(x509.get_subject().CN),
        )
    else:
        sys.exit(
            "Invalid hostname passed to resolver function: " + get_tls_expiry_hostname
        )


# Function: Determine if we're in a Warn/Error/OK state
def get_date_expiration_range(get_date_expiration_range_tuple):
    if type(get_date_expiration_range_tuple) is tuple:
        work_dict = {}
        # This will create date type objects (for now) that we need to correctly test against
        work_dict["from_issue_date"] = (
            get_date_expiration_range_tuple[0] - get_date_expiration_range_tuple[1]
        )
        work_dict["to_expiration_date"] = (
            get_date_expiration_range_tuple[2] - get_date_expiration_range_tuple[0]
        )
        work_dict["alarms"] = []
        # Checks go here
        # Verify past issue date
        if work_dict["from_issue_date"].days <= 0:
            work_dict["alarms"].append(
                "E000: Certificate issue date has not been reached!"
            )
        # Verify Expiration
        # Increasing severity alarms based on expiration date. Include Certifate ID
        if work_dict["to_expiration_date"].days <= 0:
            work_dict["alarms"].append(
                "C999: Certificate " + get_date_expiration_range_tuple[3] + " expired!"
            )
        if work_dict["to_expiration_date"].days <= 30:
            work_dict["alarms"].append(
                "E998: Certificate "
                + get_date_expiration_range_tuple[3]
                + " expires in "
                + str(work_dict["to_expiration_date"].days)
                + " days!"
            )
        if work_dict["to_expiration_date"].days <= 60:
            work_dict["alarms"].append(
                "W997: Certificate "
                + get_date_expiration_range_tuple[3]
                + " expires in "
                + str(work_dict["to_expiration_date"].days)
                + " days!"
            )
        if work_dict["to_expiration_date"].days <= 90:
            work_dict["alarms"].append(
                "I996: Certificate "
                + get_date_expiration_range_tuple[3]
                + " expires in "
                + str(work_dict["to_expiration_date"].days)
                + " days!"
            )
        # Finally, change to strings so that YAML can handle it
        work_dict["from_issue_date"] = str(work_dict["from_issue_date"])
        work_dict["to_expiration_date"] = str(work_dict["to_expiration_date"])
        return work_dict
    else:
        sys.exit(
            "Invalid value submitted to date expiration ranger: "
            + str(get_date_expiration_range_tuple)
        )


# Arguments Parsing
parser = argparse.ArgumentParser(description="TLS Expiration Fetcher")
parser.add_argument("-e", help="Host:Port notation(Individual)")
parser.add_argument("-f", help="JSON List (Batch)")
args = parser.parse_args()

# We need something to work on, either way
if not args.e and not args.f:
    sys.exit("no valid input found!")

if args.e:
    # This mode will accept a split set:
    # {{ hostname }}:{{ port }}
    # and test it
    target = args.e.split(":")
    if len(target) == 2 and FQDN(target[0]).is_valid:
        print(
            get_date_expiration_range(
                get_tls_expiry(get_tls_expiry_hostname=target[0], port=target[1])
            )
        )
    else:
        sys.exit("No valid FQDN or port found! " + json.dumps(target))
elif args.f:
    # Load the file
    # format should be a `list`` comprised of `dict`:
    # { "fqdn": "name", "port": "443" }
    work_list = []
    result_list = {}
    yaml = YAML(typ="safe")
    try:
        with open(args.f, "r") as json_file:
            work_list = json.load(json_file)
    except FileNotFoundError as f:
        sys.exit("File " + args.f + " not found! Error: " + str(f))
    except Exception as e:
        sys.exit("Unhandled exception found! " + str(e))
    # Try to access keys
    try:
        for i in work_list:
            result_list[i["fqdn"]] = {}
            result_list[i["fqdn"]]["port"] = i["port"]
            result_list[i["fqdn"]] = get_date_expiration_range(
                get_tls_expiry(i["fqdn"], port=i["port"])
            )
    except KeyError as e:
        sys.exit("Potential JSON formatting error found! " + str(e))
    except Exception as e:
        sys.exit("Unhandled worklist exception found! " + str(e))
    # Finally, dump it
    yaml.dump(result_list, sys.stdout)
    # Send Alarms to stderr. Aggregate all alarms at the end
    error_list = []
    print(json.dumps(result_list, indent=4))
    for i in result_list:
        if len(result_list[i]["alarms"]) > 0:
            for ii in result_list[i]["alarms"]:
                print(ii)
                error_list.append(ii)
    yaml.dump(error_list, sys.stderr)
    # Let's give Jenkins a hint that there are problems
    if len(error_list) > 0:
        sys.exit("Certificates are expiring soon! " + json.dumps(error_list))

gistfile1.txt

[
    { "fqdn": "vcenter.engyak.co", "port": 443 }
]