# /// script # requires-python = ">=3.10" # dependencies = ["cryptography>=42"] # /// """Scan TLS endpoints and report every certificate's issuer, SANs and days to expiry. Usage: uv run cert_inventory.py targets.txt uv run cert_inventory.py targets.txt --warn-days 30 --csv inventory.csv --json inventory.json targets.txt holds one endpoint per line: host, host:port, or host:port,servername. Blank lines and lines starting with # are ignored. Port defaults to 443. The scan deliberately skips trust verification so that self-signed, expired and private-CA certificates are reported too; it is an inventory tool, not a validator. Exit code is 0 when everything is healthy, 1 when any certificate is inside the warning window or expired, and 2 when any endpoint could not be reached. """ from __future__ import annotations import argparse import csv import json import socket import ssl import sys from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from cryptography import x509 from cryptography.x509.oid import ExtensionOID, NameOID @dataclass class Target: host: str port: int = 443 servername: str | None = None @property def label(self) -> str: return f"{self.host}:{self.port}" @dataclass class CertResult: endpoint: str status: str subject_cn: str = "" issuer: str = "" sans: list[str] = field(default_factory=list) not_after: str = "" days_left: int | None = None validity_days: int | None = None self_signed: bool = False error: str = "" def parse_targets(path: Path) -> list[Target]: targets: list[Target] = [] for raw in path.read_text().splitlines(): line = raw.strip() if not line or line.startswith("#"): continue endpoint, _, servername = line.partition(",") host, _, port = endpoint.strip().rpartition(":") if ":" in endpoint else (endpoint.strip(), "", "443") targets.append(Target(host=host, port=int(port or 443), servername=servername.strip() or None)) return targets def fetch_certificate(target: Target, timeout: float) -> x509.Certificate: context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE server_hostname = target.servername or target.host with socket.create_connection((target.host, target.port), timeout=timeout) as sock: with context.wrap_socket(sock, server_hostname=server_hostname) as tls: der = tls.getpeercert(binary_form=True) if der is None: raise ssl.SSLError("server presented no certificate") return x509.load_der_x509_certificate(der) def first_attr(name: x509.Name, oid) -> str: attrs = name.get_attributes_for_oid(oid) return str(attrs[0].value) if attrs else "" def inspect(target: Target, warn_days: int, timeout: float) -> CertResult: try: cert = fetch_certificate(target, timeout) except (OSError, ssl.SSLError, ValueError) as exc: return CertResult(endpoint=target.label, status="UNREACHABLE", error=str(exc)) try: san_ext = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME) sans = san_ext.value.get_values_for_type(x509.DNSName) except x509.ExtensionNotFound: sans = [] not_before = cert.not_valid_before_utc not_after = cert.not_valid_after_utc days_left = (not_after - datetime.now(timezone.utc)).days issuer_label = first_attr(cert.issuer, NameOID.ORGANIZATION_NAME) or first_attr(cert.issuer, NameOID.COMMON_NAME) if days_left < 0: status = "EXPIRED" elif days_left <= warn_days: status = "RENEW_SOON" else: status = "OK" return CertResult( endpoint=target.label, status=status, subject_cn=first_attr(cert.subject, NameOID.COMMON_NAME), issuer=issuer_label, sans=sans, not_after=not_after.date().isoformat(), days_left=days_left, validity_days=(not_after - not_before).days, self_signed=cert.issuer == cert.subject, ) def scan(targets: list[Target], warn_days: int, timeout: float, workers: int) -> list[CertResult]: with ThreadPoolExecutor(max_workers=workers) as pool: results = list(pool.map(lambda t: inspect(t, warn_days, timeout), targets)) order = {"EXPIRED": 0, "UNREACHABLE": 1, "RENEW_SOON": 2, "OK": 3} return sorted(results, key=lambda r: (order[r.status], r.days_left if r.days_left is not None else 0)) def print_table(results: list[CertResult]) -> None: header = f"{'STATUS':<12} {'DAYS':>5} {'EXPIRES':<11} {'ENDPOINT':<32} {'ISSUER':<28} SUBJECT" print(header) print("-" * len(header)) for r in results: days = "" if r.days_left is None else str(r.days_left) detail = r.error if r.error else r.subject_cn + (" (self-signed)" if r.self_signed else "") print(f"{r.status:<12} {days:>5} {r.not_after:<11} {r.endpoint:<32} {r.issuer[:28]:<28} {detail}") def write_csv(results: list[CertResult], path: Path) -> None: with path.open("w", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(asdict(results[0]).keys())) writer.writeheader() for r in results: row = asdict(r) row["sans"] = " ".join(r.sans) writer.writerow(row) def main() -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("targets", type=Path) parser.add_argument("--warn-days", type=int, default=30) parser.add_argument("--timeout", type=float, default=5.0) parser.add_argument("--workers", type=int, default=20) parser.add_argument("--csv", type=Path) parser.add_argument("--json", type=Path) args = parser.parse_args() targets = parse_targets(args.targets) if not targets: print("No targets found.", file=sys.stderr) return 2 results = scan(targets, args.warn_days, args.timeout, args.workers) print_table(results) if args.csv: write_csv(results, args.csv) if args.json: args.json.write_text(json.dumps([asdict(r) for r in results], indent=2)) statuses = {r.status for r in results} if "UNREACHABLE" in statuses: return 2 if statuses & {"EXPIRED", "RENEW_SOON"}: return 1 return 0 if __name__ == "__main__": sys.exit(main())