Certificate Expiry Monitor

Certificate Expiry Monitor

Problem Statement

Expired TLS certificates are one of the most common causes of production outages — and one of the most preventable. A certificate that silently expires at 3 AM can take down your API, break your mobile app, or make your website unreachable for every user. The problem is that certificates are usually valid for 90 days or more, so they're easy to forget about until the moment they fail.

This script solves that problem by acting as a lightweight, self-contained monitoring agent. It connects to each host in a list, retrieves the actual X.509 certificate presented by the server, and reports any certificate that is expiring within a configurable warning window. Alerts can be pushed to any ntfy topic (or a self-hosted ntfy instance), and the exit code reflects the health of the fleet — making it trivial to plug into cron, systemd timers, or a CI pipeline.

How It Works

The script follows a simple fetch → parse → evaluate → notify pipeline. For each host it opens a raw TCP socket (with a configurable timeout), wraps it in an ssl context, and performs a TLS handshake. The peer certificate is extracted in DER form and parsed with the cryptography library, which gives us a structured x509.Certificate object with not_valid_after_utc and subject/SAN fields.

Design decisions worth noting:

• Socket-level TLS, not requests. We use requests only for the ntfy HTTP POST. Certificate inspection is done at the socket layer so the script works for any TLS service (SMTP, IMAP, LDAP, databases) — not just HTTPS. The requests dependency is therefore reserved strictly for alerting.
• Fail-soft per host. A single unreachable host must not abort the entire run. Each check is wrapped in a try/except and its result is recorded independently.
• Deterministic exit codes. Exit 0 = all healthy, 1 = at least one warning, 2 = at least one critical/error. This makes the script composable with shell logic and monitoring systems.
• Dataclass-based results. A CertResult dataclass keeps the per-host state typed and readable.

Full Script

#!/usr/bin/env python3
"""
Certificate Expiry Monitor

Connects to a list of TLS endpoints, inspects the certificate each presents,
and reports any that are expiring within a configurable warning window.
Optionally pushes a summary to an ntfy topic.

Usage:
    python3 cert_monitor.py --hosts hosts.txt --warn-days 30 \
        --ntfy-url https://ntfy.sh/alerts
"""

from __future__ import annotations

import argparse
import os
import socket
import ssl
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable

import requests
from cryptography import x509
from cryptography.x509.oid import NameOID

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_TIMEOUT = float(os.environ.get("CERT_MONITOR_TIMEOUT", "10"))
DEFAULT_WARN_DAYS = int(os.environ.get("CERT_MONITOR_WARN_DAYS", "30"))
DEFAULT_NTFY_URL = os.environ.get("CERT_MONITOR_NTFY_URL", "")  # empty = disabled
DEFAULT_NTFY_TOKEN = os.environ.get("CERT_MONITOR_NTFY_TOKEN", "")


# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------
@dataclass
class CertResult:
    """Outcome of inspecting one host's certificate."""
    host: str
    port: int
    ok: bool
    days_left: float | None = None
    not_after: datetime | None = None
    subject: str = ""
    error: str = ""
    severity: str = "ok"  # one of: ok, warning, critical, error

    @property
    def label(self) -> str:
        return f"{self.host}:{self.port}"


# ---------------------------------------------------------------------------
# Core logic
# ---------------------------------------------------------------------------
def parse_host_line(line: str) -> tuple[str, int] | None:
    """Parse a line like 'example.com', 'example.com:8443', or '#comment'."""
    line = line.strip()
    if not line or line.startswith("#"):
        return None
    if ":" in line:
        host, _, port_str = line.rpartition(":")
        try:
            return host.strip(), int(port_str)
        except ValueError:
            # Not a port — fall back to default (e.g. IPv6 without brackets)
            return line, 443
    return line, 443


