Ajoute un rate-limiting stateful sur /login et /login/password
Les routes d'authentification n'avaient aucune limite de débit :
- POST /login permettait le mail-bombing SMTP vers n'importe quelle adresse
du domaine autorisé (CSRF + flood).
- POST /login/password permettait le brute-force d'un mot de passe sans
verrou (bcrypt est lent mais pas assez pour un attaquant patient).
Ajout d'un compteur stateful en mémoire (_LOGIN_ATTEMPTS), fenêtre
glissante de 60 s, configurable via config.login_rate_limit_per_min
(défaut 10, <= 0 désactive). Clés :
- "login:<ip>" pour POST /login (limite globale par IP)
- "pw:<ip>:<email>" pour POST /login/password (limite par compte)
Au-delà du quota, la route renvoie la page login avec le message
"Trop de tentatives. Réessaie dans une minute." plutôt que d'exécuter
le workflow. La limite par (ip, email) rend le brute-force d'un compte
impossible sans rotation d'IP.
Tests : déclenchement aux limites, indépendance par email, désactivation
via config, et helper main._LOGIN_ATTEMPTS.clear() pour isoler les tests.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
This commit is contained in:
52
app/main.py
52
app/main.py
@ -6,6 +6,8 @@ import json
|
|||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
import secrets
|
import secrets
|
||||||
|
import time
|
||||||
|
from collections import defaultdict
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from itertools import groupby
|
from itertools import groupby
|
||||||
from urllib.parse import urlsplit
|
from urllib.parse import urlsplit
|
||||||
@ -192,6 +194,40 @@ def smtp_config() -> dict:
|
|||||||
return load_config().get("smtp", {})
|
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:
|
def _safe_base_url(request: Request) -> str | None:
|
||||||
"""Retourne une URL de base fiable pour construire des liens (emails, etc.).
|
"""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)
|
@app.post("/login", response_class=HTMLResponse)
|
||||||
def login_submit(request: Request, email: str = Form("")):
|
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()
|
email = email.strip().lower()
|
||||||
m = _EMAIL_RE.match(email)
|
m = _EMAIL_RE.match(email)
|
||||||
domain = allowed_email_domain()
|
domain = allowed_email_domain()
|
||||||
@ -477,7 +521,15 @@ def login_submit(request: Request, email: str = Form("")):
|
|||||||
|
|
||||||
@app.post("/login/password", response_class=HTMLResponse)
|
@app.post("/login/password", response_class=HTMLResponse)
|
||||||
def login_password(request: Request, email: str = Form(""), password: str = Form("")):
|
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()
|
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)
|
m = _EMAIL_RE.match(email)
|
||||||
domain = allowed_email_domain()
|
domain = allowed_email_domain()
|
||||||
logger.info("POST /login/password: email=%r", email)
|
logger.info("POST /login/password: email=%r", email)
|
||||||
|
|||||||
@ -498,3 +498,102 @@ def test_force_secure_cookies_config_overrides_scheme():
|
|||||||
finally:
|
finally:
|
||||||
cfg["force_secure_cookies"] = False
|
cfg["force_secure_cookies"] = False
|
||||||
models.save_config(cfg)
|
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()
|
||||||
|
|||||||
Reference in New Issue
Block a user