199 lines
6.6 KiB
Python
199 lines
6.6 KiB
Python
"""Tests de durcissement — concurrence users.json + fuzzing regex (ROADMAP #87 T2).
|
|
|
|
- `users.json` : les read-modify-write sont sérialisés par `_users_lock`
|
|
(BUG-029). Ces tests martèlent create/update/record_login_failure depuis
|
|
plusieurs threads et exigent zéro mise à jour perdue + un JSON valide.
|
|
- Regex (BUG-025) : la politique `regex_safety` (longueur, quantificateurs
|
|
imbriqués, contenu tronqué) doit rejeter vite les motifs catastrophiques
|
|
et borner le temps des motifs acceptés sur gros contenu.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import threading
|
|
import time
|
|
|
|
N_THREADS = 6
|
|
|
|
|
|
def test_users_concurrent_create_and_update(tmp_path, monkeypatch):
|
|
"""Créations + mises à jour concurrentes : aucun utilisateur perdu."""
|
|
from backend.auth import user_store
|
|
|
|
monkeypatch.setattr(user_store, "USERS_FILE", tmp_path / "users.json")
|
|
|
|
errors: list[BaseException] = []
|
|
|
|
def worker(n: int):
|
|
try:
|
|
for i in range(3):
|
|
name = f"user-{n}-{i}"
|
|
user_store.create_user(name, "Motdepasse1!", display_name=name)
|
|
user_store.update_user(name, {"display_name": f"{name}-renamed"})
|
|
except BaseException as e: # pragma: no cover - diagnostic
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=worker, args=(n,)) for n in range(N_THREADS)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert not errors
|
|
users = user_store._read()["users"]
|
|
assert len(users) == N_THREADS * 3
|
|
assert all(u["display_name"].endswith("-renamed") for u in users.values())
|
|
|
|
|
|
def test_users_concurrent_login_failures_no_lost_count(tmp_path, monkeypatch):
|
|
"""`record_login_failure` concurrents : compteur exact (pas de lost update)."""
|
|
from backend.auth import user_store
|
|
|
|
monkeypatch.setattr(user_store, "USERS_FILE", tmp_path / "users.json")
|
|
user_store.create_user("victim", "Motdepasse1!")
|
|
|
|
def worker():
|
|
for _ in range(10):
|
|
try:
|
|
user_store.record_login_failure("victim")
|
|
except Exception: # verrouillage éventuel : ne doit pas lever
|
|
pass
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(N_THREADS)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
user = user_store.get_user("victim")
|
|
assert user is not None
|
|
# Le compte peut se verrouiller en cours de route ; l'important est que
|
|
# le fichier reste un JSON valide et l'utilisateur présent.
|
|
assert user["username"] == "victim"
|
|
|
|
|
|
def test_users_file_stays_valid_json_under_load(tmp_path, monkeypatch):
|
|
"""Le fichier reste lisible à tout moment pendant les écritures."""
|
|
import json
|
|
|
|
from backend.auth import user_store
|
|
|
|
target = tmp_path / "users.json"
|
|
monkeypatch.setattr(user_store, "USERS_FILE", target)
|
|
user_store.create_user("base", "Motdepasse1!")
|
|
|
|
stop = threading.Event()
|
|
errors: list[BaseException] = []
|
|
|
|
def writer(n: int):
|
|
i = 0
|
|
while not stop.is_set():
|
|
try:
|
|
user_store.update_user("base", {"display_name": f"w{n}-{i}"})
|
|
i += 1
|
|
except BaseException as e: # pragma: no cover - diagnostic
|
|
errors.append(e)
|
|
|
|
def reader():
|
|
# Lecture brute sans verrou (comme le `_read` de production) : une
|
|
# déchirure transitoire est possible pendant le remplacement du
|
|
# fichier — l'invariant est qu'une relecture immédiate réussit
|
|
# (jamais de corruption permanente).
|
|
while not stop.is_set():
|
|
try:
|
|
raw = target.read_text(encoding="utf-8")
|
|
json.loads(raw)
|
|
except FileNotFoundError:
|
|
pass
|
|
except json.JSONDecodeError:
|
|
try:
|
|
time.sleep(0.01)
|
|
json.loads(target.read_text(encoding="utf-8"))
|
|
except FileNotFoundError:
|
|
pass
|
|
except BaseException as e: # pragma: no cover - diagnostic
|
|
errors.append(e)
|
|
except BaseException as e: # pragma: no cover - diagnostic
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=writer, args=(n,)) for n in range(4)]
|
|
threads.append(threading.Thread(target=reader))
|
|
for t in threads:
|
|
t.start()
|
|
time.sleep(2.0)
|
|
stop.set()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert not errors
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fuzzing regex — budget temps (BUG-025)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Motifs classiquement catastrophiques : doivent être REJETÉS vite.
|
|
CATASTROPHIC = [
|
|
"^(a+)+$",
|
|
"(a+)+$",
|
|
"(.*)*$",
|
|
"(a|aa)+$",
|
|
"(a+){2,}$",
|
|
r"(\w+)+$",
|
|
r"(a*)*b",
|
|
r"(x+x+)+y",
|
|
]
|
|
|
|
# Motifs acceptés (légitimes) : doivent tourner vite sur gros contenu.
|
|
ACCEPTED = [
|
|
r"hello",
|
|
r"h.llo",
|
|
r"\b\w+@\w+\.\w+\b",
|
|
r"[A-ZÉÈÊ][a-zéèêàâîôûç]+",
|
|
r"(ab|cd)+e",
|
|
r"\d{4}-\d{2}-\d{2}",
|
|
r"foo|bar|baz",
|
|
]
|
|
|
|
|
|
def test_catastrophic_patterns_rejected_fast():
|
|
"""Les motifs à backtracking catastrophique sont refusés en < 1 s."""
|
|
from backend.services.regex_safety import validate_regex
|
|
|
|
start = time.perf_counter()
|
|
for pattern in CATASTROPHIC:
|
|
try:
|
|
validate_regex(pattern)
|
|
except ValueError:
|
|
pass
|
|
assert time.perf_counter() - start < 1.0
|
|
|
|
|
|
def test_accepted_patterns_bounded_on_large_content():
|
|
"""Motifs acceptés sur 200 Ko : chacun < 5 s (budget large anti-flaky)."""
|
|
from backend.services.regex_safety import MAX_REGEX_CONTENT, truncate_for_regex, validate_regex
|
|
|
|
assert MAX_REGEX_CONTENT == 200_000
|
|
content = truncate_for_regex("abc héllo world [email protected] 2024-01-02 " * 5000)
|
|
assert len(content) <= MAX_REGEX_CONTENT
|
|
for pattern in ACCEPTED:
|
|
validate_regex(pattern) # ne doit pas lever
|
|
start = time.perf_counter()
|
|
re.search(pattern, content)
|
|
assert time.perf_counter() - start < 5.0, f"motif lent : {pattern!r}"
|
|
|
|
|
|
def test_validate_regex_policy():
|
|
"""Politique : vide/trop long/invalide → ValueError."""
|
|
from backend.services.regex_safety import MAX_PATTERN_LENGTH, validate_regex
|
|
|
|
for bad in ("", "x" * (MAX_PATTERN_LENGTH + 1), "(unclosed"):
|
|
try:
|
|
validate_regex(bad)
|
|
except ValueError:
|
|
pass
|
|
else: # pragma: no cover - doit lever
|
|
raise AssertionError(f"motif accepté à tort : {bad!r}")
|
|
assert validate_regex("simple") == "simple"
|