SSL/TLS Certificate Checker

#!/usr/bin/env python3
"""
SSL/TLS Certificate Checker

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/ssl-cert-checker

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    # No extra packages — uses Python stdlib only
"""

from __future__ import annotations

import argparse
import json
import socket
import ssl
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path

# Signature algorithms and key sizes below current guidance (NIST SP 800-57,
# CA/Browser Forum baseline requirements).
WEAK_SIG = ("md5", "sha1")
MIN_RSA_BITS = 2048
MIN_EC_BITS = 256
WEAK_TLS = ("SSLv2", "SSLv3", "TLSv1", "TLSv1.1")

DEFAULT_PORT = 443


@dataclass
class CertInfo:
    host: str
    port: int = DEFAULT_PORT
    status: str = "unknown"          # ok | warn | critical | error
    not_before: str = ""
    not_after: str = ""
    days_left: int | None = None
    issuer: str = ""
    subject: str = ""
    sans: list[str] = field(default_factory=list)
    serial: str = ""
    sig_algorithm: str = ""
    key_type: str = ""
    key_bits: int | None = None
    tls_version: str = ""
    cipher: str = ""
    trusted: bool | None = None
    trust_error: str = ""
    hostname_ok: bool | None = None
    findings: list[str] = field(default_factory=list)
    error: str = ""


def die(msg: str) -> None:
    """Exit 3 - could not run. A SystemExit with a string would exit 1."""
    print(msg, file=sys.stderr)
    raise SystemExit(3)


def parse_target(raw: str) -> tuple[str, int]:
    """Accept 'host', 'host:port' and '[v6addr]:port'."""
    s = raw.strip()
    if s.startswith("["):
        addr, _, rest = s[1:].partition("]")
        port = int(rest.lstrip(":")) if rest.lstrip(":") else DEFAULT_PORT
        return addr, port
    if s.count(":") == 1:
        host, _, p = s.partition(":")
        try:
            return host, int(p)
        except ValueError:
            return s, DEFAULT_PORT
    return s, DEFAULT_PORT


def load_hosts(args) -> list[tuple[str, int]]:
    raw: list[str] = list(args.hosts)
    if args.hosts_file:
        p = Path(args.hosts_file)
        if not p.exists():
            die(f"Hosts file not found: {args.hosts_file}\n"
                f"Create it with one host per line, for example:\n\n"
                f"  example.com\n  api.company.com\n  mail.corp:993")
        for line in p.read_text(encoding="utf-8").splitlines():
            line = line.split("#", 1)[0].strip()
            if line:
                raw.append(line)
    if not raw:
        die("No hosts given. Pass them as arguments or use --hosts-file.")
    seen: set[tuple[str, int]] = set()
    out: list[tuple[str, int]] = []
    for r in raw:
        t = parse_target(r)
        if t not in seen:
            seen.add(t)
            out.append(t)
    return out


# ---------------------------------------------------------------------------
# Certificate retrieval
# ---------------------------------------------------------------------------

def _name_to_str(name) -> str:
    """Convert the nested tuple form getpeercert() returns into a string."""
    if not name:
        return ""
    parts = []
    for rdn in name:
        for k, v in rdn:
            parts.append(f"{k}={v}")
    return ", ".join(parts)


def fetch_unverified(host: str, port: int, timeout: float) -> tuple[dict, bytes, str, str]:
    """
    Read the certificate WITHOUT verifying it.

    This always succeeds if the host completes a TLS handshake, which is what
    lets the script report the expiry date of a certificate that has already
    expired - the case a verifying context cannot handle.
    """
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    with socket.create_connection((host, port), timeout=timeout) as sock:
        with ctx.wrap_socket(sock, server_hostname=host) as tls:
            der = tls.getpeercert(binary_form=True)
            info = tls.getpeercert()          # empty dict when unverified
            version = tls.version() or ""
            cipher = (tls.cipher() or ("", "", 0))[0]
    return info, der, version, cipher


def verify_trust(host: str, port: int, timeout: float) -> tuple[bool, str, dict]:
    """
    Verify the certificate properly, and report WHY it failed when it does.

    Returns (trusted, reason, cert_dict). The cert dict is populated only on
    success, which is precisely the limitation this script works around by
    also calling fetch_unverified().
    """
    ctx = ssl.create_default_context()
    try:
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with ctx.wrap_socket(sock, server_hostname=host) as tls:
                return True, "", tls.getpeercert() or {}
    except ssl.SSLCertVerificationError as exc:
        return False, (exc.verify_message or str(exc)), {}
    except ssl.SSLError as exc:
        return False, f"TLS error: {exc}", {}
    except (socket.timeout, TimeoutError):
        return False, "timed out during verification", {}
    except OSError as exc:
        return False, f"{exc}", {}


