Add SSL expiry alerts

services/ssl_alerts.py

@@ -0,0 +1,57 @@
+import asyncio
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+from services.npmplus import fetch_certificates, _parse_expiry
+
+
+async def monitor_ssl(cfg: dict[str, Any], notify, bot, chat_id: int):
+    npm_cfg = cfg.get("npmplus", {})
+    alert_cfg = npm_cfg.get("alerts", {})
+    if not alert_cfg.get("enabled", True):
+        return
+
+    days_list = alert_cfg.get("days", [30, 14, 7, 1])
+    days_list = sorted({int(x) for x in days_list if int(x) >= 0}, reverse=True)
+    cooldown = int(alert_cfg.get("cooldown_sec", 86400))
+    interval = int(alert_cfg.get("interval_sec", 3600))
+
+    last_sent: dict[str, float] = {}
+
+    while True:
+        now = datetime.now(timezone.utc)
+        try:
+            certs = fetch_certificates(cfg)
+        except Exception:
+            await asyncio.sleep(interval)
+            continue
+
+        for cert in certs:
+            name = cert.get("nice_name")
+            if not name:
+                domains = cert.get("domain_names") or []
+                if isinstance(domains, list):
+                    name = ", ".join(domains)
+            if not name:
+                name = "unknown"
+
+            expiry = _parse_expiry(cert.get("expires_on"))
+            if expiry is None:
+                continue
+
+            days_left = (expiry - now).days
+            for threshold in days_list:
+                if days_left <= threshold:
+                    key = f"{name}:{threshold}"
+                    last_time = last_sent.get(key, 0)
+                    if time.time() - last_time >= cooldown:
+                        await notify(
+                            bot,
+                            chat_id,
+                            f"⚠️ SSL `{name}` expires in {days_left}d (threshold {threshold}d)",
+                        )
+                        last_sent[key] = time.time()
+                    break
+
+        await asyncio.sleep(interval)