backend/tests/test_security.py

"""Suite de tests sécurité / vie privée — Allo Chat.

Lancer depuis backend/ :  ../.venv/bin/python -m pytest tests/ -v

Couvre les exigences issues des revues (interne + Gemini) avec les VRAIES routes :
énumération, signatures de photos, brute-force OTP, rate-limiting, purge EXIF,
absence de PII dans les réponses publiques, droit à l'oubli.
"""
from __future__ import annotations

import io
import os
import tempfile

# Environnement de test — AVANT l'import de l'app.
os.environ["CHATS_EMBEDDER"] = "baseline"
os.environ["CHATS_DETECT"] = "0"
os.environ["CHATS_VERIFY_ANIMAL"] = "0"   # images de test synthétiques (pas d'animal réel)
os.environ["CHATS_DEV_OTP"] = "1"
os.environ["CHATS_ADMIN_KEY"] = "test-admin-key"
os.environ.pop("CHATS_RESEND_KEY", None)
os.environ["DATABASE_URL"] = f"sqlite:///{tempfile.mkdtemp()}/test.db"

import pytest  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402
from PIL import Image  # noqa: E402

from app.main import app  # noqa: E402

ADMIN = {"X-Admin-Key": "test-admin-key"}

# --- Reproduit la contrainte VARCHAR(n) de Postgres dans les tests SQLite -------
# SQLite IGNORE les longueurs de colonnes ; Postgres (Neon) les applique et lève
# StringDataRightTruncation (→ 500). C'est ce qui a masqué le bug du téléphone
# chiffré (~150 car.) stocké dans une colonne VARCHAR(32). On installe un garde
# qui simule la contrainte en test → toute régression de troncation est attrapée
# en CI (même principe que le PRAGMA foreign_keys=ON pour les clés étrangères).
from sqlalchemy import String as _String, event as _event  # noqa: E402
from sqlalchemy.orm import Session as _SASession  # noqa: E402


def _enforce_varchar_lengths(session, flush_context, instances):  # noqa: ANN001
    for obj in list(session.new) + list(session.dirty):
        table = getattr(type(obj), "__table__", None)
        if table is None:
            continue
        for col in table.columns:
            length = getattr(col.type, "length", None)
            if isinstance(col.type, _String) and length:
                val = getattr(obj, col.key, None)
                if isinstance(val, str) and len(val) > length:
                    raise ValueError(
                        f"VARCHAR overflow simulé (Postgres) : {table.name}.{col.key} "
                        f"= {len(val)} caractères > {length}")


_event.listen(_SASession, "before_flush", _enforce_varchar_lengths)


def jpg(color=(200, 120, 40), exif: bool = False) -> bytes:
    img = Image.new("RGB", (220, 220), color)
    buf = io.BytesIO()
    if exif:
        ex = img.getexif()
        ex[0x010F] = "TestCam GPS leak"      # Make
        ex[0x0110] = "ModelX"                # Model
        img.save(buf, "JPEG", exif=ex.tobytes())
    else:
        img.save(buf, "JPEG")
    return buf.getvalue()


def ip(n: int) -> dict:
    """IP distincte par test → buckets de rate-limit isolés."""
    return {"X-Forwarded-For": f"10.0.0.{n}"}


@pytest.fixture(scope="module")
def client():
    with TestClient(app) as c:
        yield c


def sight(client, post_resp):
    """POST /sightings est ASYNCHRONE → récupère le résultat analysé via le sondage.
    En test, la tâche d'arrière-plan s'achève avant le retour du POST (TestClient
    attend tout le cycle ASGI), donc le 1er sondage renvoie déjà « ready »."""
    d = post_resp.json()
    tok = d.get("result_token")
    if not tok:
        return d   # pas async (erreur amont, ou réponse déjà complète)
    sid = d["sighting_id"]
    for _ in range(50):
        rr = client.get(f"/sightings/{sid}/result", params={"token": tok})
        if rr.status_code == 200 and rr.json().get("status") != "analyzing":
            return rr.json()
    return d


# --- Énumération / listing ---------------------------------------------------
def test_no_public_listing(client):
    assert client.get("/cats", headers=ip(1)).status_code in (403, 404, 405)
    assert client.get("/cats?limit=10&offset=0", headers=ip(1)).status_code in (403, 404, 405)
    r = client.get("/cats/", headers=ip(1), follow_redirects=True)
    assert r.status_code in (403, 404, 405)


def test_photo_enumeration_blocked(client):
    r = client.post("/cats", data={"name": "Enum", "email": "enum@t.ca"},
                    files=[("photos", ("a.jpg", jpg(), "image/jpeg"))], headers=ip(2))
    assert r.status_code == 200
    ov = client.get("/admin/api/overview", headers={**ADMIN, **ip(2)}).json()
    url = ov["cats"][0]["photos"][0]
    assert "?s=" in url, "les URLs de photos doivent être signées"
    assert client.get(url, headers=ip(2)).status_code == 200
    base = url.split("?")[0]
    assert client.get(base, headers=ip(2)).status_code == 403
    assert client.get(base + "?s=deadbeef00000000", headers=ip(2)).status_code == 403


