cleveragents-core/scripts/check-tls-cert.py

#!/usr/bin/env python3
"""TLS certificate health-check script for CleverAgents infrastructure.

Verifies that a given hostname's TLS certificate:
  - Is reachable via HTTPS
  - Presents a valid certificate trusted by the system CA bundle
  - Includes the target hostname as a Subject Alternative Name (SAN)
  - Has not expired and will not expire within the warning threshold

Usage:
    python scripts/check-tls-cert.py git.dev.cleveragents.com
    python scripts/check-tls-cert.py git.dev.cleveragents.com --port 443
    python scripts/check-tls-cert.py git.dev.cleveragents.com --warn-days 30
    python scripts/check-tls-cert.py git.dev.cleveragents.com git.cleveragents.com

Exit codes:
    0  All checks passed
    1  One or more checks failed (certificate error, expiry, SAN mismatch)
    2  Usage error (bad arguments)

This script is used by the ops team to diagnose TLS issues and by CI to
monitor certificate health.  See docs/development/ops-runbook.md for the
full certificate renewal procedure.
"""

from __future__ import annotations

import argparse
import datetime
import socket
import ssl
import sys
from collections.abc import Sequence
from dataclasses import dataclass, field

# ---------------------------------------------------------------------------
# Data types
# ---------------------------------------------------------------------------


@dataclass
class CertCheckResult:
    """Result of a single TLS certificate check for one hostname."""

    hostname: str
    port: int
    ok: bool = True
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    subject_cn: str = ""
    sans: list[str] = field(default_factory=list)
    not_after: datetime.datetime | None = None
    days_remaining: int | None = None

    def add_error(self, msg: str) -> None:
        """Record a fatal error and mark the result as failed."""
        self.errors.append(msg)
        self.ok = False

    def add_warning(self, msg: str) -> None:
        """Record a non-fatal warning (does not affect ok status)."""
        self.warnings.append(msg)


# ---------------------------------------------------------------------------
# Certificate inspection helpers
# ---------------------------------------------------------------------------


def _parse_not_after(cert: dict) -> datetime.datetime | None:
    """Parse the ``notAfter`` field from a DER-decoded certificate dict.

    Returns a timezone-aware UTC datetime, or ``None`` if the field is
    absent or cannot be parsed.
    """
    not_after_str: str | None = cert.get("notAfter")
    if not not_after_str:
        return None
    try:
        # ssl module returns strings like "Apr  2 23:59:59 2026 GMT"
        dt = datetime.datetime.strptime(not_after_str, "%b %d %H:%M:%S %Y %Z")
        return dt.replace(tzinfo=datetime.UTC)
    except ValueError:
        return None


def _extract_sans(cert: dict) -> list[str]:
    """Extract Subject Alternative Names from a DER-decoded certificate dict.

    Returns a list of SAN values (e.g. ``["git.dev.cleveragents.com",
    "git.cleveragents.com"]``).  DNS SANs are returned without the
    ``DNS:`` prefix.
    """
    sans: list[str] = []
    for san_type, san_value in cert.get("subjectAltName", ()):
        if san_type == "DNS":
            sans.append(san_value)
    return sans


def _extract_cn(cert: dict) -> str:
    """Extract the Common Name (CN) from the certificate subject."""
    for rdn in cert.get("subject", ()):
        for attr_type, attr_value in rdn:
            if attr_type == "commonName":
                return attr_value
    return ""


def _hostname_matches_san(hostname: str, sans: list[str]) -> bool:
    """Return True if *hostname* matches any SAN entry.

    Supports exact matches and single-level wildcard patterns
    (e.g. ``*.cleveragents.com`` matches ``git.cleveragents.com`` but
    not ``a.b.cleveragents.com``).
    """
    hostname_lower = hostname.lower()
    for san in sans:
        san_lower = san.lower()
        if san_lower == hostname_lower:
            return True
        if san_lower.startswith("*."):
            # Wildcard: *.example.com matches foo.example.com only
            suffix = san_lower[1:]  # ".example.com"
            if (
                hostname_lower.endswith(suffix)
                and "." not in hostname_lower[: -len(suffix)]
            ):
                return True
    return False


# ---------------------------------------------------------------------------
# Core check function
# ---------------------------------------------------------------------------


