certificate-inventory.py

#!/usr/bin/env python3
"""Inventory and verify TLS certificates for multiple endpoints."""

import argparse
import csv
import datetime as dt
import json
import socket
import ssl
import sys
from pathlib import Path


def parse_target(value):
    if value.startswith("["):
        closing = value.find("]")
        if closing < 0:
            raise ValueError("invalid bracketed IPv6 address")
        host = value[1:closing]
        port = int(value[closing + 2:]) if value[closing + 1:].startswith(":") else 443
    elif value.count(":") == 1:
        host, raw_port = value.rsplit(":", 1)
        port = int(raw_port)
    elif ":" in value:
        host, port = value, 443
    else:
        host, port = value, 443
    if not host or not 1 <= port <= 65535:
        raise ValueError("invalid host or port")
    return host, port


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("targets", nargs="*")
    parser.add_argument("--file", type=Path, help="One HOST[:PORT] per line")
    parser.add_argument("--timeout", type=float, default=5)
    parser.add_argument("--json", action="store_true", dest="as_json")
    return parser.parse_args()


def flatten_name(items):
    return ", ".join("=".join(attribute) for group in items for attribute in group)


def inspect(target, timeout):
    row = {"target": target, "host": "", "port": None, "verified": False, "subject": "",
           "issuer": "", "serial": "", "not_before": "", "not_after": "", "days_remaining": None,
           "sans": "", "tls_version": "", "cipher": "", "error": ""}
    try:
        host, port = parse_target(target)
        row.update(host=host, port=port)
        context = ssl.create_default_context()
        with socket.create_connection((host, port), timeout=timeout) as connection:
            with context.wrap_socket(connection, server_hostname=host) as tls:
                cert = tls.getpeercert()
                expiry = dt.datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z").replace(tzinfo=dt.timezone.utc)
                row.update(
                    verified=True,
                    subject=flatten_name(cert.get("subject", ())),
                    issuer=flatten_name(cert.get("issuer", ())),
                    serial=cert.get("serialNumber", ""),
                    not_before=cert.get("notBefore", ""),
                    not_after=cert.get("notAfter", ""),
                    days_remaining=(expiry - dt.datetime.now(dt.timezone.utc)).days,
                    sans=", ".join(value for kind, value in cert.get("subjectAltName", ()) if kind in {"DNS", "IP Address"}),
                    tls_version=tls.version() or "",
                    cipher=(tls.cipher() or ("",))[0],
                )
    except (OSError, ssl.SSLError, ValueError, KeyError) as error:
        row["error"] = str(error)
    return row


def main():
    args = parse_args()
    if args.timeout <= 0:
        raise SystemExit("Timeout must be positive")
    targets = list(args.targets)
    if args.file:
        targets.extend(line.strip() for line in args.file.read_text(encoding="utf-8").splitlines()
                       if line.strip() and not line.lstrip().startswith("#"))
    if not targets:
        raise SystemExit("Provide targets or --file")
    rows = [inspect(target, args.timeout) for target in dict.fromkeys(targets)]
    if args.as_json:
        print(json.dumps(rows, indent=2))
    else:
        writer = csv.DictWriter(sys.stdout, fieldnames=list(rows[0]))
        writer.writeheader(); writer.writerows(rows)
    return 1 if any(not row["verified"] for row in rows) else 0


if __name__ == "__main__":
    raise SystemExit(main())