def test_admin_requires_key(client):
    assert client.get("/admin/api/overview", headers=ip(3)).status_code == 403
    assert client.get("/admin/api/overview",
                      headers={"X-Admin-Key": "faux", **ip(3)}).status_code == 403


def test_admin_email_otp_flow(client):
    """L'accès admin exige un code (envoyé à l'email admin), pas une clé tapée."""
    rc = client.post("/admin/request-code", headers=ip(12)).json()
    assert rc["dev_code"], "code admin attendu en mode dev"
    assert client.post("/admin/verify-code", data={"code": "000000"},
                       headers=ip(12)).status_code == 400
    # H4 — anti email-bombing : re-demander un code dans les 45 s ne réenvoie rien
    # (message neutre, dev_code=None) et N'INVALIDE PAS le code déjà émis. Un essai
    # raté (attempts < max) ne brûle pas l'OTP → le premier code reste valide.
    again = client.post("/admin/request-code", headers=ip(12)).json()
    assert again["dev_code"] is None, "aucun nouveau code pendant le cooldown"
    v = client.post("/admin/verify-code", data={"code": rc["dev_code"]}, headers=ip(12))
    assert v.status_code == 200
    # C2 : un TOKEN DE SESSION opaque est remis, jamais la clé admin brute.
    key = v.json()["session_token"]
    assert client.get("/admin/api/overview",
                      headers={"X-Admin-Key": key, **ip(12)}).status_code == 200
    # code à usage unique
    assert client.post("/admin/verify-code", data={"code": rc["dev_code"]},
                       headers=ip(12)).status_code in (400, 429)


# --- OTP : brute force + rate limiting ---------------------------------------
def test_otp_brute_force_locked(client):
    email = "brute@t.ca"
    client.post("/cats", data={"name": "B", "email": email}, headers=ip(4))
    rc = client.post("/cats/request-code", data={"email": email}, headers=ip(4)).json()
    assert rc["dev_code"]
    codes = []
    for _ in range(6):
        r = client.post("/cats/verify-code",
                        data={"email": email, "code": "000000"}, headers=ip(4))
        codes.append(r.status_code)
    assert all(c in (400, 429) for c in codes)
    assert 429 in codes, "verrouillage attendu après tentatives répétées"
    # même le BON code est mort après le verrouillage (OTP détruit)
    r = client.post("/cats/verify-code",
                    data={"email": email, "code": rc["dev_code"]}, headers=ip(4))
    assert r.status_code in (400, 429)


def test_request_code_rate_limited(client):
    for i in range(5):
        assert client.post("/cats/request-code", data={"email": "rl@t.ca"},
                           headers=ip(5)).status_code == 200
    assert client.post("/cats/request-code", data={"email": "rl@t.ca"},
                       headers=ip(5)).status_code == 429


def test_otp_single_use(client):
    email = "single@t.ca"
    client.post("/cats", data={"name": "S", "email": email}, headers=ip(6))
    code = client.post("/cats/request-code", data={"email": email},
                       headers=ip(6)).json()["dev_code"]
    assert client.post("/cats/verify-code", data={"email": email, "code": code},
                       headers=ip(6)).status_code == 200
    assert client.post("/cats/verify-code", data={"email": email, "code": code},
                       headers=ip(6)).status_code in (400, 429)


# --- Vie privée ---------------------------------------------------------------
def test_exif_stripped_on_upload(client):
    """Les métadonnées EXIF (GPS, appareil…) doivent être purgées au stockage."""
    raw = jpg(exif=True)
    assert Image.open(io.BytesIO(raw)).getexif(), "le fichier de test doit avoir un EXIF"
    client.post("/cats", data={"name": "Exif", "email": "exif@t.ca"},
                files=[("photos", ("e.jpg", raw, "image/jpeg"))], headers=ip(7))
    ov = client.get("/admin/api/overview", headers={**ADMIN, **ip(7)}).json()
    cat = next(c for c in ov["cats"] if c["name"] == "Exif")
    img = client.get(cat["photos"][0], headers=ip(7))
    stored = Image.open(io.BytesIO(img.content))
    assert not dict(stored.getexif()), "EXIF non purgé sur l'image stockée"