def load_hosts(path: Path) -> list[tuple[str, int]]:
    """Read host:port pairs from a file, skipping blanks and comments."""
    if not path.exists():
        raise FileNotFoundError(f"Hosts file not found: {path}")
    hosts: list[tuple[str, int]] = []
    for raw in path.read_text(encoding="utf-8").splitlines():
        parsed = parse_host_line(raw)
        if parsed:
            hosts.append(parsed)
    if not hosts:
        raise ValueError(f"No valid hosts found in {path}")
    return hosts


def fetch_certificate(host: str, port: int, timeout: float) -> x509.Certificate:
    """
    Open a TLS connection and return the parsed peer certificate.

    Uses the default system trust store only to complete the handshake;
    we deliberately do NOT verify the chain here because we want to inspect
    the cert even if it is expired or self-signed. Verification is a
    separate concern handled by other tooling.
    """
    ctx = ssl.create_default_context()
    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)
            if not der:
                raise ssl.SSLError("Server did not present a certificate")
            return x509.load_der_x509_certificate(der)


def format_subject(cert: x509.Certificate) -> str:
    """Return a compact human-readable subject (CN if present, else full RDN)."""
    cn_attrs = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
    if cn_attrs:
        return str(cn_attrs[0].value)
    return cert.subject.rfc4514_string()


def check_host(host: str, port: int, warn_days: int, timeout: float) -> CertResult:
    """Inspect one host and classify its certificate's expiry status."""
    try:
        cert = fetch_certificate(host, port, timeout)
    except (socket.timeout, socket.gaierror, ConnectionRefusedError,
            ssl.SSLError, OSError) as exc:
        return CertResult(
            host=host, port=port, ok=False,
            error=f"{type(exc).__name__}: {exc}",
            severity="error",
        )

    not_after = cert.not_valid_after_utc
    now = datetime.now(timezone.utc)
    days_left = (not_after - now).total_seconds() / 86400.0

    if days_left < 0:
        severity = "critical"
    elif days_left <= warn_days:
        severity = "warning"
    else:
        severity = "ok"

    return CertResult(
        host=host, port=port, ok=True,
        days_left=days_left, not_after=not_after,
        subject=format_subject(cert), severity=severity,
    )


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------
def summarize(results: Iterable[CertResult]) -> tuple[str, int]:
    """Build a plain-text report and compute the process exit code."""
    lines: list[str] = []
    worst = 0  # 0 = ok, 1 = warning, 2 = error/critical

    for r in results:
        if r.severity == "error":
            lines.append(f"[ERROR]    {r.label:<40} {r.error}")
            worst = max(worst, 2)
        elif r.severity == "critical":
            lines.append(
                f"[CRITICAL] {r.label:<40} EXPIRED "
                f"{abs(r.days_left):.1f} days ago (subject: {r.subject})"
            )
            worst = max(worst, 2)
        elif r.severity == "warning":
            lines.append(
                f"[WARNING]  {r.label:<40} expires in {r.days_left:.1f} days "
                f"on {r.not_after:%Y-%m-%d} (subject: {r.subject})"
            )
            worst = max(worst, 1)
        else:
            lines.append(
                f"[OK]       {r.label:<40} {r.days_left:.1f} days left"
            )

    return "\n".join(lines), worst


