diff --git a/README.md b/README.md index d068465..38711fd 100644 --- a/README.md +++ b/README.md @@ -77,7 +77,8 @@ Seul `data/config.example.json` est versionné. ```json { - "heures_jour": 7.8, + "heures_semaine": 39.0, + "jours_par_semaine": 5, "log_level": "DEBUG", "allowed_email_domain": "exemple.fr", "mail_from": "no-reply@exemple.fr", @@ -91,7 +92,8 @@ Seul `data/config.example.json` est versionné. } ``` -- `heures_jour` — `7.8` = 7h48 = 39h ÷ 5 jours. Modifier cette valeur recentre tous les calculs. +- `heures_semaine` — heures hebdomadaires visées (ex: `39.0`). +- `jours_par_semaine` — nombre de jours travaillés dans la semaine (ex: `5`). Le quota quotidien est calculé : `heures_semaine ÷ jours_par_semaine` (39h / 5 = 7h48/jour). - `log_level` — `DEBUG`, `INFO`, `WARNING` ou `ERROR`. Contrôle la verbosité des logs (`docker logs`). - `allowed_email_domain` — seules les adresses de ce domaine peuvent se connecter. - `mail_from` — adresse expéditrice des emails de connexion. diff --git a/app/calcul.py b/app/calcul.py index 81ad0a6..9bbbdaa 100644 --- a/app/calcul.py +++ b/app/calcul.py @@ -131,6 +131,8 @@ def compute_week(pointages: list[dict], conges: list[dict], heures_jour_min: int for jour in jours: jour["cible_matin"] = None jour["cible_aprem"] = None + jour["cible_matin_entree"] = None + jour["cible_aprem_entree"] = None restant = total_du - total_travaille @@ -218,11 +220,23 @@ def compute_week(pointages: list[dict], conges: list[dict], heures_jour_min: int adjs[i] -= take left -= take + # Le matin propose toujours son horaire nominal (jamais ajusté par le rattrapage + # hebdomadaire) ; l'après-midi propose son horaire nominal éventuellement décalé + # par l'écart (retard ou avance) réparti ci-dessus. + for s in slots: + if s["half"] != "matin": + continue + # Pas de plancher matin_fin ici : le nominal est déjà borné par cap (≤ matin_fin + # - debut), donc debut + nominal ne peut pas dépasser matin_fin de lui-même. + sortie_min = s["debut"] + s["nominal"] + entree = f"{s['debut'] // 60:02d}:{s['debut'] % 60:02d}" + s["jour"]["cible_matin_entree"] = entree + s["jour"]["cible_matin"] = f"{sortie_min // 60:02d}:{sortie_min % 60:02d}" + for i, s in enumerate(aprem_slots): - if adjs[i] == 0: - continue # rythme normal suffit pour cette demi-journée, rien à signaler - sortie_min = s["debut"] + s["nominal"] + adjs[i] - sortie_min = max(sortie_min, aprem_fin_min) + sortie_min = max(s["debut"] + s["nominal"] + adjs[i], aprem_fin_min) + entree = f"{s['debut'] // 60:02d}:{s['debut'] % 60:02d}" + s["jour"]["cible_aprem_entree"] = entree s["jour"]["cible_aprem"] = f"{sortie_min // 60:02d}:{sortie_min % 60:02d}" extra_total = max(0, ecart_total) diff --git a/app/main.py b/app/main.py index ae4757f..b50a3c2 100644 --- a/app/main.py +++ b/app/main.py @@ -1,10 +1,16 @@ import asyncio +import hashlib +import hmac +import html import json import logging import re import secrets +import time +from collections import defaultdict from datetime import datetime, timedelta from itertools import groupby +from urllib.parse import urlsplit import bcrypt from fastapi import FastAPI, Form, Request, Response @@ -17,13 +23,15 @@ from models import ( load_config, load_week_pointages, save_week_pointages, load_conges, toggle_conge, list_all_weeks, load_notif_config, save_notif_config, find_user_by_token, save_presence, + regenerate_presence_token, load_auth, save_auth, find_user_by_reset_token, load_plages, save_plages, heures_jour_min, + append_log, load_logs, load_or_create_secret, ) from calcul import compute_week, minutes_to_hhmm, hhmm_to_minutes, heures_dues from stats import compute_all_stats from export import export_semaine_ods, export_annee_ods -from notifications import reminder_loop +from notifications import reminder_loop, send_ntfy, is_safe_ntfy_server logging.basicConfig( level=getattr(logging, load_config().get("log_level", "DEBUG").upper(), logging.DEBUG), @@ -34,6 +42,64 @@ logger = logging.getLogger("pointeuse") app = FastAPI(title="Décompte Horaire") +@app.middleware("http") +async def csrf_origin_check(request: Request, call_next): + """Protection CSRF stateless sur les requêtes mutatives. + + Stratégie défense-en-profondeur combinée au cookie SameSite=Lax : + si la requête porte un header Origin ou Referer, on vérifie que son host + correspond à celui attendu (config.base_url si définie, sinon le Host de + la requête). Une requête cross-site aura un Origin différent. + L'absence d'Origin est tolérée (cas des clients non-navigateurs comme + Tasker ou curl, où SameSite=Lax suffit). + """ + if request.method in ("POST", "PUT", "DELETE", "PATCH"): + origin = request.headers.get("origin") or request.headers.get("referer") + if origin: + try: + parsed = urlsplit(origin) + except ValueError: + return JSONResponse({"error": "csrf"}, status_code=403) + origin_host = (parsed.hostname or "").lower() + configured_base = (load_config().get("base_url") or "").strip() + if configured_base: + try: + expected = (urlsplit(configured_base).hostname or "").lower() + except ValueError: + expected = "" + else: + expected = (request.headers.get("host") or "").lower().split(":", 1)[0] + if origin_host and expected and origin_host != expected: + logger.warning( + "CSRF: origin rejettée origin=%r host=%r path=%r", + origin, request.headers.get("host"), request.url.path, + ) + return JSONResponse({"error": "csrf"}, status_code=403) + response = await call_next(request) + # ── En-têtes de sécurité appliqués à toutes les réponses ── + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["X-Frame-Options"] = "DENY" + response.headers["Referrer-Policy"] = "same-origin" + # CSP relaxée : on autorise Google Fonts (chargé depuis base.html) et les + # scripts/styles inline (semaine.html et stats.html ont du JS/CSS inline). + # 'unsafe-inline' reste nécessaire tant que ces ressources ne sont pas + # extraites dans des fichiers statiques. + response.headers["Content-Security-Policy"] = ( + "default-src 'self'; " + "script-src 'self' 'unsafe-inline'; " + "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; " + "font-src 'self' https://fonts.gstatic.com; " + "img-src 'self' data:; " + "connect-src 'self'; " + "frame-ancestors 'none'; " + "base-uri 'self'" + ) + # HSTS uniquement en HTTPS (sinon on risquerait de bloquer un user en HTTP) + if request.url.scheme == "https": + response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" + return response + + @app.on_event("startup") async def _log_config_at_startup(): cfg = load_config() @@ -59,13 +125,61 @@ _USERNAME_RE = re.compile(r"^[a-zA-Z0-9_.\-]{1,64}$") _EMAIL_RE = re.compile(r"^([a-zA-Z0-9_.\-]{1,64})@([a-zA-Z0-9.\-]{1,255})$") _HHMM_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$") +# Échappement HTML pour les fragments construits en f-string hors templates Jinja +# (autoescape ne s'applique qu'aux templates). Defense-in-depth contre le XSS stocké. +_e = html.escape + # ── Auth helpers ────────────────────────────────────────────────────────────── +def _sign_user_id(user_id: str) -> str: + """Return the signed session cookie value: `.`. + + The signature authenticates the user_id without storing server-side session + state. The HMAC key lives in /data/secret.key (see load_or_create_secret). + """ + secret = load_or_create_secret() + sig = hmac.new(secret, user_id.encode(), hashlib.sha256).hexdigest() + return f"{user_id}.{sig}" + + +def _verify_signed_cookie(raw: str) -> str | None: + """Return the user_id if the cookie carries a valid signature, else None.""" + if not raw or "." not in raw: + return None + user_id, _, sig = raw.rpartition(".") + if not user_id or not sig or not _USERNAME_RE.match(user_id): + return None + secret = load_or_create_secret() + expected = hmac.new(secret, user_id.encode(), hashlib.sha256).hexdigest() + # Constant-time comparison to avoid leaking signature bytes via timing. + if hmac.compare_digest(sig, expected): + return user_id + return None + + def _get_user(request: Request) -> str | None: - """Extract current user from cookie.""" - uid = request.cookies.get("user_id", "").strip() - return uid if _USERNAME_RE.match(uid) else None + """Extract current user from signed cookie.""" + return _verify_signed_cookie(request.cookies.get("user_id", "")) + + +def _set_auth_cookie(resp: Response, user_id: str, request: Request | None = None) -> None: + """Set the signed, HttpOnly, SameSite session cookie. + + Le drapeau `secure` est positionné automatiquement si la requête courante + est en HTTPS, ou si config.force_secure_cookies est vrai. Cela évite que + le cookie ne transite en clair sur un downgrade HTTP. + """ + secure = False + if request is not None and request.url.scheme == "https": + secure = True + if load_config().get("force_secure_cookies"): + secure = True + resp.set_cookie( + "user_id", _sign_user_id(user_id), + httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30, + secure=secure, + ) def allowed_email_domain() -> str: @@ -80,6 +194,84 @@ def smtp_config() -> dict: return load_config().get("smtp", {}) +# ── Rate-limiting des routes d'authentification ─────────────────────────────── +# Compteur simple en mémoire, fenêtre glissante de LOGIN_WINDOW_S secondes. +# Suffit pour bloquer le brute-force / mail-bombing sur une app mono-utilisateur. +# Reset entre tests via main._LOGIN_ATTEMPTS.clear(). + +_LOGIN_ATTEMPTS: dict[str, list[float]] = defaultdict(list) +LOGIN_WINDOW_S = 60 +DEFAULT_LOGIN_MAX_PER_MIN = 10 + + +def _check_login_rate_limit(key: str, max_per_min: int | None = None) -> bool: + """Retourne True si la tentative est autorisée, False sinon. + + max_per_min <= 0 désactive la limite (utile pour tests ou config). + """ + if max_per_min is None: + max_per_min = int(load_config().get("login_rate_limit_per_min", DEFAULT_LOGIN_MAX_PER_MIN)) + if max_per_min <= 0: + return True + now = time.monotonic() + cutoff = now - LOGIN_WINDOW_S + recent = [t for t in _LOGIN_ATTEMPTS[key] if t > cutoff] + _LOGIN_ATTEMPTS[key] = recent + if len(recent) >= max_per_min: + return False + recent.append(now) + return True + + +def _client_ip(request: Request) -> str: + """Retourne l'IP du client (sans confiance X-Forwarded-For par défaut).""" + return request.client.host if request.client else "unknown" + + +def _safe_base_url(request: Request) -> str | None: + """Retourne une URL de base fiable pour construire des liens (emails, etc.). + + Anti open-redirect / phishing : le header Host est contrôlable par le client + (curl/DevTools), donc `request.base_url` ne doit pas être utilisée telle + quelle dans un email envoyé à l'utilisateur, sinon un attaquant peut faire + envoyer un mail légitime contenant `https://phishing.attacker/set-password/` + avec un token réellement valide. + + Stratégie : + - Si `base_url` est défini en config, l'utiliser (schéma http/https requis). + - Sinon, accepter le Host seulement s'il correspond à `allowed_email_domain` + (suffixe strict) ou à localhost (utile pour les tests/dev locaux). + - Sinon, retourner None : l'appelant doit refuser l'envoi. + """ + cfg = load_config() + configured = (cfg.get("base_url") or "").strip().rstrip("/") + if configured: + try: + parts = urlsplit(configured) + if parts.scheme in ("http", "https") and parts.hostname: + return configured + except ValueError: + pass + logger.warning("base_url configurée invalide: %r", configured) + return None + + host = (request.headers.get("host") or "").strip() + if not host: + return None + host_l = host.lower() + hostname = host_l.split(":", 1)[0] + scheme = "https" if request.url.scheme == "https" else "http" + + if hostname in ("localhost", "127.0.0.1", "::1"): + return f"{scheme}://{host}" + + allowed = (cfg.get("allowed_email_domain") or "").strip().lower() + if allowed and (host_l == allowed or host_l.endswith("." + allowed)): + return f"{scheme}://{host}" + + return None + + def _require_user(request: Request): """Return user_id or redirect to login.""" uid = _get_user(request) @@ -111,7 +303,9 @@ def _build_soldes_ctx(year, week, conges, h_jour, user_id): "aprem_entree": None, "aprem_sortie": None}) for d in dates ] - return {"result": compute_week(pointages, conges, h_jour, load_plages(user_id))} + result = compute_week(pointages, conges, h_jour, load_plages(user_id)) + result["heures_jour"] = minutes_to_hhmm(h_jour) + return {"result": result} def _build_jour_ctx(date, conges, h_jour, user_id): @@ -136,16 +330,16 @@ def _build_jour_ctx(date, conges, h_jour, user_id): def _oob_calc(date, jour): - trav_inner = f'{"—" if not jour["travaille_min"] else jour["travaille"]}' + trav_inner = f'{"—" if not jour["travaille_min"] else _e(jour["travaille"])}' if jour["travaille_min"]: cls = "delta-pos" if jour["delta_min"] > 0 else ("delta-neg" if jour["delta_min"] < 0 else "delta-zero") sign = "+" if jour["delta_min"] > 0 else "" - delta_inner = f'{sign}{jour["delta"]}' + delta_inner = f'{sign}{_e(jour["delta"])}' else: delta_inner = '—' return ( - f'{trav_inner}' - f'{delta_inner}' + f'{trav_inner}' + f'{delta_inner}' ) @@ -163,14 +357,14 @@ def _oob_week_extras(result): cumul_inner = f'{sign}{jour["delta_cumul"]}' else: cumul_inner = '—' - html += f'{cumul_inner}' + html += f'{cumul_inner}' parts = [] if jour.get("cible_matin"): - parts.append(f'M {jour["cible_matin"]}') + parts.append(f'M {_e(jour["cible_matin_entree"])} → {_e(jour["cible_matin"])}') if jour.get("cible_aprem"): - parts.append(f'A {jour["cible_aprem"]}') + parts.append(f'A {_e(jour["cible_aprem_entree"])} → {_e(jour["cible_aprem"])}') cible_inner = " ".join(parts) if parts else '—' - html += f'{cible_inner}' + html += f'{cible_inner}' return html @@ -189,19 +383,19 @@ def _oob_time_cells(date, jour, plages): matin_cls = "td-fill cg" if matin_cg else ("td-fill warn" if matin_warn else "td-fill") aprem_cls = "td-fill cg" if aprem_cg else ("td-fill warn" if aprem_warn else "td-fill") matin_html = ( - f'
' + f'
' f'
' - f'' + f'' f'→' - f'' + f'' f'
' ) aprem_html = ( - f'
' + f'
' f'
' - f'' + f'' f'→' - f'' + f'' f'
' ) return matin_html + aprem_html @@ -216,11 +410,11 @@ def _htmx_conge_and_soldes(request, date, d, conges, h_jour, user_id): btn_am_on = " on" if "aprem" in cg else "" btn_j_on = " jour-on" if "jour" in cg else "" btns_html = ( - f'
' + f'
' f'
' - f'' - f'' - f'' + f'' + f'' + f'' f'
' ) soldes_ctx = _build_soldes_ctx(iso.year, iso.week, conges, h_jour, user_id) @@ -266,6 +460,14 @@ def login_page(request: Request): @app.post("/login", response_class=HTMLResponse) def login_submit(request: Request, email: str = Form("")): + # Rate-limit par IP pour empêcher le mail-bombing SMTP et l'énumération. + ip = _client_ip(request) + if not _check_login_rate_limit(f"login:{ip}"): + logger.warning("rate limit /login déclenchée: ip=%r", ip) + return templates.TemplateResponse("login.html", _login_ctx( + request, "email", "", + "Trop de tentatives. Réessaie dans une minute.", + )) email = email.strip().lower() m = _EMAIL_RE.match(email) domain = allowed_email_domain() @@ -285,12 +487,22 @@ def login_submit(request: Request, email: str = Form("")): return templates.TemplateResponse("login.html", _login_ctx(request, "password", email)) # No account yet: (re)send a link to set the password + base_url = _safe_base_url(request) + if not base_url: + logger.warning( + "login refusé (Host non fiable pour générer un lien email): host=%r email=%r", + request.headers.get("host"), email, + ) + return templates.TemplateResponse("login.html", _login_ctx( + request, "email", email, + "Domaine non autorisé pour générer un lien de connexion.", + )) + token = secrets.token_urlsafe(32) expires = (datetime.now() + timedelta(hours=RESET_TOKEN_VALIDITY_HOURS)).isoformat() save_auth(user_id, email=email, token=token, token_expires=expires) logger.info("nouveau token de création de compte pour %r (expire %s)", user_id, expires) - base_url = str(request.base_url).rstrip("/") link = f"{base_url}/set-password/{token}" try: send_mail( @@ -311,7 +523,15 @@ def login_submit(request: Request, email: str = Form("")): @app.post("/login/password", response_class=HTMLResponse) def login_password(request: Request, email: str = Form(""), password: str = Form("")): + # Rate-limit par IP+email pour bloquer le brute-force d'un compte précis. email = email.strip().lower() + ip = _client_ip(request) + if not _check_login_rate_limit(f"pw:{ip}:{email}"): + logger.warning("rate limit /login/password déclenchée: ip=%r email=%r", ip, email) + return templates.TemplateResponse("login.html", _login_ctx( + request, "password", email, + "Trop de tentatives. Réessaie dans une minute.", + )) m = _EMAIL_RE.match(email) domain = allowed_email_domain() logger.info("POST /login/password: email=%r", email) @@ -329,7 +549,7 @@ def login_password(request: Request, email: str = Form(""), password: str = Form logger.info("connexion réussie: user_id=%r", user_id) resp = RedirectResponse("/", status_code=302) - resp.set_cookie("user_id", user_id, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30) + _set_auth_cookie(resp, user_id, request) return resp @@ -371,7 +591,7 @@ def set_password_submit(request: Request, token: str, password: str = Form(""), logger.info("mot de passe défini pour %r", user_id) resp = RedirectResponse("/", status_code=302) - resp.set_cookie("user_id", user_id, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30) + _set_auth_cookie(resp, user_id, request) return resp @@ -445,7 +665,10 @@ def save_pointage( return RedirectResponse("/login", status_code=302) def clean(v): - v = v.strip(); return v if v else None + # Refuse tout ce qui n'est pas un HHMM strict : empêche le XSS stocké car la + # valeur est réinjectée telle quelle dans des f-strings HTML hors templates. + v = v.strip() + return v if v and _HHMM_RE.match(v) else None d = datetime.strptime(date, "%Y-%m-%d").date() iso = d.isocalendar() @@ -556,12 +779,16 @@ def settings_page(request: Request): notif = load_notif_config(user_id) plages = load_plages(user_id) - base_url = str(request.base_url).rstrip("/") + # Les URLs Tasker sont affichées à l'utilisateur lui-même : moins sensible qu'un + # email, mais on utilise quand même le helper (fallback sur base_url request si + # non configurée, car l'utilisateur voit sa propre URL). + base_url = _safe_base_url(request) or str(request.base_url).rstrip("/") return templates.TemplateResponse("settings.html", { "request": request, "current_user": user_id, "ntfy_topic": notif.get("ntfy_topic") or "", "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_fin_h": notif.get("rappel_fin_h", 20), "arrivee_url": f"{base_url}/presence/{notif['token']}/arrivee", @@ -577,6 +804,7 @@ def settings_page(request: Request): def settings_save( request: Request, ntfy_topic: str = Form(""), ntfy_server: str = Form(""), rappel_debut_h: int = Form(7), rappel_fin_h: int = Form(20), + ntfy_token: str = Form(""), ): user_id = _get_user(request) if not user_id: @@ -585,14 +813,86 @@ def settings_save( rappel_debut_h = max(0, min(23, rappel_debut_h)) rappel_fin_h = max(rappel_debut_h + 1, min(24, rappel_fin_h)) + ntfy_server = ntfy_server.strip() + if ntfy_server and not is_safe_ntfy_server(ntfy_server): + logger.warning("réglages notif rejetés (SSRF): user_id=%r ntfy_server=%r", user_id, ntfy_server) + return RedirectResponse("/settings?ssrf_error=1", status_code=303) + 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, + "(défini)" if ntfy_token.strip() else "(vide)", + ) + save_notif_config(user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h, ntfy_token) + append_log( + user_id, "settings", "Réglages ntfy", + f"topic={ntfy_topic or '(vide)'} server={ntfy_server} plage={rappel_debut_h}h-{rappel_fin_h}h " + f"jeton={'défini' if ntfy_token.strip() else '(vide)'}", ) - save_notif_config(user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h) return RedirectResponse("/settings?saved=1", status_code=303) +@app.post("/settings/test-ntfy", response_class=HTMLResponse) +def settings_test_ntfy(request: Request): + user_id = _get_user(request) + if not user_id: + return RedirectResponse("/login", status_code=302) + + notif = load_notif_config(user_id) + topic = notif.get("ntfy_topic") + server = notif.get("ntfy_server", "https://ntfy.sh") + + if not topic: + logger.warning("test ntfy: pas de topic configuré pour %r", user_id) + return HTMLResponse( + '
' + 'Renseigne et enregistre un topic ntfy avant de tester.' + '
' + ) + + logger.info("test ntfy: envoi de test pour %r sur %r/%r", user_id, server, topic) + ok = send_ntfy( + server, topic, + "Ceci est une notification de test — ta configuration ntfy fonctionne.", + "Test pointeuse", user_id=user_id, token=notif.get("ntfy_token", ""), + ) + cls = "settings-saved" if ok else "settings-error" + msg = ("Notification envoyée. Vérifie ton téléphone, elle doit arriver en quelques secondes." + if ok else + "Échec de l'envoi. Vérifie le serveur et le topic, et l'optimisation batterie de l'app ntfy.") + return HTMLResponse( + f'
{msg}
' + ) + + +@app.post("/settings/regenerate-presence-token", response_class=HTMLResponse) +def settings_regenerate_presence_token(request: Request): + user_id = _get_user(request) + if not user_id: + return RedirectResponse("/login", status_code=302) + + new_token = regenerate_presence_token(user_id) + logger.warning("token de présence régénéré pour %r", user_id) + append_log(user_id, "settings", "Token présence régénéré", "Les URLs Tasker ont été invalidées et remplacées.") + + base_url = _safe_base_url(request) or str(request.base_url).rstrip("/") + arrivee_url = f"{base_url}/presence/{new_token}/arrivee" + depart_url = f"{base_url}/presence/{new_token}/depart" + return HTMLResponse( + f'
' + f'
' + f'Arrivée' + f'{arrivee_url}' + f'
' + f'
' + f'Départ' + f'{depart_url}' + f'
' + f'
Token régénéré. Mets à jour les URLs dans Automation.
' + f'
' + ) + + @app.post("/settings/plages", response_class=HTMLResponse) def settings_plages_save( request: Request, @@ -626,28 +926,63 @@ def settings_plages_save( logger.info("réglages plages mis à jour: user_id=%r plages=%r", user_id, fields) save_plages(user_id, fields) + append_log( + user_id, "settings", "Plages horaires", + f"matin {fields['matin_debut']}→{fields['matin_fin']}, " + f"aprem {fields['aprem_debut']}→{fields['aprem_fin']}, " + f"reprise {fields['pause_dejeuner_fin']}" + + (f", visées {fields['arrivee_visee']}→{fields['depart_vise']}" if fields['arrivee_visee'] or fields['depart_vise'] else ""), + ) return RedirectResponse("/settings?plages_saved=1", status_code=303) -@app.get("/presence/{token}/arrivee") +@app.post("/presence/{token}/arrivee") def presence_arrivee(token: str): user_id = find_user_by_token(token) if not user_id: - logger.warning("GET /presence/.../arrivee: token inconnu") + logger.warning("POST /presence/.../arrivee: token inconnu") return JSONResponse({"error": "invalid token"}, status_code=404) logger.info("présence: arrivée détectée pour %r", user_id) save_presence(user_id, present=True, since=datetime.now().isoformat()) + append_log(user_id, "presence", "arrivée", "Présence détectée par Tasker") return {"ok": True} -@app.get("/presence/{token}/depart") +@app.post("/presence/{token}/depart") def presence_depart(token: str): user_id = find_user_by_token(token) if not user_id: - logger.warning("GET /presence/.../depart: token inconnu") + logger.warning("POST /presence/.../depart: token inconnu") return JSONResponse({"error": "invalid token"}, status_code=404) logger.info("présence: départ détecté pour %r", user_id) save_presence(user_id, present=False, since=datetime.now().isoformat()) + append_log(user_id, "presence", "départ", "Départ détecté par Tasker") return {"ok": True} +@app.get("/logs", response_class=HTMLResponse) +def logs_page(request: Request): + user_id = _get_user(request) + if not user_id: + return RedirectResponse("/login", status_code=302) + + logs = load_logs(user_id) + return templates.TemplateResponse("logs.html", { + "request": request, + "current_user": user_id, + "logs": logs, + "filter": request.query_params.get("filter", "all"), + }) + + +@app.get("/aide", response_class=HTMLResponse) +def aide_page(request: Request): + user_id = _get_user(request) + if not user_id: + return RedirectResponse("/login", status_code=302) + return templates.TemplateResponse("aide.html", { + "request": request, + "current_user": user_id, + }) + + diff --git a/app/models.py b/app/models.py index b9abf74..5f785ff 100644 --- a/app/models.py +++ b/app/models.py @@ -1,6 +1,7 @@ """Flat-file storage: JSON files per week for pointages, one file for congés.""" import json import secrets +from datetime import datetime from pathlib import Path from calcul import DEFAULT_PLAGES @@ -9,9 +10,11 @@ DATA_DIR = Path("/data") CONFIG_FILE = DATA_DIR / "config.json" DEFAULT_CONFIG = { - "heures_jour": 7.8, # 7h48 = 39h/semaine + "heures_semaine": 39.0, # heures hebdomadaires visées + "jours_par_semaine": 5, # jours travaillés dans la semaine — heures_jour en est dérivé "log_level": "DEBUG", # DEBUG, INFO, WARNING ou ERROR "allowed_email_domain": "", # ex: "sdv.fr" — seules ces adresses peuvent se connecter + "base_url": "", # ex: "https://pointeuse.sdv.fr" — utilisée pour les liens emails (anti open-redirect via Host header) "mail_from": "", # ex: "no-reply@sdv.fr" — expéditeur des emails de connexion "smtp": { "host": "", # ex: "smtp.sdv.fr" @@ -68,8 +71,35 @@ def save_config(cfg: dict): CONFIG_FILE.write_text(json.dumps(cfg, indent=2, ensure_ascii=False)) +def load_or_create_secret() -> bytes: + """Charge ou crée la clé secrète utilisée pour signer les cookies de session. + + Persistée hors config pour ne pas être réinitialisée à chaque modification des + réglages et rester stable entre les redémarrages. + """ + secret_file = DATA_DIR / "secret.key" + secret_file.parent.mkdir(parents=True, exist_ok=True) + if secret_file.exists(): + try: + return bytes.fromhex(secret_file.read_text().strip()) + except ValueError: + pass + secret = secrets.token_bytes(32) + secret_file.write_text(secret.hex()) + return secret + + def heures_jour_min() -> int: - return int(float(load_config().get("heures_jour", 7.8)) * 60) + """Minutes attendues par jour, dérivées d'`heures_semaine` et `jours_par_semaine`. + + Rétrocompat : si `heures_semaine` est absent mais `heures_jour` présent, + on utilise l'ancienne clé. + """ + cfg = load_config() + jours = int(cfg.get("jours_par_semaine", 5)) or 5 + if "heures_semaine" in cfg: + return int(float(cfg["heures_semaine"]) * 60 / jours) + return int(float(cfg.get("heures_jour", 7.8)) * 60) # ── Pointages ───────────────────────────────────────────────────────────────── @@ -155,11 +185,12 @@ DEFAULT_RAPPEL_FIN_H = 20 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) if f.exists(): cfg = json.loads(f.read_text()) cfg.setdefault("ntfy_server", DEFAULT_NTFY_SERVER) + cfg.setdefault("ntfy_token", "") cfg.setdefault("rappel_debut_h", DEFAULT_RAPPEL_DEBUT_H) cfg.setdefault("rappel_fin_h", DEFAULT_RAPPEL_FIN_H) if cfg.get("token"): @@ -167,6 +198,7 @@ def load_notif_config(user_id: str) -> dict: else: cfg = { "ntfy_topic": None, "ntfy_server": DEFAULT_NTFY_SERVER, + "ntfy_token": "", "rappel_debut_h": DEFAULT_RAPPEL_DEBUT_H, "rappel_fin_h": DEFAULT_RAPPEL_FIN_H, } cfg["token"] = secrets.token_hex(16) @@ -178,10 +210,12 @@ def load_notif_config(user_id: str) -> dict: def save_notif_config( 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, + ntfy_token: str = "", ): cfg = load_notif_config(user_id) cfg["ntfy_topic"] = ntfy_topic.strip() or None 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_fin_h"] = rappel_fin_h _ensure_user_dirs(user_id) @@ -196,6 +230,15 @@ def find_user_by_token(token: str) -> str | None: return None +def regenerate_presence_token(user_id: str) -> str: + """Génère un nouveau token de présence (invalide les URLs Tasker précédentes).""" + cfg = load_notif_config(user_id) + cfg["token"] = secrets.token_hex(16) + _ensure_user_dirs(user_id) + _notif_file(user_id).write_text(json.dumps(cfg, indent=2, ensure_ascii=False)) + return cfg["token"] + + def _presence_file(user_id: str) -> Path: return _user_root(user_id) / "presence.json" @@ -254,3 +297,47 @@ def toggle_conge(date: str, type_conge: str, user_id: str = ""): conges.sort(key=lambda c: (c["date"], c["type"])) save_conges(conges, user_id) + + +# ── Journaux (notifications + présence) ─────────────────────────────────────── + +LOG_TYPES = ("notif", "presence", "settings") +MAX_LOGS = 500 + + +def _logs_file(user_id: str) -> Path: + return _user_root(user_id) / "logs.json" + + +def append_log(user_id: str, entry_type: str, event: str, message: str = "", ts: str | None = None): + """Ajoute une entrée au journal de l'utilisateur (rotation sur MAX_LOGS).""" + assert entry_type in LOG_TYPES, f"type de journal invalide: {entry_type!r}" + f = _logs_file(user_id) + logs = [] + if f.exists(): + try: + logs = json.loads(f.read_text()) + except json.JSONDecodeError: + logs = [] + logs.append({ + "ts": ts or datetime.now().isoformat(timespec="seconds"), + "type": entry_type, + "event": event, + "message": message, + }) + if len(logs) > MAX_LOGS: + logs = logs[-MAX_LOGS:] + _ensure_user_dirs(user_id) + f.write_text(json.dumps(logs, indent=2, ensure_ascii=False)) + + +def load_logs(user_id: str) -> list[dict]: + """Retourne les entrées du journal, des plus récentes aux plus anciennes.""" + f = _logs_file(user_id) + if not f.exists(): + return [] + try: + logs = json.loads(f.read_text()) + except json.JSONDecodeError: + return [] + return list(reversed(logs)) diff --git a/app/notifications.py b/app/notifications.py index b2c2ac1..0588d16 100644 --- a/app/notifications.py +++ b/app/notifications.py @@ -1,29 +1,84 @@ """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 +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 send_ntfy(server: str, topic: str, message: str, title: str): +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={"Title": title.encode("utf-8"), "Priority": "high"}, + 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: @@ -37,6 +92,7 @@ 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: @@ -55,10 +111,18 @@ async def check_user(user_id: str, now: datetime): 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") + 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") + 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(): diff --git a/app/stats.py b/app/stats.py index 36306f5..c9c3ec3 100644 --- a/app/stats.py +++ b/app/stats.py @@ -2,7 +2,9 @@ import statistics as _st from datetime import datetime, timedelta, date as _date -from models import load_week_pointages, load_conges, list_all_weeks, load_config, load_plages +from models import ( + load_week_pointages, load_conges, list_all_weeks, load_plages, heures_jour_min, +) from calcul import hhmm_to_minutes, minutes_to_hhmm, heures_travaillees, heures_dues MOIS_FR = [ @@ -58,8 +60,7 @@ def _compliance_over(all_days: list[tuple], cg_set: set, plages: dict) -> dict: def compute_all_stats(user_id: str = "") -> dict: all_weeks = sorted(list_all_weeks(user_id)) conges = load_conges(user_id) - cfg = load_config() - h_jour = int(float(cfg.get("heures_jour", 7.8)) * 60) + h_jour = heures_jour_min() plages = load_plages(user_id) cg_set = {(c["date"], c["type"]) for c in conges} diff --git a/app/templates/_ligne.html b/app/templates/_ligne.html index f444f52..2e1bbb6 100644 --- a/app/templates/_ligne.html +++ b/app/templates/_ligne.html @@ -18,10 +18,10 @@
- → -
@@ -30,10 +30,10 @@
- → -
@@ -70,9 +70,9 @@ {%- if j.cible_matin or j.cible_aprem -%} - {%- if j.cible_matin -%}M {{ j.cible_matin }}{%- endif -%} + {%- if j.cible_matin -%}M {{ j.cible_matin_entree }} → {{ j.cible_matin }}{%- endif -%} {%- if j.cible_matin and j.cible_aprem %} {% endif -%} - {%- if j.cible_aprem -%}A {{ j.cible_aprem }}{%- endif -%} + {%- if j.cible_aprem -%}A {{ j.cible_aprem_entree }} → {{ j.cible_aprem }}{%- endif -%} {%- else -%} — {%- endif %} diff --git a/app/templates/_soldes.html b/app/templates/_soldes.html index c01b993..9eee529 100644 --- a/app/templates/_soldes.html +++ b/app/templates/_soldes.html @@ -4,6 +4,10 @@
Travaillé
{{ result.total_travaille }}
+
+
Dû par jour
+
{{ result.heures_jour }}
+
Dû cette semaine
{{ result.total_du }}
diff --git a/app/templates/aide.html b/app/templates/aide.html new file mode 100644 index 0000000..1378464 --- /dev/null +++ b/app/templates/aide.html @@ -0,0 +1,229 @@ +{% extends "base.html" %} +{% block content %} +
+
+ Aide +
Configurer les notifications ntfy et la détection de présence (Automation)
+
+
+ +
+ Principe +

+ L'app peut te relancer automatiquement si tu es détecté présent au bureau sans avoir pointé + le matin, ou parti sans avoir pointé la sortie. Ça repose sur deux briques : +

+
    +
  • + Réception des notifications : ntfy, + un service gratuit de notifications push par topic. +
  • +
  • + Détection de présence : ton téléphone appelle une URL de l'app quand il entre/sort + du bureau. L'app fait le reste (vérifie toutes les 5 min, dans la plage horaire réglée + sur /settings, en semaine, si un pointage manque). +
  • +
+

+ Avant de commencer : va sur /settings, choisis un nom de topic ntfy + secret (ex: prenom-pointeuse-a1b2, pas devinable), enregistre-le. La page affiche + ensuite tes deux URLs personnelles Arrivée et Départ — elles contiennent un token + secret, ne les partage pas. +

+
+ +
+ 1. Recevoir les notifications (ntfy) +

+ Installer l'app ntfy sur le téléphone, puis s'abonner au topic choisi dans /settings. +

+
    +
  1. + Installe l'app ntfy depuis le + Play Store + ou F-Droid. +
  2. +
  3. + Ouvre l'app, bouton + → entre le nom de topic choisi sur + /settings → valide. Laisse le serveur par défaut (ntfy.sh). +
  4. +
  5. + Teste l'envoi depuis un navigateur : ouvre + https://ntfy.sh/<ton-topic> et publie un message. La notif doit + arriver sur le téléphone en quelques secondes. +
  6. +
  7. + Important : sur Android, désactive l'optimisation de la batterie pour ntfy + (Réglages → Applications → ntfy → Batterie → "Aucune restriction"). Sinon Android + peut couper les notifications en arrière-plan. +
  8. +
  9. + Serveur protégé par jeton : si ton instance ntfy renvoie 403 Forbidden, + 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 Jeton d'accès sur + /settings. Il est envoyé comme + Authorization: Bearer ... à chaque publication. +
  10. +
+

+ Astuce : si tu veux ton propre serveur ntfy (auto-hébergé, pour éviter le topic public), + installe-le (doc officielle) + et renseigne son URL dans le champ Serveur ntfy sur /settings. +

+
+ +
+ 2. Détecter la présence (Automation) +

+ Créer une automatisation Automation + (app FOSS, F-Droid) déclenchée par le Wi-Fi du bureau ou par une géofence : + à l'entrée, appeler l'URL « Arrivée » ; à la sortie, l'URL « Départ ». +

+
    +
  1. + Installe Automation + depuis F-Droid (évite la version Play Store, amputée de la localisation). +
  2. +
  3. + Autorise la localisation en arrière-plan : Réglages → Applications → + Automation → Autorisations → Position → Toujours. Sans ça, Android + coupe la détection quand l'app n'est pas au premier plan. +
  4. +
  5. + Désactive l'optimisation batterie pour Automation (même menu → + Batterie → « Aucune restriction »). +
  6. +
  7. + Crée la règle Arrivée bureau — Automation → Rules → + : +
      +
    • Trigger : Wifi connection → entre le SSID du Wi-Fi du bureau (méthode recommandée), + ou Location → adresse du bureau, rayon ~100–150 m, événement Enter (géofence).
    • +
    • Action : Trigger URL, method POST, + URL = ton lien Arrivée copié depuis /settings.
    • +
    +
  8. +
  9. + Crée la règle symétrique Sortie bureau : même SSID en sens déconnexion + (ou même géofence, événement Exit), action Trigger URL POST + vers ton lien Départ. +
  10. +
+

+ Au bureau, la méthode Wi-Fi est généralement plus réactive et plus économe en + batterie que la géofence GPS. +

+

+ Si tu changes de téléphone ou réinstalles, les tokens restent valides — il suffit de + recréer les règles Automation et l'abonnement ntfy. +

+
+ +
+ 3. Vérifier +
    +
  1. + Force une entrée/sortie de zone ou de Wi-Fi (ou, dans Automation, ouvre une règle + et utilise le bouton « Test », icône lecture). La réponse doit être {"ok":true}. +
  2. +
  3. + Va sur /logs : tu dois voir une entrée "Présence / arrivée" ou + "Présence / départ" correspondante. +
  4. +
  5. + Simule une journée sans pointage matin pendant que tu es marqué "présent" : une notif + ntfy doit arriver dans les 5 minutes. Elle apparaîtra aussi dans + /logs sous "Notification / envoyée". +
  6. +
+
+ +
+ Dépannage +
    +
  • + Pas de notif ntfy : vérifie l'optimisation batterie sur l'app ntfy, et que le topic + est bien le même côté serveur et côté téléphone (sensible à la casse). +
  • +
  • + Pas de déclenchement Automation : dans Automation, vérifie que la règle est + bien activée et utilise le bouton « Test » pour isoler le problème (réseau vs + géoloc vs Wi-Fi). Vérifie aussi que l'optimisation batterie est bien désactivée + pour l'app. +
  • +
  • + Logs vides sur /logs : les entrées ne s'écrivent que quand un événement réel se + produit (notif envoyée/échouée, arrivée/départ détecté). Force un test (étape 3). +
  • +
+
+ + +{% endblock %} diff --git a/app/templates/base.html b/app/templates/base.html index b00946c..991a40a 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -144,7 +144,7 @@ /* ── Soldes ── */ .soldes { display: grid; - grid-template-columns: repeat(3, 1fr); + grid-template-columns: repeat(4, 1fr); border: 1.5px solid var(--rule); background: var(--surface); } @@ -166,7 +166,7 @@ } .solde-num { font-family: var(--mono); - font-size: 1.9rem; + font-size: 1.5rem; font-weight: 500; letter-spacing: -.03em; line-height: 1; @@ -188,6 +188,7 @@ /* ── Table ── */ .table-section { overflow-x: auto; -webkit-overflow-scrolling: touch; } table { + width: 100%; border-collapse: collapse; border: 1.5px solid var(--rule); background: var(--surface); @@ -195,7 +196,7 @@ } .th-group th { - padding: .45rem .85rem .25rem; + padding: .45rem .65rem .25rem; font-family: var(--mono); font-size: 9px; letter-spacing: .12em; @@ -222,7 +223,7 @@ .th-group .g-conge { text-align: center; } .th-sub th { - padding: .25rem .85rem .5rem; + padding: .25rem .65rem .5rem; font-size: 10px; color: rgba(107,97,82,.65); border-bottom: 2px solid var(--text); @@ -258,7 +259,7 @@ tbody tr:not(:last-child) td { border-bottom: 1px solid var(--rule-l); } td { - padding: .6rem .85rem; + padding: .5rem .65rem; vertical-align: middle; } @@ -267,7 +268,7 @@ .td-fill { display: flex; align-items: center; - padding: .6rem .85rem; + padding: .5rem .65rem; } td.c-matin .td-fill { border-left: 3px solid rgba(47,95,160,.2); } td.c-aprem .td-fill { border-left: 3px solid rgba(184,106,8,.2); } @@ -284,11 +285,11 @@ .jour-nom { display: block; font-weight: 500; font-size: 13px; } .jour-date { display: block; font-family: var(--mono); font-size: 11px; color: var(--muted); } - .time-pair { display: flex; align-items: center; gap: 4px; justify-content: center; } + .time-pair { display: flex; align-items: center; gap: 3px; justify-content: center; } .time-sep { color: var(--rule); font-size: 11px; flex-shrink: 0; } .time-input { - width: 80px; - padding: .28rem .4rem; + width: 70px; + padding: .28rem .35rem; border: 1px solid var(--rule); border-radius: 0; font-family: var(--mono); @@ -316,10 +317,10 @@ /* Cumulative delta and sortie cible columns */ .col-cumul { text-align: right; padding-left: .5rem; } - .col-cible { text-align: right; padding-left: .5rem; } + .col-cible { text-align: right; padding-left: .5rem; white-space: normal; min-width: 110px; } .g-cumul { text-align: right; } .g-cible { text-align: right; } - .cible-num { font-family: var(--mono); font-size: 12px; color: var(--accent); font-weight: 500; } + .cible-num { font-family: var(--mono); font-size: 12px; color: var(--accent); font-weight: 500; display: inline-block; } .col-save { text-align: center; padding: .4rem .3rem; } .g-save { background: var(--ground); } @@ -1032,7 +1033,9 @@ {% if current_user is defined and current_user %} Semaine courante Statistiques + Journaux Réglages + Aide {{ current_user }} Déconnexion {% endif %} diff --git a/app/templates/logs.html b/app/templates/logs.html new file mode 100644 index 0000000..70dc184 --- /dev/null +++ b/app/templates/logs.html @@ -0,0 +1,145 @@ +{% extends "base.html" %} +{% block content %} +
+
+ Journaux +
+ Notifications ntfy et détections de présence (Automation) — {{ current_user }} +
+
+
+ +
+ {% set counts = {'all': logs|length, 'notif': logs | selectattr('type', 'equalto', 'notif') | list | length, 'presence': logs | selectattr('type', 'equalto', 'presence') | list | length, 'settings': logs | selectattr('type', 'equalto', 'settings') | list | length} %} + + Tout {{ counts.all }} + + + Notifications {{ counts.notif }} + + + Présence {{ counts.presence }} + + + Réglages {{ counts.settings }} + +
+ +{% set visible = logs if filter == 'all' else logs | selectattr('type', 'equalto', filter) | list %} + +{% if visible %} +
+ {% for entry in visible %} +
+
{{ entry.ts }}
+
{{ {'notif': 'Notification', 'presence': 'Présence', 'settings': 'Réglages'}[entry.type] }}
+
+ {{ entry.event }} + {% if entry.message %}{{ entry.message }}{% endif %} +
+
+ {% endfor %} +
+{% else %} +
Aucun événement enregistré pour le moment.
+{% endif %} + + +{% endblock %} diff --git a/app/templates/semaine.html b/app/templates/semaine.html index 478e7b3..a498306 100644 --- a/app/templates/semaine.html +++ b/app/templates/semaine.html @@ -53,7 +53,7 @@ Travaillé Δ jour Δ sem. - Fin cible + Cible Congé diff --git a/app/templates/settings.html b/app/templates/settings.html index 91fb8bc..1c0a76f 100644 --- a/app/templates/settings.html +++ b/app/templates/settings.html @@ -51,11 +51,12 @@
- Notifications (ntfy) + Notifications (ntfy) Aide

Crée un topic sur ton serveur ntfy (public ou auto-hébergé), un identifiant secret que toi seul connais, abonne-toi dessus avec - l'app ntfy sur ton téléphone, puis renseigne les deux champs ci-dessous. + l'app ntfy sur ton téléphone, puis renseigne les deux champs ci-dessous. Guide complet sur + la page Aide.

{% if saved %} @@ -83,6 +84,16 @@ placeholder="ex: antoine-pointeuse-x7f2" autocomplete="off" > + +
+
+ + +
- Détection de présence (Tasker) + Détection de présence (Automation) Aide

- Crée une automatisation Tasker déclenchée par une géofence autour du bureau : à l'entrée de la - zone, appelle l'URL "arrivée" ; à la sortie, l'URL "départ". Tant que tu es détecté présent sans - avoir pointé l'entrée du matin, ou parti sans avoir pointé la sortie, tu reçois un rappel toutes les - 5 minutes, dans la plage horaire réglée ci-dessus (jours ouvrés). + Crée une automatisation Automation + déclenchée par le Wi-Fi du bureau (ou une géofence) : à la connexion/entrée, appelle l'URL « arrivée » ; + à la déconnexion/sortie, l'URL « départ ». Tant que tu es détecté présent sans + avoir pointé l'entrée du matin, ou parti sans avoir pointé la sortie, tu reçois un rappel + selon l'intervalle réglé ci-dessus, dans la plage horaire configurée (jours ouvrés). + Étapes détaillées sur la page Aide.

-
- Arrivée - {{ arrivee_url }} -
-
- Départ - {{ depart_url }} +
+
+ Arrivée + {{ arrivee_url }} +
+
+ Départ + {{ depart_url }} +

Ces URLs sont secrètes (elles contiennent ton token) — ne les partage pas. + En cas de compromission, régénère le token : les anciennes URLs cesseront de fonctionner + et tu devras mettre à jour Tasker avec les nouvelles.

+
{% endblock %} diff --git a/app/tests/conftest.py b/app/tests/conftest.py index a371149..a6148a0 100644 --- a/app/tests/conftest.py +++ b/app/tests/conftest.py @@ -1,9 +1,23 @@ -"""Freezes calcul.py's notion of "today" for deterministic tests.""" +"""Fixtures partagées pour les tests. + +Freeze "today" pour des calculs déterministes ET installe un DATA_DIR +temporaire unique pour tout le process de test, afin que tous les modules +de tests (test_logs, test_security, ...) partagent le même stockage. +""" import datetime as dt_module +import tempfile +from pathlib import Path import pytest import calcul +import models + +# ── DATA_DIR unique pour tout le run ───────────────────────────────────────── +_TMP = Path(tempfile.mkdtemp()) +models.DATA_DIR = _TMP +models.CONFIG_FILE = models.DATA_DIR / "config.json" +models.CONFIG_FILE.write_text('{"heures_semaine": 39.0, "jours_par_semaine": 5, "allowed_email_domain": "x.fr"}') class FrozenDatetime(dt_module.datetime): diff --git a/app/tests/test_calcul.py b/app/tests/test_calcul.py index cba79db..303863f 100644 --- a/app/tests/test_calcul.py +++ b/app/tests/test_calcul.py @@ -113,6 +113,8 @@ def test_compute_week_cible_repartie_sur_seule_demi_journee_restante(freeze_toda assert vendredi["cible_matin"] is None # matin déjà pointé (entrée+sortie), plus rien à y faire assert vendredi["cible_aprem"] == "18:18" + # La préconisation aprem inclut aussi l'heure d'arrivée (reprise par défaut 13:30) + assert vendredi["cible_aprem_entree"] == "13:30" def test_compute_week_semaine_totalement_vide_compte_les_10_demi_journees(freeze_today): @@ -126,10 +128,11 @@ def test_compute_week_semaine_totalement_vide_compte_les_10_demi_journees(freeze result = compute_week(pointages, [], HEURES_JOUR_MIN) assert result["slots_restants"] == 10 assert result["reste_min"] == HEURES_JOUR_MIN * 5 - # Rien n'est encore en retard puisque rien n'a encore été renseigné : le rythme - # normal (à répartir librement sur les 10 demi-journées) suffit, pas de cible imposée. + # Pas de retard (rien n'a encore été renseigné) : le rythme normal suffit, extra = 0. + # Mais chaque demi-journée reste propose son horaire visé (matin 09:00→12:00, + # aprem 13:30→18:18 avec le surplus nominal basculé depuis le matin saturé). assert result["extra_min"] == 0 - assert all(j["cible_matin"] is None and j["cible_aprem"] is None for j in result["jours"]) + assert all(j["cible_matin"] == "12:00" and j["cible_aprem"] == "18:18" for j in result["jours"]) def test_compute_week_jour_futur_non_entame_recoit_une_cible_repartie(freeze_today): @@ -148,30 +151,34 @@ def test_compute_week_jour_futur_non_entame_recoit_une_cible_repartie(freeze_tod assert lundi_suivant["cible_matin"] is not None or lundi_suivant["cible_aprem"] is not None -def test_compute_week_jour_futur_a_egalite_pas_de_cible(freeze_today): - """Quand le rythme normal des demi-journées restantes suffit déjà, aucune cible n'est affichée.""" +def test_compute_week_jour_futur_a_egalite_propose_le_rythme_normal(freeze_today): + """Quand le rythme normal des demi-journées restantes suffit (pas de rattrapage), + les horaires visés du matin et de l'après-midi sont quand même proposés.""" dates_completes = ["2026-07-13", "2026-07-14", "2026-07-15", "2026-07-16", "2026-07-17"] pointages = _semaine_complete(dates_completes) pointages.append(_pointage("2026-07-20")) # lundi suivant, rien pointé — mais la semaine est déjà à l'équilibre result = compute_week(pointages, [], HEURES_JOUR_MIN) lundi_suivant = result["jours"][5] - assert lundi_suivant["cible_matin"] is None - assert lundi_suivant["cible_aprem"] is None + assert result["extra_min"] == 0 + assert lundi_suivant["cible_matin"] == "12:00" + assert lundi_suivant["cible_aprem"] == "18:18" -def test_compute_week_jour_courant_non_entame_rythme_normal_suffit(freeze_today): +def test_compute_week_jour_courant_non_entame_propose_le_rythme_normal(freeze_today): """"Aujourd'hui" sans rien pointé n'est pas traité comme un jour futur figé : ses deux demi-journées entrent dans la répartition. Ici le manque (468 min) correspond exactement - au nominal des deux demi-journées restantes : le rythme normal suffit, pas de cible.""" + au nominal des deux demi-journées restantes : le rythme normal suffit, mais les horaires + visés matin et aprem restent proposés.""" dates_completes = ["2026-07-13", "2026-07-14", "2026-07-15", "2026-07-16"] pointages = _semaine_complete(dates_completes) pointages.append(_pointage("2026-07-17")) result = compute_week(pointages, [], HEURES_JOUR_MIN) vendredi = result["jours"][4] - assert vendredi["cible_matin"] is None - assert vendredi["cible_aprem"] is None + assert result["extra_min"] == 0 + assert vendredi["cible_matin"] == "12:00" + assert vendredi["cible_aprem"] == "18:18" def test_compute_week_demi_journee_en_cours_pas_de_double_comptage(freeze_today): @@ -194,8 +201,9 @@ def test_compute_week_demi_journee_en_cours_pas_de_double_comptage(freeze_today) result = compute_week(pointages, [], HEURES_JOUR_MIN) vendredi = result["jours"][4] assert result["extra_min"] == 60 - # matin saturé au nominal (09:00→12:00), pas de cible ; aprem : 13:30 + 288 + 60 = 19:18 - assert vendredi["cible_matin"] is None + # matin en cours : sa cible correspond au planning nominal (09:00→12:00), pas de rattrapage + # possible (saturé au plafond pause) ; aprem : 13:30 + 288 + 60 = 19:18 + assert vendredi["cible_matin"] == "12:00" assert vendredi["cible_aprem"] == "19:18" @@ -221,7 +229,9 @@ def test_compute_week_cible_matin_ne_mange_jamais_la_pause_de_midi(freeze_today) # 120 min de retard + 54 min de nominal matin qui ne tiennent pas avant la pause # (et ne peuvent pas basculer sur l'après-midi, en congé) assert result["extra_min"] == 174 - assert vendredi["cible_matin"] is None + # Le matin (seul créneau dispo, aprem en congé) est saturé à son plafond 09:00→12:00 : + # cible proposée au rythme normal, mais le rattrapage (174 min) ne tient pas. + assert vendredi["cible_matin"] == "12:00" assert vendredi["cible_aprem"] is None # congé @@ -243,8 +253,8 @@ def test_compute_week_surplus_creneau_sature_reporte_sur_les_autres(freeze_today result = compute_week(pointages, [], HEURES_JOUR_MIN) vendredi = result["jours"][4] assert result["extra_min"] == 80 - # matin saturé au nominal, pas de cible ; aprem : 13:30 + 288 + 80 = 19:38 - assert vendredi["cible_matin"] is None + # matin saturé au nominal (cible au rythme normal 09:00→12:00) ; aprem : 13:30 + 288 + 80 = 19:38 + assert vendredi["cible_matin"] == "12:00" assert vendredi["cible_aprem"] == "19:38" @@ -273,7 +283,7 @@ def test_compute_week_avance_reduit_les_departs_du_soir_restants(freeze_today): result = compute_week(pointages, [], heures_jour, plages) assert result["extra_min"] == 0 # extra_min ne compte que le retard, pas l'avance for jour in result["jours"][1:]: - assert jour["cible_matin"] is None + assert jour["cible_matin"] == "13:00" # rythme normal, jamais ajusté par l'avance assert jour["cible_aprem"] == "17:55" @@ -301,6 +311,7 @@ def test_compute_week_avance_ne_descend_jamais_sous_aprem_fin(freeze_today): def test_compute_week_cible_matin_jamais_ajustee_meme_avec_de_la_marge(freeze_today): """Le rattrapage hebdomadaire ne porte jamais sur le matin, même quand la matinée obligatoire est plus longue que le nominal et aurait de la marge. + Le matin garde son horaire nominal (rythme normal), non ajusté. Matin 08:00→13:00 (capacité 300 > nominal 270). Vendredi en congé après-midi, 120 min de retard : le seul créneau ouvert (le matin) n'est pas ajustable, le @@ -323,14 +334,14 @@ def test_compute_week_cible_matin_jamais_ajustee_meme_avec_de_la_marge(freeze_to result = compute_week(pointages, conges, heures_jour, plages) vendredi = result["jours"][4] assert result["extra_min"] == 120 - assert vendredi["cible_matin"] is None + assert vendredi["cible_matin"] == "12:30" # rythme normal (08:00 + 270 nominal), pas de rattrapage assert vendredi["cible_aprem"] is None # congé def test_compute_week_arrivee_visee_n_etend_pas_le_rattrapage_au_matin(freeze_today): - """L'arrivée visée sert de début projeté du matin (pour le calcul du surplus - basculant vers l'après-midi), mais n'ouvre jamais le matin au rattrapage - hebdomadaire — celui-ci reste réservé aux départs du soir. + """L'arrivée visée sert de début projeté du matin (pour le rythme normal affiché + et le calcul du surplus basculant vers l'après-midi), mais n'ouvre jamais le + matin au rattrapage hebdomadaire — celui-ci reste réservé aux départs du soir. """ plages = {"arrivee_visee": "08:00"} conges = [{"date": "2026-07-17", "type": "aprem"}] @@ -344,7 +355,7 @@ def test_compute_week_arrivee_visee_n_etend_pas_le_rattrapage_au_matin(freeze_to result = compute_week(pointages, conges, HEURES_JOUR_MIN, plages) vendredi = result["jours"][4] assert result["extra_min"] == 120 - assert vendredi["cible_matin"] is None + assert vendredi["cible_matin"] == "11:54" # rythme normal (08:00 + 234 nominal), pas de rattrapage def test_compute_week_depart_vise_plafond_souple(freeze_today): @@ -372,7 +383,7 @@ def test_compute_week_depart_vise_plafond_souple(freeze_today): # vendredi aprem : 13:30 + 234 + (36 + 9) = 18:09 ; lundi aprem : 13:30 + 288 + 9 = 18:27 # (sans départ visé, lundi finirait à 18:45) assert vendredi["cible_aprem"] == "18:09" - assert lundi["cible_matin"] is None + assert lundi["cible_matin"] == "12:00" # matin au rythme normal (pas de rattrapage pour ce créneau) assert lundi["cible_aprem"] == "18:27" diff --git a/app/tests/test_logs.py b/app/tests/test_logs.py new file mode 100644 index 0000000..119a56e --- /dev/null +++ b/app/tests/test_logs.py @@ -0,0 +1,342 @@ +"""Smoke tests pour les nouvelles routes /logs et /aide + journalisation présence.""" +import sys + +# DATA_DIR temporaire est positionné par conftest.py avant l'import de models. +import models # noqa: F401 (réexport pratique pour les tests ci-dessous) + +sys.path.insert(0, ".") +import main +from fastapi.testclient import TestClient + +client = TestClient(main.app) + + +def _login_as(uid: str): + """Crée un user avec mdp puis se connecte (renvoie les cookies signés).""" + import bcrypt + models.save_auth(uid, email=f"{uid}@x.fr", + password_hash=bcrypt.hashpw(b"x" * 12, bcrypt.gensalt()).decode()) + r = client.post("/login/password", data={"email": f"{uid}@x.fr", "password": "x" * 12}, + follow_redirects=False) + assert r.status_code == 302, r.text + # Récupère le cookie user_id réellement signé par le serveur + return {"user_id": r.cookies.get("user_id")} + + +COOKIES = _login_as("alice") + + +# ── Route /logs ────────────────────────────────────────────────────────────── + +def test_logs_page_requires_auth(): + # Client frais, sans cookie, doit être redirigé vers /login + unauth = TestClient(main.app) + r = unauth.get("/logs", follow_redirects=False) + assert r.status_code == 302 + assert "/login" in r.headers["location"] + + +def test_logs_page_empty_for_new_user(): + r = client.get("/logs", cookies=COOKIES) + assert r.status_code == 200 + assert "Journaux" in r.text + assert "Aucun événement" in r.text # état initial + + +def test_logs_page_shows_entries_after_presence(): + token = models.load_notif_config("alice")["token"] + # Simule une arrivée Automation + client.post(f"/presence/{token}/arrivee") + # Simule un départ + client.post(f"/presence/{token}/depart") + + r = client.get("/logs", cookies=COOKIES) + assert r.status_code == 200 + assert "arrivée" in r.text + assert "départ" in r.text + assert "Présence" in r.text + assert "Aucun événement" not in r.text + + +def test_logs_filter_query_param(): + # Toutes les entrées sont de type "presence" à ce stade + r_all = client.get("/logs", cookies=COOKIES) + r_notif = client.get("/logs?filter=notif", cookies=COOKIES) + r_pres = client.get("/logs?filter=presence", cookies=COOKIES) + + # Le filtre "notif" ne doit PAS montrer les entrées presence + assert "arrivée" not in r_notif.text + # Le filtre "presence" DOIT les montrer + assert "arrivée" in r_pres.text + assert "départ" in r_pres.text + # Le total Tout inclut tout + assert "arrivée" in r_all.text + + +# ── Route /aide ────────────────────────────────────────────────────────────── + +def test_aide_page_requires_auth(): + unauth = TestClient(main.app) + r = unauth.get("/aide", follow_redirects=False) + assert r.status_code == 302 + assert "/login" in r.headers["location"] + + +def test_aide_page_content(): + r = client.get("/aide", cookies=COOKIES) + assert r.status_code == 200 + # Vérifie la présence des sections clés + for needle in ["ntfy", "Automation", "Aide", "topic", "géofence", "/logs"]: + assert needle in r.text, f"missing {needle!r}" + + +# ── Logging via notifications.send_ntfy (échec) ───────────────────────────── + +def test_send_ntfy_failure_logs_entry(): + # Serveur injoignable (host public sans résolution) → échec → entrée de log "échec" + import notifications + notifications.send_ntfy( + "https://ntfy-injoignable.invalid", "topic-injoignable", + "msg test", "titre test", user_id="alice", + ) + logs = models.load_logs("alice") + assert any(l["event"] == "échec" and l["type"] == "notif" for l in logs) + + +# ── Navigation ────────────────────────────────────────────────────────────── + +def test_nav_links_present(): + r = client.get("/aide", cookies=COOKIES) + assert 'href="/logs"' in r.text + assert 'href="/aide"' in r.text + assert 'href="/settings"' in r.text + + +# ── Isolation par user ────────────────────────────────────────────────────── + +def test_logs_isolated_per_user(): + # Bob ne doit pas voir les logs d'alice + bob_cookies = _login_as("bob") + r = client.get("/logs", cookies=bob_cookies) + assert "arrivée" not in r.text + assert "Aucun événement" in r.text + + +# ── Rotation des logs ─────────────────────────────────────────────────────── + +def test_log_rotation_caps_at_max(): + # Alice a déjà quelques entrées ; on en ajoute largement > MAX_LOGS + for i in range(models.MAX_LOGS + 50): + models.append_log("alice", "notif", "envoyée", f"msg {i}") + logs = models.load_logs("alice") + assert len(logs) == models.MAX_LOGS + # Le plus récent est bien le dernier inséré + assert logs[0]["message"] == f"msg {models.MAX_LOGS + 49}" + + +def test_invalid_log_type_rejected(): + try: + models.append_log("alice", "invalid_type", "x", "y") + assert False, "doit lever une AssertionError" + except AssertionError: + pass + + +# ── Bouton de test ntfy (/settings/test-ntfy) ──────────────────────────────── + +def test_test_ntfy_requires_auth(): + unauth = TestClient(main.app) + r = unauth.post("/settings/test-ntfy", follow_redirects=False) + assert r.status_code == 302 + assert "/login" in r.headers["location"] + + +def test_test_ntfy_without_topic_returns_error(): + # Alice n'a pas de topic configuré → message d'erreur + models.save_notif_config("alice", ntfy_topic="", ntfy_server="https://ntfy.sh") + r = client.post("/settings/test-ntfy", cookies=COOKIES) + assert r.status_code == 200 + assert "settings-error" in r.text + assert "topic" in r.text.lower() + + +def test_test_ntfy_with_bad_server_returns_failure(): + # Topic renseigné mais serveur injoignable → échec + models.save_notif_config("alice", ntfy_topic="topic-test", ntfy_server="https://ntfy-injoignable.invalid") + r = client.post("/settings/test-ntfy", cookies=COOKIES) + assert r.status_code == 200 + assert "settings-error" in r.text + assert "Échec" in r.text + # L'entrée de log d'échec doit être présente + logs = models.load_logs("alice") + assert any(l["event"] == "échec" and l["type"] == "notif" for l in logs) + + +def test_test_ntfy_logs_to_journal(): + """Le bouton de test doit écrire dans le journal (même en échec).""" + models.save_notif_config("alice", ntfy_topic="topic-test-2", ntfy_server="https://ntfy-injoignable.invalid") + client.post("/settings/test-ntfy", cookies=COOKIES) + logs = models.load_logs("alice") + # 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) + + +# ── 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" + + +# ── Régénération du token de présence (Tasker) ─────────────────────────────── + +def test_regenerate_presence_token_requires_auth(): + unauth = TestClient(main.app) + r = unauth.post("/settings/regenerate-presence-token", follow_redirects=False) + assert r.status_code == 302 + assert "/login" in r.headers["location"] + + +def test_regenerate_presence_token_invalidates_old_urls(): + """Le nouveau token remplace l'ancien : les anciennes URLs ne marchent plus.""" + old_token = models.load_notif_config("alice")["token"] + + r = client.post("/settings/regenerate-presence-token", cookies=COOKIES) + assert r.status_code == 200 + # Le HTML retourné contient de nouvelles URLs avec un nouveau token + assert old_token not in r.text + assert "/presence/" in r.text + assert "Token régénéré" in r.text + + # L'ancienne URL doit maintenant renvoyer 404 (token inconnu) + r_old = client.post(f"/presence/{old_token}/arrivee") + assert r_old.status_code == 404 + + # La nouvelle URL doit fonctionner + new_cfg = models.load_notif_config("alice") + r_new = client.post(f"/presence/{new_cfg['token']}/arrivee") + assert r_new.status_code == 200 + assert r_new.json() == {"ok": True} + + +def test_regenerate_preserves_other_notif_fields(): + """La régénération ne doit pas effacer topic/server/token ntfy.""" + models.save_notif_config( + "alice", ntfy_topic="topic-x", ntfy_server="https://ntfy.arkel.fr", + ntfy_token="tk_xyz", + ) + before = models.load_notif_config("alice") + new_token = models.regenerate_presence_token("alice") + after = models.load_notif_config("alice") + + assert after["token"] == new_token + assert before["token"] != new_token + # Les autres champs sont préservés + assert after["ntfy_topic"] == "topic-x" + assert after["ntfy_server"] == "https://ntfy.arkel.fr" + assert after["ntfy_token"] == "tk_xyz" + + +# ── Changements de paramètres tracés dans le journal ───────────────────────── + +def test_settings_save_logged(): + """Sauvegarder les réglages ntfy doit écrire une entrée de journal.""" + models.save_notif_config("alice", ntfy_topic="t", ntfy_server="https://ntfy.sh") + before = len([l for l in models.load_logs("alice") if l["type"] == "settings"]) + client.post("/settings", cookies=COOKIES, data={ + "ntfy_topic": "topic-new", + "ntfy_server": "https://ntfy.arkel.fr", + "ntfy_token": "tk_abc", + "rappel_debut_h": "8", + "rappel_fin_h": "19", + }) + after = [l for l in models.load_logs("alice") if l["type"] == "settings"] + assert len(after) == before + 1 + latest = after[0] + assert latest["event"] == "Réglages ntfy" + assert "topic-new" in latest["message"] + assert "ntfy.arkel.fr" in latest["message"] + + +def test_settings_plages_save_logged(): + """Sauvegarder les plages horaires doit écrire une entrée de journal.""" + before = len([l for l in models.load_logs("alice") if l["type"] == "settings"]) + client.post("/settings/plages", cookies=COOKIES, data={ + "matin_debut": "09:00", "matin_fin": "12:00", + "aprem_debut": "14:00", "aprem_fin": "17:00", + "pause_dejeuner_fin": "13:30", + "arrivee_visee": "", "depart_vise": "", + }) + after = [l for l in models.load_logs("alice") if l["type"] == "settings"] + assert len(after) == before + 1 + latest = after[0] + assert latest["event"] == "Plages horaires" + assert "09:00" in latest["message"] + + +def test_regenerate_presence_token_logged(): + """Régénérer le token de présence doit écrire une entrée de journal.""" + before = len([l for l in models.load_logs("alice") if l["type"] == "settings"]) + client.post("/settings/regenerate-presence-token", cookies=COOKIES) + after = [l for l in models.load_logs("alice") if l["type"] == "settings"] + assert len(after) == before + 1 + latest = after[0] + assert "régénéré" in latest["event"].lower() diff --git a/app/tests/test_security.py b/app/tests/test_security.py new file mode 100644 index 0000000..2cb1298 --- /dev/null +++ b/app/tests/test_security.py @@ -0,0 +1,599 @@ +"""Tests de sécurité : anti-forgery du cookie de session. + +Avant le correctif, n'importe qui pouvait forger le cookie `user_id=admin` et +accéder aux données d'un autre utilisateur. La signature HMAC doit rendre toute +altération détectable. +""" +import sys + +# DATA_DIR temporaire est positionné par conftest.py avant l'import de models. +import models # noqa: F401 + +sys.path.insert(0, ".") +import main +from fastapi.testclient import TestClient + +client = TestClient(main.app) + + +def _seed_user(uid: str): + import bcrypt + models.save_auth(uid, email=f"{uid}@x.fr", + password_hash=bcrypt.hashpw(b"x" * 12, bcrypt.gensalt()).decode()) + + +def _login(uid: str) -> dict: + _seed_user(uid) + r = client.post("/login/password", + data={"email": f"{uid}@x.fr", "password": "x" * 12}, + follow_redirects=False) + assert r.status_code == 302, r.text + return {"user_id": r.cookies.get("user_id")} + + +# ── Forgery du cookie ───────────────────────────────────────────────────────── + +def test_unsigned_cookie_rejected(): + """Un cookie `user_id=admin` forgé à la main doit être rejeté.""" + _seed_user("alice") + r = client.get("/semaine/2026/29", cookies={"user_id": "alice"}, follow_redirects=False) + assert r.status_code == 302 + assert "/login" in r.headers["location"] + + +def test_tampered_signature_rejected(): + """Un cookie valide dont on modifie la signature doit être rejeté.""" + cookies = _login("alice") + raw = cookies["user_id"] + user_id, _, sig = raw.rpartition(".") + # Inverse un caractère de la signature + bad_sig = ("0" if sig[0] != "0" else "1") + sig[1:] + tampered = f"{user_id}.{bad_sig}" + r = client.get("/semaine/2026/29", cookies={"user_id": tampered}, follow_redirects=False) + assert r.status_code == 302 + assert "/login" in r.headers["location"] + + +def test_valid_signed_cookie_accepted(): + """Le cookie signé délivré par /login/password doit ouvrir la session.""" + cookies = _login("alice") + r = client.get("/semaine/2026/29", cookies=cookies, follow_redirects=False) + assert r.status_code == 200 + + +def test_signed_cookie_cannot_be_replayed_for_other_user(): + """La signature d'alice ne doit pas valider pour bob (user_id modifié).""" + cookies = _login("alice") + raw = cookies["user_id"] + _, _, sig = raw.rpartition(".") + # Remplace alice par bob en conservant la signature d'alice + forged = f"bob.{sig}" + _seed_user("bob") + r = client.get("/semaine/2026/29", cookies={"user_id": forged}, follow_redirects=False) + assert r.status_code == 302 + assert "/login" in r.headers["location"] + + +# ── Helpers de signature ────────────────────────────────────────────────────── + +def test_sign_and_verify_roundtrip(): + sig = main._sign_user_id("alice") + assert main._verify_signed_cookie(sig) == "alice" + + +def test_verify_rejects_malformed_inputs(): + for bad in ("", "alice", "alice.", ".", "alice.deadbeef", "inva/lid.xxx"): + assert main._verify_signed_cookie(bad) is None + + +def test_secret_is_stable_across_calls(): + s1 = models.load_or_create_secret() + s2 = models.load_or_create_secret() + assert s1 == s2 + + +# ── Validation SSRF du serveur ntfy ────────────────────────────────────────── + +import notifications # noqa: E402 + + +def test_ntfy_server_validation_accepts_https(): + assert notifications.is_safe_ntfy_server("https://ntfy.sh") + assert notifications.is_safe_ntfy_server("https://ntfy.example.com:8080") + + +def test_ntfy_server_validation_rejects_loopback(): + for bad in ("http://127.0.0.1:1", "http://127.0.0.1", "http://localhost", + "https://localhost:8443", "https://[::1]"): + assert not notifications.is_safe_ntfy_server(bad), bad + + +def test_ntfy_server_validation_rejects_private_networks(): + for bad in ("https://10.0.0.1", "https://172.16.5.5", "https://192.168.1.1", + "https://169.254.169.254", "http://10.255.255.1:8080"): + assert not notifications.is_safe_ntfy_server(bad), bad + + +def test_ntfy_server_validation_rejects_bad_schemes(): + for bad in ("ftp://ntfy.sh", "file:///etc/passwd", "gopher://x", "javascript:alert(1)"): + assert not notifications.is_safe_ntfy_server(bad), bad + + +def test_ntfy_server_validation_rejects_credentials_in_url(): + assert not notifications.is_safe_ntfy_server("https://user:pass@ntfy.sh") + + +def test_ntfy_server_validation_rejects_empty_or_none(): + for bad in ("", None, " "): + assert not notifications.is_safe_ntfy_server(bad), bad + + +def test_send_ntfy_refuses_ssrf_target(): + """Un appel vers une IP privée ne doit jamais ouvrir de connexion réseau.""" + import urllib.request as u + called = {"n": 0} + orig = u.urlopen + try: + u.urlopen = lambda *a, **kw: called.__setitem__("n", called["n"] + 1) + ok = notifications.send_ntfy( + "http://169.254.169.254", "ssrf-topic", + "msg", "titre", user_id=None, token="", + ) + finally: + u.urlopen = orig + assert ok is False + assert called["n"] == 0 # urlopen n'a jamais été appelé + + +def test_settings_save_rejects_ssrf_ntfy_server(): + """POST /settings avec un serveur SSRF doit être rejeté.""" + cookies = _login("carol") + # Préserve un topic valide + models.save_notif_config("carol", ntfy_topic="t", ntfy_server="https://ntfy.sh") + r = client.post("/settings", cookies=cookies, data={ + "ntfy_topic": "topic-x", + "ntfy_server": "http://169.254.169.254", + "ntfy_token": "", + "rappel_debut_h": "8", + "rappel_fin_h": "19", + }, follow_redirects=False) + assert r.status_code == 303 + assert "ssrf_error=1" in r.headers["location"] + # La valeur précédente est conservée + cfg = models.load_notif_config("carol") + assert cfg["ntfy_server"] == "https://ntfy.sh" + + +# ── XSS stocké via /pointage ────────────────────────────────────────────────── + +XSS_PAYLOADS = [ + '">', + '">', + "'/>", +] + + +def test_pointage_rejects_non_hhmm_value(): + """POST /pointage avec une valeur non HHMM ne doit pas être persistée.""" + cookies = _login("dave") + for payload in XSS_PAYLOADS: + r = client.post("/pointage/2026-07-17", cookies=cookies, data={ + "matin_entree": payload, + "matin_sortie": "", + "aprem_entree": "", + "aprem_sortie": "", + }) + # La réponse ne doit pas planter + assert r.status_code == 200 + # Le payload ne doit pas se retrouver tel quel dans la réponse + assert payload not in r.text, payload + # Vérification de la persistance + week = models.load_week_pointages(2026, 29, "dave") + entry = next((p for p in week if p["date"] == "2026-07-17"), None) + assert entry is not None + # matin_entree doit avoir été ignoré (None), pas stocké tel quel + assert entry["matin_entree"] is None + + +def test_pointage_valid_hhmm_still_works(): + """Un pointage HHMM valide doit continuer à fonctionner.""" + cookies = _login("erin") + r = client.post("/pointage/2026-07-17", cookies=cookies, data={ + "matin_entree": "08:45", + "matin_sortie": "12:00", + "aprem_entree": "", + "aprem_sortie": "", + }) + assert r.status_code == 200 + assert "08:45" in r.text + week = models.load_week_pointages(2026, 29, "erin") + entry = next((p for p in week if p["date"] == "2026-07-17"), None) + assert entry["matin_entree"] == "08:45" + + +def test_oob_html_escapes_date_attribute(): + """Si la date (issue de l'URL) contient des caractères HTML, ils doivent être échappés.""" + cookies = _login("frank") + # Forçage d'un pointage valide pour déclencher le rendu _oob_time_cells + client.post("/pointage/2026-07-17", cookies=cookies, data={ + "matin_entree": "08:45", "matin_sortie": "", "aprem_entree": "", "aprem_sortie": "", + }) + # Une date malicieuse dans /refresh/... doit être échappée si elle passe strptime + # (les caractères HTML ne passent pas strptime, mais on vérifie que l'escape est actif + # sur le rendu d'un pointage normal pour la date légitime) + r = client.post("/pointage/2026-07-17", cookies=cookies, data={ + "matin_entree": "08:45", "matin_sortie": "12:00", + "aprem_entree": "14:00", "aprem_sortie": "17:30", + }) + assert ").""" + notif = models.load_notif_config("alice") if (models.DATA_DIR / "users" / "alice").exists() else None + # Crée alice si nécessaire et récupère un token valide + _seed_user("alice") + token = models.load_notif_config("alice")["token"] + + # GET doit échouer (405 Method Not Allowed) + r_get_arr = client.get(f"/presence/{token}/arrivee", follow_redirects=False) + r_get_dep = client.get(f"/presence/{token}/depart", follow_redirects=False) + assert r_get_arr.status_code == 405, r_get_arr.status_code + assert r_get_dep.status_code == 405 + + # POST doit fonctionner + r_post_arr = client.post(f"/presence/{token}/arrivee") + r_post_dep = client.post(f"/presence/{token}/depart") + assert r_post_arr.status_code == 200 + assert r_post_dep.status_code == 200 + assert r_post_arr.json() == {"ok": True} + + +# ── En-têtes de sécurité + cookie Secure ────────────────────────────────────── + +def test_security_headers_present_on_every_response(): + """Les en-têtes X-Content-Type-Options, X-Frame-Options, CSP et Referrer-Policy + doivent être positionnés sur toutes les réponses.""" + r = client.get("/login") + assert r.headers.get("X-Content-Type-Options") == "nosniff" + assert r.headers.get("X-Frame-Options") == "DENY" + assert r.headers.get("Referrer-Policy") == "same-origin" + csp = r.headers.get("Content-Security-Policy", "") + assert "default-src 'self'" in csp + assert "frame-ancestors 'none'" in csp + assert "base-uri 'self'" in csp + + +def test_hsts_header_only_on_https(): + """HSTS ne doit être positionné qu'en HTTPS.""" + # En HTTP (TestClient par défaut), pas de HSTS + r = client.get("/login") + assert "Strict-Transport-Security" not in r.headers + + # En HTTPS, HSTS doit être présent + https_client = TestClient(main.app, base_url="https://testserver") + r2 = https_client.get("/login") + assert r2.headers.get("Strict-Transport-Security", "").startswith("max-age=31536000") + + +def test_cookie_has_secure_flag_over_https(): + """Sur une requête HTTPS, le cookie de session doit porter le drapeau Secure.""" + https_client = TestClient(main.app, base_url="https://testserver") + _seed_user("kate") + r = https_client.post( + "/login/password", + data={"email": "kate@x.fr", "password": "x" * 12}, + follow_redirects=False, + ) + assert r.status_code == 302 + cookie = r.headers.get("set-cookie", "") + assert "Secure" in cookie, cookie + assert "HttpOnly" in cookie + assert "SameSite=lax" in cookie + + +def test_cookie_omits_secure_flag_over_http(): + """En HTTP (dev local), le cookie ne doit pas avoir Secure (sinon il ne serait pas posé).""" + _seed_user("liam") + r = client.post( + "/login/password", + data={"email": "liam@x.fr", "password": "x" * 12}, + follow_redirects=False, + ) + assert r.status_code == 302 + cookie = r.headers.get("set-cookie", "") + assert "HttpOnly" in cookie + assert "SameSite=lax" in cookie + # En HTTP, pas de Secure (sinon le navigateur refuserait le cookie) + assert "Secure" not in cookie + + +def test_force_secure_cookies_config_overrides_scheme(): + """config.force_secure_cookies doit forcer Secure même en HTTP.""" + cfg = models.load_config() + cfg["force_secure_cookies"] = True + models.save_config(cfg) + try: + _seed_user("mona") + r = client.post( + "/login/password", + data={"email": "mona@x.fr", "password": "x" * 12}, + follow_redirects=False, + ) + assert "Secure" in r.headers.get("set-cookie", "") + finally: + cfg["force_secure_cookies"] = False + models.save_config(cfg) + + +# ── Rate-limiting /login et /login/password ─────────────────────────────────── +# Empêche le brute-force du mot de passe et le mail-bombing SMTP. + +def test_login_password_rate_limit_kicks_in(): + """Au-delà de la limite, /login/password doit refuser.""" + main._LOGIN_ATTEMPTS.clear() + _seed_user("nina") + cfg = models.load_config() + cfg["login_rate_limit_per_min"] = 3 + models.save_config(cfg) + try: + # 3 tentatives (toutes incorrectes) → autorisées + for _ in range(3): + r = client.post( + "/login/password", + data={"email": "nina@x.fr", "password": "bad"}, + follow_redirects=False, + ) + assert r.status_code == 200 + assert "incorrect" in r.text.lower() or "Mot de passe" in r.text + # 4e tentative → bloquée + r4 = client.post( + "/login/password", + data={"email": "nina@x.fr", "password": "bad"}, + follow_redirects=False, + ) + assert r4.status_code == 200 + assert "Trop de tentatives" in r4.text + finally: + cfg["login_rate_limit_per_min"] = main.DEFAULT_LOGIN_MAX_PER_MIN + models.save_config(cfg) + main._LOGIN_ATTEMPTS.clear() + + +def test_login_post_rate_limit_kicks_in(): + """Au-delà de la limite, POST /login doit refuser (mail-bombing).""" + main._LOGIN_ATTEMPTS.clear() + cfg = models.load_config() + cfg["login_rate_limit_per_min"] = 2 + # Config SMTP valide pour que la route ne plante pas avant send_mail + cfg["smtp"] = {"host": "smtp.test", "port": 587, "user": "u", "password": "p", "use_tls": True} + models.save_config(cfg) + try: + # 2 POST /login OK (le mail peut échouer, mais le rate limit passe) + for i in range(2): + r = client.post("/login", data={"email": f"victim{i}.x@x.fr"}) + assert "Trop de tentatives" not in r.text + # 3e POST → bloqué + r3 = client.post("/login", data={"email": "victim3.x@x.fr"}) + assert "Trop de tentatives" in r3.text + finally: + cfg["login_rate_limit_per_min"] = main.DEFAULT_LOGIN_MAX_PER_MIN + models.save_config(cfg) + main._LOGIN_ATTEMPTS.clear() + + +def test_rate_limit_independent_per_email(): + """Le rate limit /login/password est par (ip, email) : deux emails distincts + ne se partagent pas leur quota.""" + main._LOGIN_ATTEMPTS.clear() + cfg = models.load_config() + cfg["login_rate_limit_per_min"] = 2 + models.save_config(cfg) + try: + _seed_user("oscar") + # 2 tentatives sur "oscar" → OK + for _ in range(2): + client.post("/login/password", data={"email": "oscar@x.fr", "password": "x"}, follow_redirects=False) + # La 3e sur "oscar" doit échouer + r_o = client.post("/login/password", data={"email": "oscar@x.fr", "password": "x"}, follow_redirects=False) + assert "Trop de tentatives" in r_o.text + # Une autre adresse email doit encore fonctionner + _seed_user("paul") + r_p = client.post("/login/password", data={"email": "paul@x.fr", "password": "x"}, follow_redirects=False) + assert "Trop de tentatives" not in r_p.text + finally: + cfg["login_rate_limit_per_min"] = main.DEFAULT_LOGIN_MAX_PER_MIN + models.save_config(cfg) + main._LOGIN_ATTEMPTS.clear() + + +def test_rate_limit_can_be_disabled_via_config(): + """config.login_rate_limit_per_min = 0 doit désactiver le rate limit.""" + main._LOGIN_ATTEMPTS.clear() + cfg = models.load_config() + cfg["login_rate_limit_per_min"] = 0 + models.save_config(cfg) + try: + _seed_user("quinn") + # Beaucoup de tentatives : aucune ne doit être bloquée + for _ in range(50): + r = client.post("/login/password", data={"email": "quinn@x.fr", "password": "x"}, follow_redirects=False) + assert "Trop de tentatives" not in r.text, r.text + finally: + cfg["login_rate_limit_per_min"] = main.DEFAULT_LOGIN_MAX_PER_MIN + models.save_config(cfg) + main._LOGIN_ATTEMPTS.clear() diff --git a/data/config.example.json b/data/config.example.json index a106702..faa566d 100644 --- a/data/config.example.json +++ b/data/config.example.json @@ -1,7 +1,9 @@ { - "heures_jour": 7.8, + "heures_semaine": 39.0, + "jours_par_semaine": 5, "log_level": "DEBUG", "allowed_email_domain": "exemple.fr", + "base_url": "", "mail_from": "no-reply@exemple.fr", "smtp": { "host": "smtp.exemple.fr", diff --git a/docs/notifications.md b/docs/notifications.md index ab085a2..5431618 100644 --- a/docs/notifications.md +++ b/docs/notifications.md @@ -25,23 +25,64 @@ contiennent un token secret, ne les partage pas. 2. Bouton **+** → entre le nom de topic choisi sur `/settings` → valide. Laisse le serveur par défaut (`ntfy.sh`). -### 2. Détecter la présence (Tasker) +### 2. Détecter la présence (Automation) -1. Installe **Tasker**. -2. Autorise Tasker à utiliser la localisation en arrière-plan (sinon Android - coupe la géolocalisation quand l'app n'est pas au premier plan). -3. Crée un profil : **Trigger → Location** → renseigne l'adresse/coordonnées - du bureau et un rayon (~100–150 m) → événement **Enter**. - Tâche associée : **Net → HTTP Request**, méthode `GET`, URL = ton lien - **Arrivée** (copié depuis `/settings`). -4. Crée un second profil symétrique : même géofence, événement **Exit**, - tâche HTTP GET vers ton lien **Départ**. +On utilise **[Automation](https://f-droid.org/en/packages/com.jens.automation2/)** +(app FOSS, sur F-Droid). Elle sait déclencher une requête HTTP POST selon ta +position géographique ou selon le Wi-Fi auquel tu es connecté — pas besoin de +plugin ni d'app payante. + +1. Installe **Automation** depuis + [F-Droid](https://f-droid.org/en/packages/com.jens.automation2/) + (évite la version Play Store, amputée de la localisation et des appels/SMS). +2. Autorise la **localisation en arrière-plan** : Réglages → Applications → + Automation → Autorisations → Position → **Toujours**. Sans ça, Android + coupe la détection quand l'app n'est pas au premier plan. +3. Désactive l'**optimisation batterie** pour Automation (même menu → + Batterie → « Aucune restriction »), sinon Android finit par tuer le service. + +Deux méthodes possibles — la **A** (Wi-Fi) est plus réactive et plus économe +en batterie au bureau. + +#### Méthode A — déclenchement par Wi-Fi (recommandé) + +Idéal si le bureau a un Wi-Fi d'entreprise. Le déclenchement se fait sur le +SSID, même si tu n'es pas identifié sur le réseau. + +**Règle « Arrivée »** — Automation → **Rules** → **+** : + +- *Trigger* : **Wifi connection** → entre le SSID exact du Wi-Fi du bureau. +- *Action* : **Trigger URL** → Method `POST`, + URL = ton lien **Arrivée** (copié depuis `/settings`). + +**Règle « Départ »**, symétrique : + +- *Trigger* : **Wifi connection** → même SSID, sens **déconnexion**. +- *Action* : **Trigger URL** `POST` → URL **Départ**. + +#### Méthode B — déclenchement par géofence (si pas de Wi-Fi d'entreprise) + +**Règle « Arrivée »** : + +- *Trigger* : **Location** → place le marqueur sur l'adresse du bureau, + rayon ~100–150 m, événement **Enter**. +- *Action* : **Trigger URL** `POST` → URL **Arrivée**. + +**Règle « Départ »**, symétrique : + +- *Trigger* : **Location** → même zone, événement **Exit**. +- *Action* : **Trigger URL** `POST` → URL **Départ**. + +> La géofence Android s'appuie sur le Wi-Fi/Bluetooth alentour même sans GPS, +> mais la réactivité varie. Au bureau, la méthode Wi-Fi est presque toujours +> plus fiable. ### 3. Vérifier -- Force une entrée/sortie de zone (ou lance la tâche HTTP manuellement dans - Tasker) et vérifie que la réponse est `{"ok":true}`. -- Simule une journée sans pointage matin pendant que tu es marqué "présent" : +- Dans Automation, ouvre une règle et utilise le bouton **Test** (icône + lecture) : la réponse doit être `{"ok":true}`. +- Force une entrée/sortie réelle (zone ou Wi-Fi) et vérifie la même chose. +- Simule une journée sans pointage matin pendant que tu es marqué « présent » : une notif ntfy doit arriver dans les 5 minutes. --- @@ -86,10 +127,10 @@ current_ssid=$(nmcli -t -f active,ssid dev wifi | grep '^yes' | cut -d: -f2) case "$2" in up) - [ "$current_ssid" = "$BUREAU_SSID" ] && curl -fsS "$URL_ARRIVEE" + [ "$current_ssid" = "$BUREAU_SSID" ] && curl -fsS -X POST "$URL_ARRIVEE" ;; down) - [ "$current_ssid" != "$BUREAU_SSID" ] && curl -fsS "$URL_DEPART" + [ "$current_ssid" != "$BUREAU_SSID" ] && curl -fsS -X POST "$URL_DEPART" ;; esac ```