def test_no_low_confidence_candidates(client):
    """CHOIX PRODUIT : on n'affiche JAMAIS un candidat sous 50 % de ressemblance.
    En dessous, la « correspondance » est trop incertaine — elle donne de faux
    espoirs et décrédibilise l'IA. Mieux vaut « aucune correspondance »."""
    from app import config

    assert config.CANDIDATE_THRESHOLD >= 0.50, (
        "le seuil d'affichage des candidats ne doit pas redescendre sous 50 %")

    # Deux chats de couleurs opposées : le signalement ne doit rien proposer
    # plutôt que de sortir un candidat faible.
    client.post("/cats", data={"name": "LowConf", "email": "low@t.ca"},
                files=[("photos", ("l.jpg", jpg((255, 0, 0)), "image/jpeg"))],
                headers=ip(40))
    sd = sight(client, client.post(
        "/sightings", files=[("photos", ("m.jpg", jpg((0, 0, 255)), "image/jpeg"))],
        headers=ip(40)))
    for m in sd.get("matches", []):
        assert float(m["confidence"]) >= 0.50, (
            f"candidat sous 50 % renvoyé au trouveur : {m['confidence']}")


def test_finder_response_has_no_pii(client):
    client.post("/cats", data={"name": "Pii", "email": "secret@t.ca",
                               "phone": "514-000-0000", "last_seen_lat": 45.5,
                               "last_seen_lng": -73.5},
                files=[("photos", ("p.jpg", jpg((10, 200, 80)), "image/jpeg"))],
                headers=ip(8))
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("q.jpg", jpg((12, 198, 82)), "image/jpeg"))],
                     headers=ip(8)))
    assert sd["matches"], "un candidat est attendu (même couleur, baseline)"
    m = next(m for m in sd["matches"] if m["name"] == "Pii")
    leaked = {"email", "phone", "contact_info", "last_seen_lat", "last_seen_lng",
              "claim_token"} & set(m.keys())
    assert not leaked, f"PII exposée au trouveur : {leaked}"
    # Audit MIT #10 — la vue trouveur ne doit pas fuiter les attributs internes
    # de détection (boîte, méthode) des photos du propriétaire ; seul un résumé
    # lisible du pelage est toléré.
    for ph in m.get("photos", []):
        attrs = ph.get("attributes") or {}
        assert "detection" not in attrs and "box" not in attrs, \
            f"attributs de détection fuités au trouveur : {attrs}"
        assert set(attrs.keys()) <= {"summary"}, \
            f"attributs inattendus exposés au trouveur : {attrs}"


def test_candidate_description_redacts_contact(client):
    """La description montrée au trouveur ne doit pas exposer un téléphone/courriel
    glissé en texte libre par le propriétaire (promesse de confidentialité)."""
    desc = "Tortie timide. Contactez-moi au 514-555-1234 ou jean@exemple.com"
    client.post("/cats", data={"name": "Redact", "email": "redact@t.ca",
                               "description": desc},
                files=[("photos", ("r.jpg", jpg((40, 160, 210)), "image/jpeg"))],
                headers=ip(20))
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("s.jpg", jpg((42, 158, 208)), "image/jpeg"))],
                     headers=ip(20)))
    m = next((m for m in sd.get("matches", []) if m["name"] == "Redact"), None)
    assert m, "un candidat est attendu (même couleur, baseline)"
    d = m.get("description") or ""
    assert "514-555-1234" not in d and "jean@exemple.com" not in d, \
        f"contact non masqué dans la description : {d!r}"
    assert "[masqué]" in d, "le masquage doit laisser une trace explicite"


def test_candidate_color_pattern_redacts_contact(client):
    """Audit MIT #11 — color/pattern montrés au trouveur ne doivent pas exposer
    un contact glissé dans ces champs libres par le propriétaire."""
    client.post("/cats", data={"name": "ColorPii", "email": "cp@t.ca",
                               "color": "noir, joignable au 514-555-9876",
                               "pattern": "écrivez à cp@exemple.com"},
                files=[("photos", ("c.jpg", jpg((33, 99, 180)), "image/jpeg"))],
                headers=ip(40))
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("d.jpg", jpg((35, 97, 178)), "image/jpeg"))],
                     headers=ip(40)))
    m = next((m for m in sd.get("matches", []) if m["name"] == "ColorPii"), None)
    assert m, "un candidat est attendu (même couleur, baseline)"
    blob = f"{m.get('color') or ''} {m.get('pattern') or ''}"
    assert "514-555-9876" not in blob and "cp@exemple.com" not in blob, \
        f"contact non masqué dans color/pattern : {blob!r}"
    assert "[masqué]" in blob, "le masquage doit laisser une trace explicite"