def parse_der(der: bytes) -> dict:
    """
    Pull the fields getpeercert() hides when the certificate is unverified.

    ssl.DER_cert_to_PEM_cert plus the stdlib's own decoder gives subject,
    issuer, validity and SANs without needing the cryptography package.
    """
    out: dict = {}
    if not der:
        return out
    try:
        pem = ssl.DER_cert_to_PEM_cert(der)
    except Exception:  # noqa: BLE001 - malformed cert should not abort the host
        return out

    # _ssl._test_decode_cert needs a file; writing one temporarily is the only
    # stdlib route to a full decode without third-party packages.
    try:
        import tempfile
        import _ssl  # type: ignore[import-not-found]

        with tempfile.NamedTemporaryFile("w", suffix=".pem", delete=False,
                                         encoding="ascii") as fh:
            fh.write(pem)
            tmp = fh.name
        try:
            out = _ssl._test_decode_cert(tmp)
        finally:
            Path(tmp).unlink(missing_ok=True)
    except Exception:  # noqa: BLE001 - fall back to whatever we already have
        return {}
    return out


def _read_len(der: bytes, i: int) -> tuple[int, int]:
    """Decode a DER length at offset i. Returns (length, offset_after_length)."""
    b = der[i]
    if b < 0x80:
        return b, i + 1
    n = b & 0x7F
    return int.from_bytes(der[i + 1:i + 1 + n], "big"), i + 1 + n


def _rsa_modulus_bits(der: bytes) -> int | None:
    """
    Read the real RSA key size out of the certificate.

    Walks to the rsaEncryption OID, steps over the AlgorithmIdentifier to the
    BIT STRING holding RSAPublicKey ::= SEQUENCE { modulus INTEGER, ... } and
    measures the modulus. This reports any key size - 1024, 2048, 8192 - rather
    than only the sizes a lookup table happens to list.
    """
    oid = bytes.fromhex("2a864886f70d010101")
    pos = der.find(oid)
    if pos < 0:
        return None
    i = pos + len(oid)
    # Skip the optional NULL parameters, then find the BIT STRING (0x03).
    while i < len(der) and der[i] != 0x03:
        i += 1
        if i - pos > 32:          # BIT STRING should follow immediately
            return None
    if i >= len(der):
        return None
    _, i = _read_len(der, i + 1)
    if i >= len(der) or der[i] != 0x00:   # BIT STRING unused-bits octet
        return None
    i += 1
    if i >= len(der) or der[i] != 0x30:   # RSAPublicKey SEQUENCE
        return None
    _, i = _read_len(der, i + 1)
    if i >= len(der) or der[i] != 0x02:   # modulus INTEGER
        return None
    mod_len, i = _read_len(der, i + 1)
    if i + mod_len > len(der):
        return None
    mod = der[i:i + mod_len].lstrip(b"\x00")    # drop the DER sign padding
    if not mod:
        return None
    return (len(mod) - 1) * 8 + mod[0].bit_length()


def key_details(der: bytes) -> tuple[str, int | None, str]:
    """
    Best-effort public key type, size and signature algorithm.

    Parsed from the DER structure directly: the stdlib exposes no API for this,
    and requiring the `cryptography` package for a script whose selling point
    is zero dependencies is a poor trade. The markers below are the OIDs as
    they appear in a DER-encoded certificate.
    """
    if not der:
        return "", None, ""

    OIDS = {
        bytes.fromhex("2a864886f70d010105"): "sha1WithRSA",
        bytes.fromhex("2a864886f70d01010b"): "sha256WithRSA",
        bytes.fromhex("2a864886f70d01010c"): "sha384WithRSA",
        bytes.fromhex("2a864886f70d01010d"): "sha512WithRSA",
        bytes.fromhex("2a864886f70d010104"): "md5WithRSA",
        bytes.fromhex("2a8648ce3d040302"): "ecdsaWithSHA256",
        bytes.fromhex("2a8648ce3d040303"): "ecdsaWithSHA384",
    }
    sig = ""
    for oid, name in OIDS.items():
        if oid in der:
            sig = name
            break

    key_type, bits = "", None
    if bytes.fromhex("2a864886f70d010101") in der:           # rsaEncryption
        key_type = "RSA"
        bits = _rsa_modulus_bits(der)
    elif bytes.fromhex("2a8648ce3d0201") in der:             # id-ecPublicKey
        key_type = "EC"
        for curve_oid, size in ((bytes.fromhex("2a8648ce3d030107"), 256),
                                (bytes.fromhex("2b81040022"), 384),
                                (bytes.fromhex("2b81040023"), 521)):
            if curve_oid in der:
                bits = size
                break
    return key_type, bits, sig


# ---------------------------------------------------------------------------
# Per-host check
# ---------------------------------------------------------------------------

