"""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}"