def test_match_radius_is_bounded(client):
    """Audit MIT #4 — un radius_km démesuré ne doit pas neutraliser le filtre géo
    (oracle visuel sur toute la base). Deux chats éloignés de ~900 km ne doivent
    pas matcher même avec un rayon de 100000 km (clampé à MAX_MATCH_RADIUS_KM)."""
    # Chat perdu à Montréal.
    client.post("/cats", data={"name": "FarCat", "email": "far@t.ca",
                               "last_seen_lat": 45.50, "last_seen_lng": -73.56},
                files=[("photos", ("f.jpg", jpg((70, 130, 190)), "image/jpeg"))],
                headers=ip(41))
    # Signalement à Toronto (~500 km) avec un rayon absurde.
    sd = sight(client, client.post("/sightings",
                     data={"lat": 43.65, "lng": -79.38, "radius_km": 100000},
                     files=[("photos", ("g.jpg", jpg((72, 128, 188)), "image/jpeg"))],
                     headers=ip(41)))
    names = [m["name"] for m in sd.get("matches", [])]
    assert "FarCat" not in names, \
        "le rayon démesuré aurait dû être clampé → chat hors zone écarté"


def test_sighting_async_result_token_required(client):
    """L'analyse d'un signalement est ASYNCHRONE : le POST répond tout de suite
    (status=analyzing + result_token) ; le sondage du résultat EXIGE le bon token
    (empêche un tiers de lire les correspondances d'autrui par énumération d'id)."""
    r = client.post("/sightings",
                    files=[("photos", ("a.jpg", jpg((100, 50, 25)), "image/jpeg"))],
                    headers=ip(58))
    d = r.json()
    assert d["status"] == "analyzing" and d["result_token"], "le POST doit répondre sans attendre l'IA"
    sid = d["sighting_id"]
    assert client.get(f"/sightings/{sid}/result",
                      params={"token": "z" * 24}).status_code == 403, "mauvais token → 403"
    rr = client.get(f"/sightings/{sid}/result", params={"token": d["result_token"]})
    assert rr.status_code == 200 and rr.json()["status"] == "ready", "bon token → résultat prêt"


def test_security_headers_present(client):
    r = client.get("/health", headers=ip(21))
    assert r.headers.get("X-Content-Type-Options") == "nosniff"
    assert r.headers.get("X-Frame-Options") == "DENY"
    assert r.headers.get("Referrer-Policy") == "no-referrer"


def test_admin_dashboard_escapes_html(client):
    """La page admin ne doit pas interpoler de données brutes : elle doit
    contenir le helper d'échappement (anti-XSS stocké)."""
    html = client.get("/admin", headers=ip(22)).text
    assert "const esc=" in html, "helper d'échappement HTML attendu dans la page admin"


def test_bad_file_is_not_500(client):
    r = client.post("/sightings",
                    files=[("photos", ("x.jpg", b"pas une image", "image/jpeg"))],
                    headers=ip(9))
    assert r.status_code < 500


def test_training_requires_consent(client):
    r = client.post("/training/photos",
                    files=[("photos", ("t.jpg", jpg((90, 90, 90)), "image/jpeg"))],
                    headers=ip(10))
    assert r.status_code == 400
    r = client.post("/training/photos", data={"consent": "1"},
                    files=[("photos", ("t.jpg", jpg((90, 90, 90)), "image/jpeg"))],
                    headers=ip(10))
    assert r.status_code == 200


# --- Anti-abus : leads, uploads ----------------------------------------------
def test_lead_only_on_real_candidate(client):
    """On ne peut pas déposer un lead sur un cat_id qui n'est PAS candidat du signalement."""
    # un chat existe (id réel), mais on crée un signalement avec une couleur opposée
    client.post("/cats", data={"name": "Far", "email": "far@t.ca"},
                files=[("photos", ("f.jpg", jpg((255, 0, 0)), "image/jpeg"))], headers=ip(12))
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("d.jpg", jpg((0, 0, 255)), "image/jpeg"))],
                     headers=ip(12)))
    # C4 : un lead cible un token signé sighting×cat, pas un cat_id brut. Un token
    # bidon (ou pour un chat non-candidat) doit être refusé.
    r = client.post(f"/sightings/{sd['sighting_id']}/lead",
                    json={"candidate_token": "bogus-token", "finder_contact": "x"},
                    headers=ip(12))
    assert r.status_code == 403, "lead sur token non-candidat doit être refusé"


def test_lead_dedup_no_double_notify(client):
    """Audit MIT #7/#9 — un lead identique rejoué (double-clic, token déterministe)
    ne crée pas un second signalement ni une seconde notification au propriétaire."""
    client.post("/cats", data={"name": "Dedup", "email": "dedup@t.ca"},
                files=[("photos", ("a.jpg", jpg((20, 200, 90)), "image/jpeg"))],
                headers=ip(42))
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("b.jpg", jpg((22, 198, 92)), "image/jpeg"))],
                     headers=ip(42)))
    m = next((m for m in sd.get("matches", []) if m["name"] == "Dedup"), None)
    assert m, "un candidat est attendu (même couleur, baseline)"
    sid = sd["sighting_id"]
    payload = {"candidate_token": m["candidate_token"],
               "finder_contact": "trouveur@x.ca", "message": "vu ici"}
    r1 = client.post(f"/sightings/{sid}/lead", json=payload, headers=ip(42))
    r2 = client.post(f"/sightings/{sid}/lead", json=payload, headers=ip(42))
    assert r1.status_code == 200 and r2.status_code == 200
    assert r1.json()["ref"] == r2.json()["ref"], \
        "le lead rejoué doit être idempotent (même réf, pas de doublon)"