def check_host(target: tuple[str, int], args) -> CertInfo:
    host, port = target
    c = CertInfo(host=host, port=port)

    try:
        raw_info, der, version, cipher = fetch_unverified(host, port, args.timeout)
    except socket.gaierror:
        c.status, c.error = "error", "DNS lookup failed - the name does not resolve"
        return c
    except ConnectionRefusedError:
        c.status, c.error = "error", f"connection refused on port {port} - nothing is listening"
        return c
    except (socket.timeout, TimeoutError):
        c.status, c.error = "error", f"timed out connecting to port {port}"
        return c
    except ssl.SSLError as exc:
        c.status, c.error = "error", f"TLS handshake failed: {exc}"
        return c
    except OSError as exc:
        c.status, c.error = "error", f"{type(exc).__name__}: {exc}"
        return c

    c.tls_version, c.cipher = version, cipher

    trusted, reason, verified_info = verify_trust(host, port, args.timeout)
    c.trusted, c.trust_error = trusted, reason

    info = verified_info or raw_info or parse_der(der)
    if not info:
        c.status = "error"
        c.error = "certificate could not be decoded"
        return c

    c.subject = _name_to_str(info.get("subject"))
    c.issuer = _name_to_str(info.get("issuer"))
    c.serial = str(info.get("serialNumber", ""))
    c.sans = [v for k, v in info.get("subjectAltName", ()) if k == "DNS"]
    c.not_before = info.get("notBefore", "")
    c.not_after = info.get("notAfter", "")

    c.key_type, c.key_bits, c.sig_algorithm = key_details(der)

    now = datetime.now(timezone.utc)
    expiry = None
    if c.not_after:
        try:
            expiry = datetime.strptime(c.not_after, "%b %d %H:%M:%S %Y %Z").replace(
                tzinfo=timezone.utc)
            c.days_left = (expiry - now).days
        except ValueError:
            c.findings.append(f"could not parse notAfter '{c.not_after}'")

    if c.not_before:
        try:
            start = datetime.strptime(c.not_before, "%b %d %H:%M:%S %Y %Z").replace(
                tzinfo=timezone.utc)
            if start > now:
                c.findings.append(
                    f"CRITICAL: not valid until {start:%Y-%m-%d} - clock skew, or "
                    f"the certificate was deployed before its start date")
        except ValueError:
            pass

    # Expiry
    if c.days_left is not None:
        if c.days_left < 0:
            c.findings.append(
                f"CRITICAL: EXPIRED {abs(c.days_left)} day(s) ago "
                f"({expiry:%Y-%m-%d})")
        elif c.days_left <= args.critical_days:
            c.findings.append(
                f"CRITICAL: expires in {c.days_left} day(s) ({expiry:%Y-%m-%d})")
        elif c.days_left <= args.warn_days:
            c.findings.append(
                f"WARNING: expires in {c.days_left} day(s) ({expiry:%Y-%m-%d})")

    # Trust - reported separately from expiry, because the reasons differ
    if trusted is False:
        low = reason.lower()
        if "self signed" in low or "self-signed" in low:
            c.findings.append(
                "CRITICAL: self-signed certificate - no chain to a trusted root. "
                "Expected on internal PKI if your corporate root is not installed here")
        elif "hostname mismatch" in low or "doesn't match" in low:
            names = ", ".join(c.sans[:5]) or c.subject
            c.findings.append(
                f"CRITICAL: hostname mismatch - the certificate is valid, but not "
                f"for '{host}'. It covers: {names}")
            c.hostname_ok = False
        elif "expired" in low:
            c.findings.append("CRITICAL: certificate chain contains an expired certificate")
        elif "unable to get local issuer" in low or "unable to verify" in low:
            c.findings.append(
                "CRITICAL: issuer is not trusted here - an intermediate certificate "
                "is missing from the server, or the root is not in this trust store")
        else:
            c.findings.append(f"CRITICAL: trust verification failed - {reason}")
    elif trusted:
        c.hostname_ok = True

    # Crypto strength
    if c.sig_algorithm and any(w in c.sig_algorithm.lower() for w in WEAK_SIG):
        c.findings.append(
            f"WARNING: signed with {c.sig_algorithm} - SHA-1 and MD5 signatures "
            f"are rejected by modern browsers")
    if c.key_type == "RSA" and c.key_bits and c.key_bits < MIN_RSA_BITS:
        c.findings.append(
            f"CRITICAL: {c.key_bits}-bit RSA key - below the {MIN_RSA_BITS}-bit minimum")
    if c.key_type == "EC" and c.key_bits and c.key_bits < MIN_EC_BITS:
        c.findings.append(
            f"WARNING: {c.key_bits}-bit EC key - below the {MIN_EC_BITS}-bit minimum")
    if c.tls_version in WEAK_TLS:
        c.findings.append(
            f"CRITICAL: negotiated {c.tls_version} - deprecated, disable it and "
            f"require TLS 1.2 or 1.3")

    if any(f.startswith("CRITICAL") for f in c.findings):
        c.status = "critical"
    elif any(f.startswith("WARNING") for f in c.findings):
        c.status = "warn"
    else:
        c.status = "ok"
    return c


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------

