server-up/server-up/core/notify.py

"""Meldingen bij storingen.

Geplande backups, updatechecks en de zelf-update draaien zonder dat er iemand
kijkt. Zonder meldingen komt een mislukte backup alleen in het auditlog terecht
— en dat ontdek je op het moment dat je hem nodig hebt.

Drie kanalen, alle drie met de standaardbibliotheek:

* **ntfy** — POST naar een topic-URL. Het eenvoudigst: een app op je telefoon.
* **webhook** — POST met een JSON-body, voor Discord, Slack, Gotify of eigen spul.
* **e-mail** — SMTP met STARTTLS of SSL.

`send()` gooit nooit: een storing in het meldingskanaal mag de taak die de
melding veroorzaakte niet alsnog laten omvallen.
"""
from __future__ import annotations

import json
import smtplib
import ssl
import urllib.error
import urllib.request
from email.message import EmailMessage

import core as cfg

TIMEOUT = 10

# Gebeurtenissen waarover gemeld kan worden. De sleutel staat in de config, de
# omschrijving is wat de gebruiker in de interface ziet.
EVENTS = {
    "backup_failed":    "Backup mislukt",
    "backup_verify":    "Backup niet te verifiëren",
    "job_failed":       "Geplande taak mislukt",
    "disk_low":         "Weinig schijfruimte",
    "update_available": "Update beschikbaar",
    "stack_down":       "Stack onverwacht gestopt",
}

# Standaard alleen de dingen die misgaan. "Update beschikbaar" is nuttig maar
# geen storing, dus die zet je er zelf bij.
STANDAARD_EVENTS = ["backup_failed", "backup_verify", "job_failed", "disk_low"]

KANALEN = ("", "ntfy", "webhook", "email")


def instellingen() -> dict:
    c = cfg.load()
    events = c.get("NOTIFY_EVENTS")
    if not isinstance(events, list):
        events = list(STANDAARD_EVENTS)
    return {
        "channel": (c.get("NOTIFY_CHANNEL") or "").strip(),
        "url": (c.get("NOTIFY_URL") or "").strip(),
        "token": (c.get("NOTIFY_TOKEN") or "").strip(),
        "events": [e for e in events if e in EVENTS],
        "email_host": (c.get("NOTIFY_EMAIL_HOST") or "").strip(),
        "email_port": int(c.get("NOTIFY_EMAIL_PORT") or 587),
        "email_user": (c.get("NOTIFY_EMAIL_USER") or "").strip(),
        "email_password": c.get("NOTIFY_EMAIL_PASSWORD") or "",
        "email_from": (c.get("NOTIFY_EMAIL_FROM") or "").strip(),
        "email_to": (c.get("NOTIFY_EMAIL_TO") or "").strip(),
        "email_security": (c.get("NOTIFY_EMAIL_SECURITY") or "starttls").strip(),
    }


def ingeschakeld(event: str = "") -> bool:
    i = instellingen()
    if i["channel"] not in KANALEN or not i["channel"]:
        return False
    if event and event not in i["events"]:
        return False
    return True


# ── Kanalen ──────────────────────────────────────────────────────────────────

def _post(url: str, data: bytes, headers: dict) -> tuple[bool, str]:
    verzoek = urllib.request.Request(url, data=data, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(verzoek, timeout=TIMEOUT) as a:
            return True, f"HTTP {a.status}"
    except urllib.error.HTTPError as e:
        return False, f"HTTP {e.code}: {(e.read() or b'')[:200].decode(errors='replace')}"
    except Exception as e:                       # netwerk, DNS, TLS, timeout
        return False, f"{type(e).__name__}: {e}"


def _ntfy(i: dict, titel: str, tekst: str, prioriteit: str) -> tuple[bool, str]:
    if not i["url"]:
        return False, "Geen ntfy-URL ingesteld."
    headers = {"Title": titel, "Content-Type": "text/plain; charset=utf-8"}
    if prioriteit == "hoog":
        headers["Priority"] = "high"
    if i["token"]:
        headers["Authorization"] = f"Bearer {i['token']}"
    return _post(i["url"], tekst.encode("utf-8"), headers)


def _webhook(i: dict, titel: str, tekst: str, prioriteit: str) -> tuple[bool, str]:
    if not i["url"]:
        return False, "Geen webhook-URL ingesteld."
    # `content` en `text` staan erbij omdat Discord respectievelijk Slack die
    # verwachten; zo werkt dezelfde instelling voor allebei zonder keuzemenu.
    body = {"title": titel, "message": tekst, "priority": prioriteit,
            "content": f"**{titel}**\n{tekst}", "text": f"*{titel}*\n{tekst}",
            "source": "server-up"}
    headers = {"Content-Type": "application/json"}
    if i["token"]:
        headers["Authorization"] = f"Bearer {i['token']}"
    return _post(i["url"], json.dumps(body).encode("utf-8"), headers)


def _email(i: dict, titel: str, tekst: str, prioriteit: str) -> tuple[bool, str]:
    if not i["email_host"] or not i["email_to"]:
        return False, "SMTP-server of ontvanger ontbreekt."
    bericht = EmailMessage()
    bericht["Subject"] = f"[Server Up] {titel}"
    bericht["From"] = i["email_from"] or i["email_user"] or "server-up@localhost"
    bericht["To"] = i["email_to"]
    bericht.set_content(tekst)

    try:
        if i["email_security"] == "ssl":
            server = smtplib.SMTP_SSL(i["email_host"], i["email_port"],
                                      timeout=TIMEOUT,
                                      context=ssl.create_default_context())
        else:
            server = smtplib.SMTP(i["email_host"], i["email_port"], timeout=TIMEOUT)
        with server:
            if i["email_security"] == "starttls":
                server.starttls(context=ssl.create_default_context())
            if i["email_user"]:
                server.login(i["email_user"], i["email_password"])
            server.send_message(bericht)
        return True, "Verstuurd."
    except Exception as e:
        return False, f"{type(e).__name__}: {e}"


# ── Versturen ────────────────────────────────────────────────────────────────

def verstuur(titel: str, tekst: str, prioriteit: str = "normaal") -> tuple[bool, str]:
    """Stuur een bericht via het ingestelde kanaal, ongeacht de gebeurtenislijst.

    Dit is wat de testknop gebruikt.
    """
    i = instellingen()
    if not i["channel"]:
        return False, "Er is geen meldingskanaal ingesteld."
    if i["channel"] == "ntfy":
        return _ntfy(i, titel, tekst, prioriteit)
    if i["channel"] == "webhook":
        return _webhook(i, titel, tekst, prioriteit)
    if i["channel"] == "email":
        return _email(i, titel, tekst, prioriteit)
    return False, f"Onbekend kanaal: {i['channel']}"


def send(event: str, titel: str, tekst: str, prioriteit: str = "hoog") -> bool:
    """Meld een gebeurtenis, als daarvoor gekozen is.

    Gooit nooit: het kanaal mag de taak die de melding veroorzaakte niet alsnog
    laten omvallen. Het resultaat gaat wel naar het auditlog, zodat je kunt zien
    dát er gemeld is — en of dat lukte.
    """
    try:
        if not ingeschakeld(event):
            return False
        ok, melding = verstuur(titel, tekst, prioriteit)
    except Exception as e:                                    # nooit doorgeven
        ok, melding = False, f"{type(e).__name__}: {e}"

    try:
        from core import audit
        audit.log("notify", event, "ok" if ok else "fout", ref=melding[:200])
    except Exception:
        pass
    return ok