Johan LEROY
ef933bea1a
Connexion, lecture du compte connecté, RBAC à trois rôles ordonnés et limitation de débit à fenêtre glissante. Ajoute `login_attempt`, le compteur de la limitation, et `audit_log`, en ajout seul. Trois ordres d'exécution portent la sécurité de ce commit, et chacun a son test : - les compteurs sont lus AVANT le hachage Argon2, sinon chaque requête rejetée coûterait quand même 17 ms et 19 Mio, et la protection serait l'amplificateur de déni de service qu'elle doit empêcher ; - un haché leurre est vérifié quand l'adresse est inconnue, sinon l'écart entre 2 ms et 17 ms est un oracle d'existence de compte ; - la tentative échouée est validée en base avant que l'erreur ne soit levée, `get_session()` ne validant pas de lui-même. Pas de verrouillage de compte : il suffirait de cinq requêtes pour mettre un administrateur dehors, et il ne fait rien contre le bourrage d'identifiants horizontal. Trois seuils le remplacent, dont un par couple (identifiant, IP) qui garantit qu'un attaquant ne peut pas empêcher la victime de se connecter depuis sa propre adresse. `audit_log` est en ajout seul au niveau de PostgreSQL, par deux déclencheurs. Le second n'est pas redondant : TRUNCATE ne passe pas par les déclencheurs de ligne. `test_route_protection.py` interroge réellement chaque route sans jeton. Rendre une route publique impose donc de modifier une liste dans un fichier de test, ce qui se voit en revue. Le gestionnaire de 422 arrive ici et non plus tard : la réponse par défaut de FastAPI contient la valeur rejetée, donc le mot de passe. Le test qui le prouve serait rouge sans lui.
119 lines
3.6 KiB
Python
119 lines
3.6 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.login_attempt import LoginOutcome
|
|
from app.repositories.login_attempt import LoginAttemptRepository
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
IP = "203.0.113.10"
|
|
AUTRE_IP = "198.51.100.7"
|
|
|
|
|
|
def adresse() -> str:
|
|
return f"tentative-{uuid.uuid4().hex[:12]}@enervision.fr"
|
|
|
|
|
|
async def echoue(
|
|
depot: LoginAttemptRepository, email: str, ip: str | None, combien: int = 1
|
|
) -> None:
|
|
for _ in range(combien):
|
|
await depot.record(email=email, client_ip=ip, outcome=LoginOutcome.IDENTIFIANTS_INVALIDES)
|
|
|
|
|
|
async def test_count_recent_failures_separates_the_three_counters(
|
|
session: AsyncSession,
|
|
) -> None:
|
|
depot = LoginAttemptRepository(session)
|
|
cible, voisin = adresse(), adresse()
|
|
await echoue(depot, cible, IP, combien=3)
|
|
await echoue(depot, cible, AUTRE_IP, combien=2)
|
|
await echoue(depot, voisin, IP, combien=4)
|
|
await session.flush()
|
|
|
|
compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
|
|
await session.rollback()
|
|
|
|
assert compteurs.per_identifier_and_ip == 3
|
|
assert compteurs.per_identifier == 5
|
|
assert compteurs.per_ip == 7
|
|
|
|
|
|
async def test_count_recent_failures_ignores_successful_attempts(
|
|
session: AsyncSession,
|
|
) -> None:
|
|
depot = LoginAttemptRepository(session)
|
|
cible = adresse()
|
|
await echoue(depot, cible, IP, combien=2)
|
|
await depot.record(email=cible, client_ip=IP, outcome=LoginOutcome.SUCCES)
|
|
await session.flush()
|
|
|
|
compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
|
|
await session.rollback()
|
|
|
|
assert compteurs.per_identifier_and_ip == 2
|
|
|
|
|
|
async def test_count_recent_failures_forgets_what_falls_outside_the_window(
|
|
session: AsyncSession,
|
|
) -> None:
|
|
depot = LoginAttemptRepository(session)
|
|
cible = adresse()
|
|
await echoue(depot, cible, IP, combien=2)
|
|
await session.flush()
|
|
await session.execute(
|
|
text(
|
|
"update login_attempt set occurred_at = now() - interval '2 hours' "
|
|
"where email_tried = :e"
|
|
),
|
|
{"e": cible},
|
|
)
|
|
|
|
compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
|
|
await session.rollback()
|
|
|
|
assert compteurs.per_identifier_and_ip == 0
|
|
|
|
|
|
async def test_count_recent_failures_still_counts_when_the_address_is_unknown(
|
|
session: AsyncSession,
|
|
) -> None:
|
|
depot = LoginAttemptRepository(session)
|
|
inconnu = adresse()
|
|
await echoue(depot, inconnu, IP, combien=5)
|
|
await session.flush()
|
|
|
|
compteurs = await depot.count_recent_failures(email=inconnu, client_ip=IP, window_seconds=900)
|
|
await session.rollback()
|
|
|
|
assert compteurs.per_identifier_and_ip == 5
|
|
|
|
|
|
async def test_record_normalises_the_address_before_counting(session: AsyncSession) -> None:
|
|
depot = LoginAttemptRepository(session)
|
|
cible = adresse()
|
|
await echoue(depot, cible.upper(), IP, combien=2)
|
|
await session.flush()
|
|
|
|
compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
|
|
await session.rollback()
|
|
|
|
assert compteurs.per_identifier_and_ip == 2
|
|
|
|
|
|
async def test_count_recent_failures_tolerates_a_missing_client_address(
|
|
session: AsyncSession,
|
|
) -> None:
|
|
depot = LoginAttemptRepository(session)
|
|
cible = adresse()
|
|
await echoue(depot, cible, None, combien=2)
|
|
await session.flush()
|
|
|
|
compteurs = await depot.count_recent_failures(email=cible, client_ip=None, window_seconds=900)
|
|
await session.rollback()
|
|
|
|
assert compteurs.per_identifier == 2
|