TAG = {"ok": "[OK]      ", "warn": "[WARN]    ",
       "critical": "[CRITICAL]", "error": "[ERROR]   "}


def render(c: CertInfo, args) -> None:
    label = f"{c.host}:{c.port}" if c.port != DEFAULT_PORT else c.host

    if c.status == "error":
        print(f"{TAG['error']} {label:<34} {c.error}")
        return

    days = f"{c.days_left}d" if c.days_left is not None else "?"
    expiry = c.not_after.split()[0:3]
    exp_s = " ".join(expiry) if expiry else "?"
    print(f"{TAG[c.status]} {label:<34} expires {exp_s:<13} ({days})")

    for f in c.findings:
        print(f"           - {f}")

    if args.detail:
        print(f"           subject    {c.subject or '?'}")
        print(f"           issuer     {c.issuer or '?'}")
        if c.sans:
            shown = ", ".join(c.sans[:6])
            more = f"  (+{len(c.sans) - 6} more)" if len(c.sans) > 6 else ""
            print(f"           SANs       {shown}{more}")
        key = f"{c.key_type}-{c.key_bits}" if c.key_type and c.key_bits else (c.key_type or "?")
        print(f"           key/sig    {key} / {c.sig_algorithm or '?'}")
        print(f"           TLS        {c.tls_version or '?'} / {c.cipher or '?'}")
        print(f"           trusted    {c.trusted}"
              + (f"  ({c.trust_error})" if c.trust_error else ""))


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Check TLS certificate expiry, trust, hostname match and strength.",
        epilog="Exit codes: 0 all good, 1 warnings, 2 critical, 3 could not run.",
    )
    ap.add_argument("hosts", nargs="*",
                    help="Hosts to check. Accepts host or host:port")
    ap.add_argument("--hosts-file",
                    help="File with one host per line. Blank lines and # comments ignored")
    ap.add_argument("--warn-days", type=int, default=30,
                    help="Warn when a certificate expires within this many days. Default: 30")
    ap.add_argument("--critical-days", type=int, default=7,
                    help="Treat expiry within this many days as critical. Default: 7")
    ap.add_argument("--timeout", type=float, default=8.0,
                    help="Seconds to wait per connection. Default: 8")
    ap.add_argument("--detail", action="store_true",
                    help="Print issuer, SANs, key size, signature algorithm and TLS version")
    ap.add_argument("--quiet", action="store_true",
                    help="Only print hosts with findings")
    ap.add_argument("--workers", type=int, default=16,
                    help="How many hosts to check at once. Default: 16")
    ap.add_argument("--json-out", help="Write machine-readable results to this file")
    args = ap.parse_args()

    if args.critical_days > args.warn_days:
        die("--critical-days must not exceed --warn-days.")

    targets = load_hosts(args)
    print(f"Checking {len(targets)} host(s) "
          f"(warn < {args.warn_days}d, critical < {args.critical_days}d)")
    print("=" * 78)

    results: list[CertInfo] = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(check_host, t, args): t for t in targets}
        for fut in as_completed(futures):
            host, port = futures[fut]
            try:
                results.append(fut.result())
            except Exception as exc:  # noqa: BLE001 - one host must not abort the run
                c = CertInfo(host=host, port=port, status="error",
                             error=f"{type(exc).__name__}: {exc}")
                results.append(c)

    order = {"critical": 0, "error": 1, "warn": 2, "ok": 3}
    for c in sorted(results, key=lambda x: (order[x.status], x.host)):
        if args.quiet and c.status == "ok":
            continue
        render(c, args)

    crit = sum(1 for c in results if c.status == "critical")
    warn = sum(1 for c in results if c.status == "warn")
    err = sum(1 for c in results if c.status == "error")
    ok = sum(1 for c in results if c.status == "ok")

    print("=" * 78)
    print(f"{ok} ok, {warn} warning, {crit} critical, {err} unreachable "
          f"(of {len(results)} host(s))")

    soon = [c for c in results if c.days_left is not None and 0 <= c.days_left <= args.warn_days]
    if soon:
        nearest = min(soon, key=lambda c: c.days_left)
        print(f"Nearest expiry: {nearest.host} in {nearest.days_left} day(s)")

    if args.json_out:
        Path(args.json_out).write_text(
            json.dumps([asdict(c) for c in results], indent=2), encoding="utf-8")
        print(f"Results written to {args.json_out}")

    if crit:
        return 2
    if warn or err:
        return 1
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)