def test_lead_message_length_bounded(client):
    """Audit MIT #7 — le message du trouveur (transmis par courriel au proprio)
    est borné en longueur (anti-charge utile / flood)."""
    client.post("/cats", data={"name": "LenCat", "email": "len@t.ca"},
                files=[("photos", ("a.jpg", jpg((60, 60, 200)), "image/jpeg"))],
                headers=ip(43))
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("b.jpg", jpg((62, 62, 198)), "image/jpeg"))],
                     headers=ip(43)))
    m = next((m for m in sd.get("matches", []) if m["name"] == "LenCat"), None)
    assert m, "un candidat est attendu (même couleur, baseline)"
    r = client.post(f"/sightings/{sd['sighting_id']}/lead",
                    json={"candidate_token": m["candidate_token"],
                          "message": "x" * 5000}, headers=ip(43))
    assert r.status_code == 422, "un message de 5000 caractères doit être rejeté (max 1000)"


# --- Régression : upload multi-images + colonnes Postgres (bug 500 « enc1: » phone) ---
def test_phone_column_is_unbounded_text(client):
    """Verrou structurel : le téléphone est CHIFFRÉ (~150 car.) → la colonne doit
    être Text/illimitée. VARCHAR(32) faisait planter l'INSERT sur Postgres (500)."""
    from app.models import Cat
    length = getattr(Cat.__table__.c.phone.type, "length", None)
    assert not length or length >= 200, \
        f"cats.phone doit être Text/illimité (valeur chiffrée ~150 car.) — length={length}"


def test_declare_cat_with_phone_and_multiple_photos(client):
    """LE scénario du bug : déclaration AVEC téléphone + plusieurs photos. Le
    téléphone chiffré dépassait VARCHAR(32) → 500. Avec la colonne Text + le garde
    de longueur ci-dessus, l'insert doit réussir et toutes les photos être prises."""
    files = [("photos", (f"p{i}.jpg", jpg((40 + i * 10, 120, 200 - i * 10)), "image/jpeg"))
             for i in range(5)]
    r = client.post("/cats",
                    data={"name": "Multi", "email": "multi@t.ca", "phone": "514-555-0188"},
                    files=files, headers=ip(50))
    assert r.status_code == 200, r.text
    assert r.json()["photos_count"] == 5


def test_sighting_with_multiple_photos(client):
    files = [("photos", (f"s{i}.jpg", jpg((10, 150 + i * 8, 60)), "image/jpeg"))
             for i in range(5)]
    r = client.post("/sightings", files=files, headers=ip(51))
    assert r.status_code == 200, r.text


def test_poster_with_multiple_photos(client):
    files = [("photos", (f"a{i}.jpg", jpg((200 - i * 5, 90, 120)), "image/jpeg"))
             for i in range(5)]
    r = client.post("/cats", data={"name": "Affiche", "source": "poster"},
                    files=files, headers=ip(52))
    assert r.status_code == 200, r.text
    assert r.json()["photos_count"] == 5


def test_long_freetext_inputs_do_not_500(client):
    """Une saisie libre très longue (name/color/pattern/breed) ne doit pas faire
    500 sur Postgres : create_cat tronque à la taille de colonne (_clip)."""
    r = client.post("/cats",
                    data={"name": "N" * 400, "email": "long@t.ca",
                          "color": "C" * 500, "pattern": "P" * 500, "breed": "B" * 500},
                    files=[("photos", ("x.jpg", jpg((9, 9, 9)), "image/jpeg"))],
                    headers=ip(53))
    assert r.status_code == 200, r.text


def test_sighting_long_reporter_email_clamped(client):
    """Audit : reporter_email (Form libre) → reporter_contact VARCHAR(255). Une
    saisie >255 car. doit être tronquée (clamp), pas faire 500 sur Postgres."""
    r = client.post("/sightings", data={"reporter_email": "a" * 400},
                    files=[("photos", ("s.jpg", jpg((11, 12, 13)), "image/jpeg"))],
                    headers=ip(56))
    assert r.status_code == 200, r.text


def test_training_long_label_group_clamped(client):
    """Audit : label_group (Form libre) → VARCHAR(120). Saisie >120 car. tronquée."""
    r = client.post("/training/photos",
                    data={"consent": "1", "label_group": "g" * 400},
                    files=[("photos", ("t.jpg", jpg((90, 90, 90)), "image/jpeg"))],
                    headers=ip(57))
    assert r.status_code == 200, r.text


