How to Monitor Client Domains and SSL Certificates for Free Before They Expire

clients: - name: Acme Traders domain: acmetraders.com # the registered domain, not a subdomain sites: - url: https://acmetraders.com expect: "Acme Traders" # text that must appear on the page - url: https://www.acmetraders.com - url: https://app.acmetraders.com/login expect: "Sign in" - name: Example Studio domain: examplestudio.xyz domain_expires: 2027-03-14 # optional hand-entered date, see Step 3 sites: - url: https://examplestudio.xyz 

from __future__ import annotations import json import os import re import socket import ssl import sys import time import urllib.error import urllib.request from datetime import date, datetime, timezone from urllib.parse import urlencode, urlparse import yaml TIMEOUT = 15 HEADERS = {"User-Agent": "client-watchdog/1.0"} # ---------- Check 1: SSL certificate expiry ----------  def ssl_days_left(host: str) -> int: """Days until the certificate served for `host` expires.""" ctx = ssl.create_default_context() with socket.create_connection((host, 443), timeout=TIMEOUT) as sock: # server_hostname sends SNI, so we get the certificate for this exact name  with ctx.wrap_socket(sock, server_hostname=host) as tls: cert = tls.getpeercert() expires = ssl.cert_time_to_seconds(cert["notAfter"]) return int((expires - time.time()) // 86400) # ---------- Check 2: domain expiry via RDAP ----------  def get_json(url: str): headers = {**HEADERS, "Accept": "application/rdap+json, application/json"} req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return json.load(resp) _bootstrap = None def rdap_base_url(tld: str) -> str | None: """Ask IANA which RDAP server is responsible for this TLD.""" global _bootstrap if _bootstrap is None: _bootstrap = get_json("https://data.iana.org/rdap/dns.json") for tlds, urls in _bootstrap["services"]: if tld in tlds: return next((u for u in urls if u.startswith("https://")), None) return None def domain_days_left(client: dict) -> int | None: """Days until the domain expires, or None if we cannot tell.""" manual = client.get("domain_expires") # hand-entered fallback or override  if manual: return (date.fromisoformat(str(manual)) - datetime.now(timezone.utc).date()).days domain = client["domain"].lower() base = rdap_base_url(domain.rsplit(".", 1)[-1]) if base is None: return None # this TLD has no RDAP server  data = get_json(f"{base}domain/{domain}") for event in data.get("events", []): if event.get("eventAction") == "expiration": stamp = re.sub(r"\.\d+", "", event["eventDate"]).replace("Z", "+00:00") expires = datetime.fromisoformat(stamp) if expires.tzinfo is None: expires = expires.replace(tzinfo=timezone.utc) return (expires - datetime.now(timezone.utc)).days return None # ---------- Check 3: is the site up and showing the right content? ----------  def check_http(url: str, expect: str | None = None) -> str | None: """Return a short problem description, or None if the site looks healthy.""" problem = "unknown error" for attempt in range(2): # one retry stops a single network blip from paging you  try: req = urllib.request.Request(url, headers=HEADERS) with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: body = resp.read(500_000).decode("utf-8", errors="replace") if expect is None or expect in body: return None problem = f'page loaded but "{expect}" was not found' except urllib.error.HTTPError as e: problem = f"HTTP {e.code}" except Exception as e: # DNS failure, timeout, TLS error, and so on  problem = f"{type(e).__name__}: {e}" if attempt == 0: time.sleep(5) return problem 

# ---------- Alert rules ----------  def should_alert(days_left: int) -> bool: """Alert at 30, 14 and 7 days, then every day once 3 or fewer remain.""" return days_left in (30, 14, 7) or days_left <= 3 def when(days: int) -> str: return f"expired {-days} days ago" if days < 0 else f"expires in {days} days" def expiry_messages(client: dict) -> list[str]: name = client["name"] msgs: list[str] = [] # Domain  try: days = domain_days_left(client) if days is None: msgs.append( f"[{name}] No expiry date found for {client['domain']}. " "Add domain_expires to clients.yml." ) elif should_alert(days): msgs.append(f"[{name}] Domain {client['domain']} {when(days)}") except Exception as e: msgs.append(f"[{name}] Domain lookup failed for {client['domain']}: {e}") # SSL, once per HTTPS hostname  hosts = sorted( {urlparse(s["url"]).hostname for s in client["sites"] if s["url"].startswith("https://")} ) for host in hosts: try: days = ssl_days_left(host) except ssl.SSLCertVerificationError as e: # expired, wrong host, bad chain  reason = getattr(e, "verify_message", None) or str(e) msgs.append(f"[{name}] SSL for {host} is invalid: {reason}") continue except OSError as e: msgs.append(f"[{name}] SSL check for {host} could not connect: {e}") continue if should_alert(days): msgs.append(f"[{name}] SSL certificate for {host} {when(days)}") return msgs def uptime_messages(client: dict, state: dict) -> list[str]: """Alert when a site goes down and when it recovers, not on every run.""" msgs: list[str] = [] for site in client["sites"]: url = site["url"] problem = check_http(url, site.get("expect")) now = "down" if problem else "up" before = state.get(url, "up") if now == "down" and before == "up": msgs.append(f"[{client['name']}] DOWN: {url} ({problem})") elif now == "up" and before == "down": msgs.append(f"[{client['name']}] RECOVERED: {url}") state[url] = now return msgs # ---------- Telegram, state and heartbeat ----------  STATE_FILE = os.environ.get("STATE_FILE", "state.json") def load_state() -> dict: try: with open(STATE_FILE) as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {} def save_state(state: dict) -> None: with open(STATE_FILE, "w") as f: json.dump(state, f, indent=2) def send_telegram(text: str) -> None: token = os.environ["TELEGRAM_TOKEN"] chat_id = os.environ["TELEGRAM_CHAT_ID"] data = urlencode({"chat_id": chat_id, "text": text[:4000]}).encode() url = f"https://api.telegram.org/bot{token}/sendMessage" urllib.request.urlopen(url, data=data, timeout=TIMEOUT).read() def ping_heartbeat() -> None: url = os.environ.get("HEARTBEAT_URL") if url: # skipped while you test locally  urllib.request.urlopen( urllib.request.Request(url, headers=HEADERS), timeout=TIMEOUT ).read() # ---------- Main ----------  def main() -> None: mode = sys.argv[1] if len(sys.argv) > 1 else "expiry" if mode not in ("expiry", "uptime"): sys.exit("usage: python monitor.py [expiry|uptime]") with open("clients.yml") as f: clients = yaml.safe_load(f)["clients"] state = load_state() if mode == "uptime" else {} messages: list[str] = [] for client in clients: if mode == "expiry": messages += expiry_messages(client) else: messages += uptime_messages(client, state) if messages: send_telegram("Client watchdog\n\n" + "\n".join(f"- {m}" for m in messages)) if mode == "uptime": save_state(state) # saved only after the alert went out  print("\n"Source：https://dev.to/sanjay_singh_1/how-to-monitor-client-domains-and-ssl-certificates-for-free-before-they-expire-36fl
My favorites