pointeuse-optimisator/app/notifications.py

"""Reminders: nag via ntfy when present-but-unpointed or departed-but-not-depointed."""
import asyncio
import ipaddress
import logging
import urllib.request
from datetime import datetime
from urllib.parse import urlsplit

from models import list_users, load_notif_config, load_presence, load_week_pointages, append_log

logger = logging.getLogger("pointeuse.notifications")

CHECK_INTERVAL_SECONDS = 300


def is_safe_ntfy_server(url: str) -> bool:
    """Valide qu'une URL ntfy est sûre (pas de SSRF vers le réseau interne/cloud metadata).

    Règles :
    - Schéma http(s) uniquement
    - Pas de credentials embarqués dans l'URL
    - Pas d'IP littérale privée (loopback, link-local, privé, multicast, réservé)
      → bloque 127.x, 10.x, 172.16/12, 192.168.x, 169.254.x (metadata AWS/GCP), fc00::/7, ::1…
    - Pas de hostname localhost
    """
    if not url or not isinstance(url, str):
        return False
    try:
        parts = urlsplit(url.strip())
    except ValueError:
        return False
    if parts.scheme not in ("http", "https"):
        return False
    if parts.username or parts.password:
        return False
    host = parts.hostname
    if not host:
        return False
    host = host.lower().rstrip(".")
    if host in ("localhost",):
        return False
    # IP littérale ?
    try:
        ip = ipaddress.ip_address(host)
    except ValueError:
        # hostname : on accepte (la résolution DNS n'est pas vérifiée ici, mais on a déjà
        # éliminé les IP littérales dangereuses, ce qui ferme le cas le plus exploité).
        return True
    if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified:
        return False
    return True


def send_ntfy(server: str, topic: str, message: str, title: str, user_id: str | None = None, token: str = "") -> bool:
    if not is_safe_ntfy_server(server):
        logger.warning("ntfy serveur refusé (SSRF): server=%r topic=%r title=%r", server, topic, title)
        if user_id:
            append_log(user_id, "notif", "échec",
                       f"{title} — {message} (serveur interdit: {server})")
        return False
    headers = {"Title": title.encode("utf-8"), "Priority": "high"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    req = urllib.request.Request(
        f"{server}/{topic}",
        data=message.encode("utf-8"),
        headers=headers,
        method="POST",
    )
    try:
        urllib.request.urlopen(req, timeout=10).close()
        logger.info("notif ntfy envoyée: server=%r topic=%r title=%r", server, topic, title)
        if user_id:
            append_log(user_id, "notif", "envoyée", f"{title} — {message}")
        return True
    except Exception:
        logger.exception("échec de l'envoi ntfy: server=%r topic=%r title=%r", server, topic, title)
        if user_id:
            append_log(user_id, "notif", "échec", f"{title} — {message}")
        # best-effort: a missed reminder isn't worth crashing the loop
        return False


def _todays_pointage(user_id: str, today: datetime) -> dict:
    iso = today.isocalendar()
    pointages = load_week_pointages(iso.year, iso.week, user_id)
    date_str = today.strftime("%Y-%m-%d")
    return next((p for p in pointages if p["date"] == date_str), {})


async def check_user(user_id: str, now: datetime):
    notif = load_notif_config(user_id)
    topic = notif.get("ntfy_topic")
    server = notif.get("ntfy_server", "https://ntfy.sh")
    token = notif.get("ntfy_token", "")
    debut_h = notif.get("rappel_debut_h", 7)
    fin_h = notif.get("rappel_fin_h", 20)
    if not topic:
        logger.debug("check_user(%r): pas de topic ntfy configuré, ignoré", user_id)
        return
    if now.weekday() >= 5 or not (debut_h <= now.hour < fin_h):
        logger.debug("check_user(%r): hors plage de rappel (%s, plage %sh-%sh)", user_id, now, debut_h, fin_h)
        return

    presence = load_presence(user_id)
    jour = _todays_pointage(user_id, now)
    logger.debug(
        "check_user(%r): presence=%s matin_entree=%r aprem_sortie=%r",
        user_id, presence.get("present"), jour.get("matin_entree"), jour.get("aprem_sortie"),
    )

    if presence.get("present") and not jour.get("matin_entree"):
        logger.info("rappel 'pense à pointer' pour %r", user_id)
        await asyncio.to_thread(
            send_ntfy, server, topic,
            "Tu es dans les locaux mais pas encore pointé ce matin.",
            "⏰ Pense à pointer", user_id, token,
        )
    elif not presence.get("present") and jour.get("matin_entree") and not jour.get("aprem_sortie"):
        logger.info("rappel 'pense à dépointer' pour %r", user_id)
        await asyncio.to_thread(
            send_ntfy, server, topic,
            "Tu as quitté les locaux sans dépointer la sortie.",
            "⏰ Pense à dépointer", user_id, token,
        )


async def reminder_loop():
    logger.info("boucle de rappels démarrée (intervalle=%ss)", CHECK_INTERVAL_SECONDS)
    while True:
        now = datetime.now()
        users = list_users()
        logger.debug("cycle de vérification pour %d utilisateur(s)", len(users))
        for user_id in users:
            await check_user(user_id, now)
        await asyncio.sleep(CHECK_INTERVAL_SECONDS)