def test_upload_skips_invalid_photo_keeps_valid(client):
    """Audit multi-images : une photo illisible parmi d'autres est SAUTÉE (pas de
    500, pas d'abandon de tout l'envoi) ; la fiche garde les photos valides."""
    files = [
        ("photos", ("ok.jpg", jpg((10, 20, 30)), "image/jpeg")),
        ("photos", ("bad.jpg", b"pas une image", "image/jpeg")),
        ("photos", ("ok2.jpg", jpg((30, 20, 10)), "image/jpeg")),
    ]
    r = client.post("/cats", data={"name": "Tol", "email": "tol@t.ca"},
                    files=files, headers=ip(54))
    assert r.status_code == 200, r.text
    assert r.json()["photos_count"] == 2, "la photo illisible doit être sautée, les 2 valides gardées"


def test_upload_all_invalid_photos_rejected(client):
    """Audit multi-images : si AUCUNE photo n'est exploitable → 422 explicite
    (jamais une fiche vide ni une 500)."""
    r = client.post("/cats", data={"name": "Bad", "email": "bad@t.ca"},
                    files=[("photos", ("b.jpg", b"nope pas une image", "image/jpeg"))],
                    headers=ip(55))
    assert r.status_code == 422, r.text


def test_too_many_photos_rejected(client):
    files = [("photos", (f"{i}.jpg", jpg(), "image/jpeg")) for i in range(20)]
    r = client.post("/cats", data={"name": "Many", "email": "many@t.ca"},
                    files=files, headers=ip(13))
    assert r.status_code == 413


def test_oversized_upload_rejected(client):
    big = b"\xff\xd8\xff" + b"0" * (16 * 1024 * 1024)  # > 15 Mo
    r = client.post("/cats", data={"name": "Big", "email": "big@t.ca"},
                    files=[("photos", ("b.jpg", big, "image/jpeg"))], headers=ip(14))
    assert r.status_code == 413


def test_email_cooldown(client):
    """Deux demandes de code rapprochées pour le même email → pas de 2e envoi."""
    r1 = client.post("/cats/request-code", data={"email": "cool@t.ca"}, headers=ip(15)).json()
    r2 = client.post("/cats/request-code", data={"email": "cool@t.ca"}, headers=ip(16)).json()
    assert r1.get("dev_code") and not r2.get("dev_code"), "2e demande rapide ne doit pas renvoyer un nouveau code"


# --- Droit à l'oubli ----------------------------------------------------------
def test_right_to_be_forgotten(client):
    r = client.post("/cats", data={"name": "Oubli", "email": "oubli@t.ca"},
                    files=[("photos", ("o.jpg", jpg((250, 30, 30)), "image/jpeg"))],
                    headers=ip(11))
    token = r.json()["claim_token"]
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("o2.jpg", jpg((248, 32, 33)), "image/jpeg"))],
                     headers=ip(11)))
    target = next((m for m in sd.get("matches", []) if m["name"] == "Oubli"), None)
    if target:
        client.post(f"/sightings/{sd['sighting_id']}/lead",
                    json={"candidate_token": target["candidate_token"], "finder_contact": "x"},
                    headers=ip(11))
        photo_url = target["photos"][0]["url"]
    r = client.post("/cats/track/delete", data={"token": token}, headers=ip(11))
    assert r.status_code == 200 and r.json()["deleted"]
    # Audit #3 : le claim_token passe par l'en-tête X-Track-Token (plus en query).
    assert client.get("/cats/track",
                      headers={"X-Track-Token": token, **ip(11)}).status_code == 404
    if target:
        assert client.get(photo_url, headers=ip(11)).status_code == 404, \
            "les photos doivent disparaître avec la fiche"


# --- Correctifs de l'audit confidentialité/sécurité (2026-06) ----------------
def test_training_oversized_upload_rejected(client):
    """Audit #1 : /training/photos doit borner la taille comme les autres uploads."""
    big = b"\xff\xd8\xff" + b"0" * (16 * 1024 * 1024)  # > 15 Mo
    r = client.post("/training/photos",
                    data={"consent": "1"},
                    files=[("photos", ("b.jpg", big, "image/jpeg"))], headers=ip(40))
    assert r.status_code == 413


def test_training_data_retention_purges_old_photos(client):
    """Loi 25 : les photos d'entraînement trop vieilles sont purgées, et les
    TrainingLabel qui les référencent sont supprimés D'ABORD (FK NO ACTION →
    sinon violation FK sur Postgres ; PRAGMA foreign_keys=ON le simule en test)."""
    import datetime as _dt
    from app.database import SessionLocal
    from app import retention, models
    db = SessionLocal()
    try:
        old = _dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(days=900)
        pa = models.TrainingPhoto(filename="ra.jpg", content_type="image/jpeg",
                                  embedding="[0.1]", created_at=old)
        pb = models.TrainingPhoto(filename="rb.jpg", content_type="image/jpeg",
                                  embedding="[0.2]", created_at=old)
        db.add_all([pa, pb]); db.flush()
        db.add(models.TrainingLabel(photo_a_id=pa.id, photo_b_id=pb.id, verdict="same"))
        db.commit()
        pa_id, pb_id = pa.id, pb.id
        orig = retention.TRAINING_DAYS
        retention.TRAINING_DAYS = 730            # 900 j > 730 → purge (et non désactivé)
        try:
            counts = retention.purge(db)         # ne doit PAS lever (labels supprimés d'abord)
        finally:
            retention.TRAINING_DAYS = orig
        assert counts.get("training_photos", 0) >= 2
        assert db.get(models.TrainingPhoto, pa_id) is None
        assert db.get(models.TrainingPhoto, pb_id) is None
        assert db.query(models.TrainingLabel).count() == 0, "le label référençant doit avoir été purgé"
    finally:
        db.close()


