"""Notifications Web Push (VAPID). Les clés VAPID sont générées au premier démarrage et stockées dans la table `settings`, donc elles survivent aux redémarrages tant que /data est monté. pywebpush s'appuie sur `requests` (bloquant) : les envois partent dans un thread pool pour ne pas figer l'event loop asyncio. """ from __future__ import annotations import asyncio import base64 import json import logging from concurrent.futures import ThreadPoolExecutor from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat from pywebpush import WebPushException, webpush from . import config, db log = logging.getLogger("push") # Peu de threads : les envois sont courts et le tier gratuit n'a que 2 vCPU. _executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="push") def _b64(raw: bytes) -> str: return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") def _ensure_keys() -> tuple[str, str]: """Renvoie (clé privée b64url, clé publique b64url), en les créant au besoin.""" priv = db.get_setting("vapid_private") pub = db.get_setting("vapid_public") if priv and pub: return priv, pub key = ec.generate_private_key(ec.SECP256R1()) private_raw = key.private_numbers().private_value.to_bytes(32, "big") public_raw = key.public_key().public_bytes( encoding=Encoding.X962, format=PublicFormat.UncompressedPoint ) priv, pub = _b64(private_raw), _b64(public_raw) db.set_setting("vapid_private", priv) db.set_setting("vapid_public", pub) log.info("Nouvelles clés VAPID générées.") return priv, pub def public_key() -> str: """Clé publique à passer à `pushManager.subscribe` côté navigateur.""" return _ensure_keys()[1] def _send_one(sub_id: int, endpoint: str, p256dh: str, auth_secret: str, payload: str) -> None: priv, _ = _ensure_keys() try: webpush( subscription_info={ "endpoint": endpoint, "keys": {"p256dh": p256dh, "auth": auth_secret}, }, data=payload, vapid_private_key=priv, vapid_claims={"sub": config.VAPID_SUBJECT}, ttl=86400, timeout=10, ) except WebPushException as exc: code = getattr(exc.response, "status_code", None) # 404/410 : l'abonnement n'existe plus côté navigateur, on le purge. if code in (404, 410): db.execute("DELETE FROM push_subscriptions WHERE id = ?", (sub_id,)) log.info("Abonnement push %s expiré, supprimé.", sub_id) else: log.warning("Échec d'envoi push (%s) : %s", code, exc) except Exception as exc: # réseau, DNS, etc. log.warning("Erreur push inattendue : %s", exc) async def notify(phones: list[str], payload: dict) -> None: """Envoie une notification à tous les appareils des numéros donnés.""" if not phones: return placeholders = ",".join("?" * len(phones)) rows = db.query( f"SELECT id, endpoint, p256dh, auth FROM push_subscriptions WHERE phone IN ({placeholders})", phones, ) if not rows: return body = json.dumps(payload, ensure_ascii=False) loop = asyncio.get_running_loop() for row in rows: loop.run_in_executor( _executor, _send_one, row["id"], row["endpoint"], row["p256dh"], row["auth"], body ) def save_subscription(phone: str, subscription: dict) -> None: endpoint = subscription.get("endpoint", "") keys = subscription.get("keys") or {} p256dh, auth_secret = keys.get("p256dh", ""), keys.get("auth", "") if not (endpoint and p256dh and auth_secret): raise ValueError("Abonnement push incomplet.") db.execute( "INSERT INTO push_subscriptions(phone, endpoint, p256dh, auth, created_at) " "VALUES(?, ?, ?, ?, ?) " "ON CONFLICT(endpoint) DO UPDATE SET phone = excluded.phone, " "p256dh = excluded.p256dh, auth = excluded.auth", (phone, endpoint, p256dh, auth_secret, db.now_ms()), ) def delete_subscription(endpoint: str) -> None: db.execute("DELETE FROM push_subscriptions WHERE endpoint = ?", (endpoint,)) def shutdown() -> None: _executor.shutdown(wait=False, cancel_futures=True)