Ajoute le support d'un jeton d'accès pour les serveurs ntfy protégés

Certains serveurs ntfy auto-hébergés (ex: ntfy.arkel.fr) renvoient
403 Forbidden sans authentification. Un nouveau champ "Jeton d'accès"
(optionnel, type password) sur /settings permet de renseigner un token
qui sera envoyé comme "Authorization: Bearer ..." à chaque publication,
pour les rappels réels comme pour le bouton de test. La page /aide
documente le cas 403 et la marche à suivre.

💘 Generated with Crush

Assisted-by: Crush:glm-5.2
This commit is contained in:
Jacquin Antoine
2026-07-20 00:01:12 +02:00
parent 2ee317a46e
commit 076b5a4fa2
6 changed files with 101 additions and 8 deletions

View File

@ -563,6 +563,7 @@ def settings_page(request: Request):
"current_user": user_id, "current_user": user_id,
"ntfy_topic": notif.get("ntfy_topic") or "", "ntfy_topic": notif.get("ntfy_topic") or "",
"ntfy_server": notif.get("ntfy_server") or "https://ntfy.sh", "ntfy_server": notif.get("ntfy_server") or "https://ntfy.sh",
"ntfy_token": notif.get("ntfy_token") or "",
"rappel_debut_h": notif.get("rappel_debut_h", 7), "rappel_debut_h": notif.get("rappel_debut_h", 7),
"rappel_fin_h": notif.get("rappel_fin_h", 20), "rappel_fin_h": notif.get("rappel_fin_h", 20),
"arrivee_url": f"{base_url}/presence/{notif['token']}/arrivee", "arrivee_url": f"{base_url}/presence/{notif['token']}/arrivee",
@ -578,6 +579,7 @@ def settings_page(request: Request):
def settings_save( def settings_save(
request: Request, ntfy_topic: str = Form(""), ntfy_server: str = Form(""), request: Request, ntfy_topic: str = Form(""), ntfy_server: str = Form(""),
rappel_debut_h: int = Form(7), rappel_fin_h: int = Form(20), rappel_debut_h: int = Form(7), rappel_fin_h: int = Form(20),
ntfy_token: str = Form(""),
): ):
user_id = _get_user(request) user_id = _get_user(request)
if not user_id: if not user_id:
@ -587,10 +589,11 @@ def settings_save(
rappel_fin_h = max(rappel_debut_h + 1, min(24, rappel_fin_h)) rappel_fin_h = max(rappel_debut_h + 1, min(24, rappel_fin_h))
logger.info( logger.info(
"réglages notif mis à jour: user_id=%r ntfy_topic=%r ntfy_server=%r plage=%sh-%sh", "réglages notif mis à jour: user_id=%r ntfy_topic=%r ntfy_server=%r plage=%sh-%sh token=%s",
user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h, user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h,
"(défini)" if ntfy_token.strip() else "(vide)",
) )
save_notif_config(user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h) save_notif_config(user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h, ntfy_token)
return RedirectResponse("/settings?saved=1", status_code=303) return RedirectResponse("/settings?saved=1", status_code=303)
@ -616,7 +619,7 @@ def settings_test_ntfy(request: Request):
ok = send_ntfy( ok = send_ntfy(
server, topic, server, topic,
"Ceci est une notification de test — ta configuration ntfy fonctionne.", "Ceci est une notification de test — ta configuration ntfy fonctionne.",
"Test pointeuse", user_id=user_id, "Test pointeuse", user_id=user_id, token=notif.get("ntfy_token", ""),
) )
cls = "settings-saved" if ok else "settings-error" cls = "settings-saved" if ok else "settings-error"
msg = ("Notification envoyée. Vérifie ton téléphone, elle doit arriver en quelques secondes." msg = ("Notification envoyée. Vérifie ton téléphone, elle doit arriver en quelques secondes."

View File

@ -156,11 +156,12 @@ DEFAULT_RAPPEL_FIN_H = 20
def load_notif_config(user_id: str) -> dict: def load_notif_config(user_id: str) -> dict:
"""Return {"ntfy_topic", "ntfy_server", "rappel_debut_h", "rappel_fin_h", "token"}.""" """Return {ntfy_topic, ntfy_server, ntfy_token, rappel_debut_h, rappel_fin_h, token}."""
f = _notif_file(user_id) f = _notif_file(user_id)
if f.exists(): if f.exists():
cfg = json.loads(f.read_text()) cfg = json.loads(f.read_text())
cfg.setdefault("ntfy_server", DEFAULT_NTFY_SERVER) cfg.setdefault("ntfy_server", DEFAULT_NTFY_SERVER)
cfg.setdefault("ntfy_token", "")
cfg.setdefault("rappel_debut_h", DEFAULT_RAPPEL_DEBUT_H) cfg.setdefault("rappel_debut_h", DEFAULT_RAPPEL_DEBUT_H)
cfg.setdefault("rappel_fin_h", DEFAULT_RAPPEL_FIN_H) cfg.setdefault("rappel_fin_h", DEFAULT_RAPPEL_FIN_H)
if cfg.get("token"): if cfg.get("token"):
@ -168,6 +169,7 @@ def load_notif_config(user_id: str) -> dict:
else: else:
cfg = { cfg = {
"ntfy_topic": None, "ntfy_server": DEFAULT_NTFY_SERVER, "ntfy_topic": None, "ntfy_server": DEFAULT_NTFY_SERVER,
"ntfy_token": "",
"rappel_debut_h": DEFAULT_RAPPEL_DEBUT_H, "rappel_fin_h": DEFAULT_RAPPEL_FIN_H, "rappel_debut_h": DEFAULT_RAPPEL_DEBUT_H, "rappel_fin_h": DEFAULT_RAPPEL_FIN_H,
} }
cfg["token"] = secrets.token_hex(16) cfg["token"] = secrets.token_hex(16)
@ -179,10 +181,12 @@ def load_notif_config(user_id: str) -> dict:
def save_notif_config( def save_notif_config(
user_id: str, ntfy_topic: str, ntfy_server: str = "", user_id: str, ntfy_topic: str, ntfy_server: str = "",
rappel_debut_h: int = DEFAULT_RAPPEL_DEBUT_H, rappel_fin_h: int = DEFAULT_RAPPEL_FIN_H, rappel_debut_h: int = DEFAULT_RAPPEL_DEBUT_H, rappel_fin_h: int = DEFAULT_RAPPEL_FIN_H,
ntfy_token: str = "",
): ):
cfg = load_notif_config(user_id) cfg = load_notif_config(user_id)
cfg["ntfy_topic"] = ntfy_topic.strip() or None cfg["ntfy_topic"] = ntfy_topic.strip() or None
cfg["ntfy_server"] = ntfy_server.strip().rstrip("/") or DEFAULT_NTFY_SERVER cfg["ntfy_server"] = ntfy_server.strip().rstrip("/") or DEFAULT_NTFY_SERVER
cfg["ntfy_token"] = ntfy_token.strip()
cfg["rappel_debut_h"] = rappel_debut_h cfg["rappel_debut_h"] = rappel_debut_h
cfg["rappel_fin_h"] = rappel_fin_h cfg["rappel_fin_h"] = rappel_fin_h
_ensure_user_dirs(user_id) _ensure_user_dirs(user_id)

View File

@ -11,11 +11,14 @@ logger = logging.getLogger("pointeuse.notifications")
CHECK_INTERVAL_SECONDS = 300 CHECK_INTERVAL_SECONDS = 300
def send_ntfy(server: str, topic: str, message: str, title: str, user_id: str | None = None) -> bool: def send_ntfy(server: str, topic: str, message: str, title: str, user_id: str | None = None, token: str = "") -> bool:
headers = {"Title": title.encode("utf-8"), "Priority": "high"}
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request( req = urllib.request.Request(
f"{server}/{topic}", f"{server}/{topic}",
data=message.encode("utf-8"), data=message.encode("utf-8"),
headers={"Title": title.encode("utf-8"), "Priority": "high"}, headers=headers,
method="POST", method="POST",
) )
try: try:
@ -43,6 +46,7 @@ async def check_user(user_id: str, now: datetime):
notif = load_notif_config(user_id) notif = load_notif_config(user_id)
topic = notif.get("ntfy_topic") topic = notif.get("ntfy_topic")
server = notif.get("ntfy_server", "https://ntfy.sh") server = notif.get("ntfy_server", "https://ntfy.sh")
token = notif.get("ntfy_token", "")
debut_h = notif.get("rappel_debut_h", 7) debut_h = notif.get("rappel_debut_h", 7)
fin_h = notif.get("rappel_fin_h", 20) fin_h = notif.get("rappel_fin_h", 20)
if not topic: if not topic:
@ -64,14 +68,14 @@ async def check_user(user_id: str, now: datetime):
await asyncio.to_thread( await asyncio.to_thread(
send_ntfy, server, topic, send_ntfy, server, topic,
"Tu es dans les locaux mais pas encore pointé ce matin.", "Tu es dans les locaux mais pas encore pointé ce matin.",
"⏰ Pense à pointer", user_id, "⏰ Pense à pointer", user_id, token,
) )
elif not presence.get("present") and jour.get("matin_entree") and not jour.get("aprem_sortie"): 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) logger.info("rappel 'pense à dépointer' pour %r", user_id)
await asyncio.to_thread( await asyncio.to_thread(
send_ntfy, server, topic, send_ntfy, server, topic,
"Tu as quitté les locaux sans dépointer la sortie.", "Tu as quitté les locaux sans dépointer la sortie.",
"⏰ Pense à dépointer", user_id, "⏰ Pense à dépointer", user_id, token,
) )

View File

@ -57,6 +57,13 @@
(Réglages → Applications → ntfy → Batterie → "Aucune restriction"). Sinon Android (Réglages → Applications → ntfy → Batterie → "Aucune restriction"). Sinon Android
peut couper les notifications en arrière-plan. peut couper les notifications en arrière-plan.
</li> </li>
<li>
<b>Serveur protégé par jeton</b> : si ton instance ntfy renvoie <code>403 Forbidden</code>,
elle exige un jeton d'accès. Crée-le sur le serveur (panneau d'admin ntfy, onglet
"Access tokens"), puis colle-le dans le champ <b>Jeton d'accès</b> sur
<a href="/settings">/settings</a>. Il est envoyé comme
<code>Authorization: Bearer ...</code> à chaque publication.
</li>
</ol> </ol>
<p class="settings-hint"> <p class="settings-hint">
Astuce : si tu veux ton propre serveur ntfy (auto-hébergé, pour éviter le topic public), Astuce : si tu veux ton propre serveur ntfy (auto-hébergé, pour éviter le topic public),

View File

@ -84,6 +84,16 @@
placeholder="ex: antoine-pointeuse-x7f2" placeholder="ex: antoine-pointeuse-x7f2"
autocomplete="off" autocomplete="off"
> >
<label class="login-label" for="ntfy_token">Jeton d'accès (optionnel, serveur protégé)</label>
<input
id="ntfy_token"
class="login-input"
type="password"
name="ntfy_token"
value="{{ ntfy_token }}"
placeholder="ex: tk_xxxxxxxxxxxxxxxx"
autocomplete="off"
>
<label class="login-label" for="rappel_debut_h">Plage horaire des rappels</label> <label class="login-label" for="rappel_debut_h">Plage horaire des rappels</label>
<div class="settings-range"> <div class="settings-range">
<input <input

View File

@ -185,3 +185,68 @@ def test_test_ntfy_logs_to_journal():
logs = models.load_logs("alice") logs = models.load_logs("alice")
# L'entrée la plus récente doit être l'échec du test (message contient "Test pointeuse") # L'entrée la plus récente doit être l'échec du test (message contient "Test pointeuse")
assert any(l["event"] == "échec" and "Test pointeuse" in l["message"] for l in logs) assert any(l["event"] == "échec" and "Test pointeuse" in l["message"] for l in logs)
# ── Jeton d'accès ntfy (serveur protégé) ─────────────────────────────────────
def test_send_ntfy_passes_authorization_header(monkeypatch):
"""Quand un token est fourni, send_ntfy doit envoyer un header Authorization: Bearer."""
import notifications
captured = {}
class _DummyReq:
def __init__(self, url, data, headers, method):
captured["url"] = url
captured["headers"] = headers
captured["method"] = method
class _DummyResp:
def close(self): pass
def _fake_urlopen(req, timeout):
captured["timeout"] = timeout
return _DummyResp()
monkeypatch.setattr(notifications.urllib.request, "Request", _DummyReq)
monkeypatch.setattr(notifications.urllib.request, "urlopen", _fake_urlopen)
ok = notifications.send_ntfy(
"https://ntfy.example", "topic-x",
"msg", "titre", user_id="alice", token="tk_abc123",
)
assert ok is True
assert captured["headers"]["Authorization"] == "Bearer tk_abc123"
def test_send_ntfy_omits_authorization_when_no_token(monkeypatch):
"""Sans token, pas de header Authorization."""
import notifications
captured = {}
class _DummyReq:
def __init__(self, url, data, headers, method):
captured["headers"] = headers
class _DummyResp:
def close(self): pass
monkeypatch.setattr(notifications.urllib.request, "Request", _DummyReq)
monkeypatch.setattr(notifications.urllib.request, "urlopen", lambda req, timeout: _DummyResp())
notifications.send_ntfy("https://x", "t", "m", "ti", user_id=None, token="")
assert "Authorization" not in captured["headers"]
def test_settings_save_persists_ntfy_token():
"""Le token saisi dans /settings doit être persisté."""
models.save_notif_config("alice", ntfy_topic="t", ntfy_server="https://ntfy.sh")
client.post("/settings", cookies=COOKIES, data={
"ntfy_topic": "topic-persisted",
"ntfy_server": "https://ntfy.arkel.fr",
"ntfy_token": "tk_persisted_123",
"rappel_debut_h": "8",
"rappel_fin_h": "19",
})
cfg = models.load_notif_config("alice")
assert cfg["ntfy_token"] == "tk_persisted_123"
assert cfg["ntfy_topic"] == "topic-persisted"