def test_feedback_requires_admin(client):
    """Audit #6 : /feedback (oracle d'énumération) est réservé à l'admin."""
    r = client.post("/feedback",
                    json={"photo_a_id": 1, "photo_b_id": 2, "verdict": "same"},
                    headers=ip(41))
    assert r.status_code == 403


def test_event_lookup_requires_admin(client):
    """Audit : la recherche par référence ne doit jamais être publique."""
    r = client.get("/admin/api/event/VU-000001", headers=ip(42))
    assert r.status_code == 403


def test_health_does_not_leak_db_host(client):
    """Audit #21 : /health n'expose pas l'hôte Neon ni le backend ML."""
    d = client.get("/health", headers=ip(43)).json()
    assert d["status"] == "ok"
    assert "@" not in d.get("db", ""), "l'hôte de la base ne doit pas fuiter"
    assert "embedder" not in d


def test_description_redacts_url_and_address(client):
    """Audit #14 : URL et adresse civique masquées dans la description publique."""
    desc = "Vu au 123 rue Saint-Denis. Infos sur http://mon-site.example/chat"
    client.post("/cats", data={"name": "RedactURL", "email": "ru@t.ca",
                               "description": desc},
                files=[("photos", ("r.jpg", jpg((90, 200, 90)), "image/jpeg"))],
                headers=ip(44))
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("s.jpg", jpg((92, 198, 92)), "image/jpeg"))],
                     headers=ip(44)))
    m = next((m for m in sd.get("matches", []) if m["name"] == "RedactURL"), None)
    assert m, "candidat attendu"
    d = m.get("description") or ""
    assert "mon-site.example" not in d and "123 rue" not in d, \
        f"URL/adresse non masquée : {d!r}"


def test_forget_with_shared_sighting_multi_leads(client):
    """Régression (audit round 3) : un signalement partagé portant des leads vers
    DEUX chats candidats ne doit pas bloquer la suppression de l'un (FK Postgres).
    Avec PRAGMA foreign_keys=ON, ce test échouerait sur l'ancien delete_my_cat."""
    # deux chats de couleur quasi identique → tous deux candidats du même signalement
    ra = client.post("/cats", data={"name": "Jumeau A", "email": "ja@t.ca"},
                     files=[("photos", ("a.jpg", jpg((180, 90, 40)), "image/jpeg"))],
                     headers=ip(50))
    client.post("/cats", data={"name": "Jumeau B", "email": "jb@t.ca"},
                files=[("photos", ("b.jpg", jpg((181, 91, 41)), "image/jpeg"))],
                headers=ip(50))
    token_a = ra.json()["claim_token"]
    sd = sight(client, client.post("/sightings",
                     files=[("photos", ("s.jpg", jpg((180, 90, 40)), "image/jpeg"))],
                     headers=ip(50)))
    sid = sd["sighting_id"]
    # déposer un lead sur CHAQUE candidat proposé (même signalement)
    for m in sd.get("matches", []):
        client.post(f"/sightings/{sid}/lead",
                    json={"candidate_token": m["candidate_token"], "finder_contact": "x"},
                    headers=ip(50))
    # supprimer le chat A : ne doit PAS lever d'IntegrityError malgré le lead vers B
    r = client.post("/cats/track/delete", data={"token": token_a}, headers=ip(50))
    assert r.status_code == 200 and r.json()["deleted"], \
        "la suppression doit réussir même avec un signalement partagé multi-leads"