def check_tls_certificate(
    hostname: str,
    port: int = 443,
    warn_days: int = 30,
    timeout: float = 10.0,
    ssl_context: ssl.SSLContext | None = None,
) -> CertCheckResult:
    """Perform a full TLS certificate check for *hostname*:*port*.

    Args:
        hostname:    The hostname to connect to and verify.
        port:        TCP port (default 443).
        warn_days:   Warn if the certificate expires within this many days.
        timeout:     Socket connection timeout in seconds.
        ssl_context: Optional custom SSLContext (used in tests to inject
                     mock behaviour).  When ``None``, a default context
                     with system CA verification is used.

    Returns:
        A :class:`CertCheckResult` describing the outcome.
    """
    result = CertCheckResult(hostname=hostname, port=port)

    # Build default SSL context if none provided
    ctx = ssl.create_default_context() if ssl_context is None else ssl_context

    # --- Attempt TLS handshake ---
    try:
        with (
            socket.create_connection((hostname, port), timeout=timeout) as raw_sock,
            ctx.wrap_socket(raw_sock, server_hostname=hostname) as tls_sock,
        ):
            cert = tls_sock.getpeercert()
    except ssl.SSLCertVerificationError as exc:
        reason = getattr(exc, "reason", None) or str(exc)
        result.add_error(f"TLS verification failed: {reason}")
        return result
    except ssl.SSLError as exc:
        result.add_error(f"TLS handshake error: {exc}")
        return result
    except TimeoutError:
        result.add_error(f"Connection timed out after {timeout}s")
        return result
    except OSError as exc:
        result.add_error(f"Connection failed: {exc}")
        return result

    if cert is None:
        result.add_error("No certificate returned by server")
        return result

    # --- Parse certificate fields ---
    result.subject_cn = _extract_cn(cert)
    result.sans = _extract_sans(cert)
    result.not_after = _parse_not_after(cert)

    # --- SAN check ---
    if not _hostname_matches_san(hostname, result.sans):
        result.add_error(
            f"Hostname '{hostname}' not found in certificate SANs: "
            f"{result.sans!r}. The certificate may be missing this hostname "
            "or SNI routing is misconfigured."
        )

    # --- Expiry check ---
    if result.not_after is not None:
        now = datetime.datetime.now(tz=datetime.UTC)
        delta = result.not_after - now
        result.days_remaining = delta.days
        if delta.total_seconds() <= 0:
            expiry = result.not_after.strftime("%Y-%m-%d %H:%M:%S UTC")
            result.add_error(f"Certificate expired on {expiry}")
        elif delta.days <= warn_days:
            expiry_date = result.not_after.strftime("%Y-%m-%d")
            result.add_warning(
                f"Certificate expires in {delta.days} day(s) "
                f"({expiry_date}). "
                f"Renewal recommended (threshold: {warn_days} days)."
            )
    else:
        result.add_warning("Could not determine certificate expiry date")

    return result


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------


def _print_result(result: CertCheckResult) -> None:
    """Print a human-readable summary of a :class:`CertCheckResult`."""
    status = "OK" if result.ok else "FAIL"
    print(f"\n[{status}] {result.hostname}:{result.port}")
    if result.subject_cn:
        print(f"  Subject CN : {result.subject_cn}")
    if result.sans:
        print(f"  SANs       : {', '.join(result.sans)}")
    if result.not_after:
        days = result.days_remaining
        expiry = result.not_after.strftime("%Y-%m-%d %H:%M:%S UTC")
        print(f"  Expires    : {expiry} ({days} day(s) remaining)")
    for err in result.errors:
        print(f"  ERROR      : {err}")
    for warn in result.warnings:
        print(f"  WARNING    : {warn}")


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------


def main(argv: Sequence[str] | None = None) -> int:
    """Parse arguments and run TLS checks.  Returns the process exit code."""
    parser = argparse.ArgumentParser(
        description="Check TLS certificate validity for one or more hostnames.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument(
        "hostnames",
        nargs="+",
        metavar="HOSTNAME",
        help="One or more hostnames to check (e.g. git.dev.cleveragents.com)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=443,
        help="HTTPS port to connect to (default: 443)",
    )
    parser.add_argument(
        "--warn-days",
        type=int,
        default=30,
        metavar="DAYS",
        help="Warn if certificate expires within DAYS days (default: 30)",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=10.0,
        metavar="SECONDS",
        help="Connection timeout in seconds (default: 10)",
    )

    args = parser.parse_args(argv)

    all_ok = True
    for hostname in args.hostnames:
        result = check_tls_certificate(
            hostname=hostname,
            port=args.port,
            warn_days=args.warn_days,
            timeout=args.timeout,
        )
        _print_result(result)
        if not result.ok:
            all_ok = False

    print()
    if all_ok:
        print("All TLS checks PASSED")
        return 0
    else:
        print("One or more TLS checks FAILED")
        return 1


if __name__ == "__main__":
    sys.exit(main())