def send_ntfy(url: str, token: str, title: str, body: str,
              priority: str = "default") -> None:
    """POST a notification to an ntfy topic. Silently no-ops if url is empty."""
    if not url:
        return
    headers = {
        "Title": title,
        "Priority": priority,
        "Tags": "warning,lock",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        # ntfy accepts the message body as the raw POST payload.
        resp = requests.post(url, data=body.encode("utf-8"),
                             headers=headers, timeout=10)
        resp.raise_for_status()
    except requests.RequestException as exc:
        # Alerting failure must not mask the actual certificate findings.
        print(f"WARN: failed to send ntfy notification: {exc}", file=sys.stderr)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Monitor TLS certificate expiry across a fleet of hosts.",
    )
    p.add_argument("--hosts", required=True, type=Path,
                   help="Path to a file containing one host[:port] per line.")
    p.add_argument("--warn-days", type=int, default=DEFAULT_WARN_DAYS,
                   help=f"Warn when fewer than N days remain (default: {DEFAULT_WARN_DAYS}).")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
                   help=f"Per-host connection timeout in seconds (default: {DEFAULT_TIMEOUT}).")
    p.add_argument("--ntfy-url", default=DEFAULT_NTFY_URL,
                   help="ntfy topic URL to POST alerts to. Omit to disable.")
    p.add_argument("--ntfy-token", default=DEFAULT_NTFY_TOKEN,
                   help="Optional Bearer token for protected ntfy topics.")
    p.add_argument("--quiet", action="store_true",
                   help="Suppress stdout; rely on exit code and ntfy only.")
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    try:
        hosts = load_hosts(args.hosts)
    except (FileNotFoundError, ValueError) as exc:
        print(f"FATAL: {exc}", file=sys.stderr)
        return 2

    results = [check_host(h, p, args.warn_days, args.timeout) for h, p in hosts]
    report, worst = summarize(results)

    if not args.quiet:
        print(report)

    # Only notify when something is actually wrong.
    if worst >= 1:
        title = "Certificate expiry alert"
        priority = "urgent" if worst == 2 else "high"
        send_ntfy(args.ntfy_url, args.ntfy_token, title, report, priority)

    return worst


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

Configuration

Usage Examples

1. Basic check against a small hosts file

$ cat hosts.txt
example.com
api.github.com:443
smtp.gmail.com:465

$ python3 cert_monitor.py --hosts hosts.txt --warn-days 30
[OK]       example.com:443                          71.2 days left
[OK]       api.github.com:443                       58.4 days left
[WARNING]  smtp.gmail.com:465                       expires in 21.3 days on 2024-11-02 (subject: smtp.gmail.com)
$ echo $?
1

2. Push alerts to ntfy when something is expiring

$ python3 cert_monitor.py --hosts hosts.txt --warn-days 45 \
    --ntfy-url https://ntfy.sh/my-alerts
# ... report printed to stdout, plus a push notification titled
# "Certificate expiry alert" with priority "high".

3. Cron-friendly quiet mode with a protected ntfy topic

$ CERT_MONITOR_NTFY_TOKEN=tok_abc123 \
  python3 cert_monitor.py --hosts /etc/cert-monitor/hosts.txt \
    --warn-days 14 --quiet --ntfy-url https://ntfy.example.com/ops
$ echo $?
0   # all certs healthy; no notification sent

Error Handling

• DNS / connection failures (socket.gaierror, ConnectionRefusedError, socket.timeout) are caught per host and reported as [ERROR] entries. They set the exit code to 2 but do not stop the run.
• TLS handshake failures (ssl.SSLError) — including servers that refuse the handshake or present no certificate — are handled the same way.
• Missing or empty hosts file raises FileNotFoundError/ValueError and exits with code 2 before any network activity.
• ntfy delivery failures are logged to stderr but deliberately do not change the exit code, so a flaky alerting endpoint can't mask the real certificate status.
• Unparseable DER from cryptography would raise ValueError; in practice ssl rejects malformed certs during the handshake, so this path is effectively covered by the SSLError branch.

Extending Ideas

• Chain validation: add a --verify flag that uses ctx.verify_mode = ssl.CERT_REQUIRED and a custom CA bundle, so you can also detect untrusted or mismatched issuers.
• Structured output: emit JSON (--format json) with one object per host so results can be ingested by Prometheus textfile collector, Loki, or a custom dashboard.
• Concurrency: swap the sequential loop for a concurrent.futures.ThreadPoolExecutor (or asyncio with asyncio.open_connection) to scan hundreds of hosts without hitting the per-host timeout serially.
• Scheduling & dedup: add a small state file that records the last alert date per host, and only re-notify every N hours — turning the script from a stateless checker into a proper alerting daemon.