def test_photos_encrypted_at_rest(client):
    """Option A : la photo est CHIFFRÉE en base (pas de JPEG en clair) mais se sert
    déchiffrée via l'URL signée."""
    from app import crypto, models
    from app.database import SessionLocal
    r = client.post("/cats", data={"name": "Crypt", "email": "crypt@t.ca",
                                    "phone": "514-555-7777"},
                    files=[("photos", ("c.jpg", jpg((30, 60, 120)), "image/jpeg"))],
                    headers=ip(60))
    assert r.status_code == 200
    db = SessionLocal()
    try:
        cat = db.query(models.Cat).filter(models.Cat.email == "crypt@t.ca").first()
        blob = cat.photos[0].data
        assert blob.startswith(b"enc1:"), "le blob stocké doit être chiffré"
        assert b"\xff\xd8\xff" not in blob[:8], "pas d'en-tête JPEG en clair"
        assert crypto.decrypt_bytes(blob)[:3] == b"\xff\xd8\xff", "déchiffré = JPEG"
        # téléphone chiffré en base, déchiffrable
        assert cat.phone.startswith("enc1:") and crypto.decrypt_str(cat.phone) == "5145557777"
        # contact_info ne contient PAS le téléphone en clair
        assert "5145557777" not in (cat.contact_info or "")
    finally:
        db.close()
    # la photo se sert bien déchiffrée
    ov = client.get("/admin/api/overview", headers={**ADMIN, **ip(60)}).json()
    cat_o = next(c for c in ov["cats"] if c["name"] == "Crypt")
    img = client.get(cat_o["photos"][0], headers=ip(60))
    assert img.status_code == 200 and img.content[:3] == b"\xff\xd8\xff"
    assert cat_o["phone"] == "5145557777", "l'admin voit le téléphone déchiffré"


def test_retention_purge(client):
    """Purge de rétention (Loi 25) : fiche inactive + OTP expiré supprimés,
    fiche récente conservée."""
    from datetime import datetime, timezone, timedelta
    from app import models, retention
    from app.database import SessionLocal
    old = datetime.now(timezone.utc) - timedelta(days=400)
    cid_old = client.post("/cats", data={"name": "Vieux", "email": "vieux@t.ca"},
                          files=[("photos", ("v.jpg", jpg((11, 22, 33)), "image/jpeg"))],
                          headers=ip(70)).json()["id"]
    cid_new = client.post("/cats", data={"name": "Recent", "email": "recent@t.ca"},
                          files=[("photos", ("r.jpg", jpg((44, 55, 66)), "image/jpeg"))],
                          headers=ip(70)).json()["id"]
    db = SessionLocal()
    try:
        db.query(models.Cat).filter(models.Cat.id == cid_old).update(
            {models.Cat.last_activity_at: old}, synchronize_session=False)
        db.add(models.EmailOTP(email="exp@t.ca", code_hash="x",
                               expires_at=datetime.now(timezone.utc) - timedelta(hours=1)))
        db.commit()
        res = retention.purge(db)
    finally:
        db.close()
    assert res["cats"] >= 1 and res["otps"] >= 1
    db = SessionLocal()
    try:
        assert db.get(models.Cat, cid_old) is None, "fiche inactive doit être purgée"
        assert db.get(models.Cat, cid_new) is not None, "fiche récente doit rester"
    finally:
        db.close()


def test_purge_requires_admin(client):
    assert client.post("/admin/api/purge", headers=ip(71)).status_code == 403


def test_daily_image_quota(client):
    """Anti-flood : pas plus de N images/jour/IP (ici N=2 pour le test)."""
    from app import ratelimit
    orig = ratelimit._DAILY_IMAGE_LIMIT
    ratelimit._DAILY_IMAGE_LIMIT = 2
    try:
        h = ip(80)
        codes = []
        for i in range(3):
            r = client.post("/cats", data={"name": f"Q{i}", "email": "q@t.ca"},
                            files=[("photos", (f"{i}.jpg", jpg(), "image/jpeg"))], headers=h)
            codes.append(r.status_code)
        assert codes[0] == 200 and codes[1] == 200
        assert codes[2] == 429, f"la 3e photo dépasse le quota de 2/jour : {codes}"
    finally:
        ratelimit._DAILY_IMAGE_LIMIT = orig


def test_animal_gate_disabled_in_tests():
    """En test la vérif animale est OFF → has_animal renvoie None (fail-open),
    sans charger de modèle."""
    from app.ml import detection
    assert detection.has_animal(Image.new("RGB", (40, 40), (9, 9, 9))) is None


def test_color_compatibility():
    """Vérifie que des chats aux attributs de couleur polarisés subissent une pénalité."""
    import json
    from app import models
    from app.matching import _color_compatibility

    p_black = models.Photo(
        attributes=json.dumps({"brightness": 0.15, "dominant_colors": [["noir", 0.9]]})
    )
    p_white = models.Photo(
        attributes=json.dumps({"brightness": 0.85, "dominant_colors": [["blanc", 0.9]]})
    )
    p_similar = models.Photo(
        attributes=json.dumps({"brightness": 0.80, "dominant_colors": [["blanc", 0.8]]})
    )

    factor_diff = _color_compatibility([p_black], [p_white])
    factor_same = _color_compatibility([p_white], [p_similar])

    assert factor_diff == 0.75, f"Expected 0.75 penalty for black vs white cat, got {factor_diff}"
    assert factor_same == 1.0, f"Expected 1.0 for similar cats, got {factor_same}"