diff --git a/app/main.py b/app/main.py index f0744b8..391aebd 100644 --- a/app/main.py +++ b/app/main.py @@ -6,6 +6,8 @@ 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 @@ -192,6 +194,40 @@ 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.). @@ -422,6 +458,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() @@ -477,7 +521,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) diff --git a/app/tests/test_security.py b/app/tests/test_security.py index 31f0c71..2cb1298 100644 --- a/app/tests/test_security.py +++ b/app/tests/test_security.py @@ -498,3 +498,102 @@ def test_force_secure_cookies_config_overrides_scheme(): 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()