58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
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)
|