fix: aligne la version exposee sur la 1.1.0 et documente la config de deploiement

L'API se declarait en 1.0.0 (config.py et docs/openapi.json) alors que le
CHANGELOG et le dossier annoncent la 1.1.0 comme version livree : GET /health
et Swagger renvoyaient donc une version fausse.

- API_VERSION 1.0.0 -> 1.1.0 (config.py + docs/openapi.json, spec inchangee
  par ailleurs : 29 chemins, 40 operations)
- ajoute api.env.example, modele de configuration sans valeurs
- ignore api.env : il porte la chaine de connexion et la cle JWT

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.gitea/workflows/build.yml

@@ -16,7 +16,10 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Build image
-        run: docker build -t "${IMAGE}:latest" -t "${IMAGE}:${GITHUB_SHA::12}" .
+        run: |
+          docker build \
+            --build-arg APP_BUILD="${GITHUB_SHA::12}" \
+            -t "${IMAGE}:latest" -t "${IMAGE}:${GITHUB_SHA::12}" .
 
       - name: Tests (pytest dans l'image, BDD simulée)
         run: |

.gitignore

@@ -150,6 +150,9 @@ activemq-data/
 # Environments
 .env
 .envrc
+# Configuration de déploiement : contient la chaîne de connexion et
+# la clé JWT. Seul api.env.example (sans valeurs) est versionné.
+api.env
 .venv
 env/
 venv/

Dockerfile

@@ -17,5 +17,10 @@ COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 COPY . .
 
+# SHA du commit construit, injecté par la CI et exposé par GET /health :
+# permet de vérifier quelle version tourne réellement après un déploiement.
+ARG APP_BUILD=local
+ENV APP_BUILD=$APP_BUILD
+
 EXPOSE 8000
 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -34,7 +34,8 @@ sur `localhost`. Pour une instance nommée, définir `DB_SERVER` (ex.
 | `DB_TRUSTED_CONNECTION` | force l'authentification Windows même si `DB_USER` est défini | — |
 | `DB_DRIVER` | pilote ODBC | `ODBC Driver 18 for SQL Server` |
 | `CORS_ORIGINS` | origines autorisées (séparées par `,`) | `localhost:5173,localhost:3000` |
-| `JWT_SECRET` | clé de signature JWT | placeholder (à définir en prod) |
+| `JWT_SECRET` | clé de signature JWT — **obligatoire en déploiement** | clé aléatoire régénérée à chaque démarrage |
+| `APP_BUILD` | SHA du commit construit, injecté par la CI et exposé par `/health` | `local` |
 | `JWT_ALGORITHM` / `JWT_EXPIRE_MINUTES` | algo / durée du token | `HS256` / `60` |
 
 ## Authentification & rôles
@@ -96,8 +97,26 @@ routeurs). Chaque domaine fonctionnel vit dans `routers/` :
 
 Spécification complète : `GET /openapi.json` (export dans `docs/openapi.json`).
 
+## Vérifier qu'un déploiement a pris
+
+`GET /health` expose le SHA du commit dont l'image a été construite :
+
+```bash
+curl -s https://datasentinel-api.nfteam.ovh/health
+# {"api":"ok","database":"ok","version":"1.0.0","build":"cb84e20e1f2a",...}
+```
+
+Comparer `build` au dernier commit poussé sur `main`. S'ils diffèrent, le
+conteneur tourne encore une ancienne image : `docker compose pull` puis
+`docker compose up -d` (un `up -d` seul ne retélécharge rien). Le front expose
+la même information sur `/version.json`.
+
 ## Sécurité
 
+- `JWT_SECRET` doit être défini en déploiement. À défaut, l'API démarre quand
+  même mais tire une clé aléatoire à chaque lancement (sessions perdues au
+  redémarrage) : aucun secret de repli n'est écrit dans le dépôt, un secret
+  public permettrait de forger un jeton d'administrateur.
 - En-têtes : `X-Content-Type-Options`, `X-Frame-Options`, `Referrer-Policy`, `Strict-Transport-Security`.
 - CORS restreint aux origines `CORS_ORIGINS`, tous verbes + credentials.
 - Requêtes SQL **paramétrées** (noms de tables/colonnes whitelistés) ; le compte applicatif
@@ -110,9 +129,19 @@ Spécification complète : `GET /openapi.json` (export dans `docs/openapi.json`)
 ```bash
 pip install pytest httpx && pytest -q
 ```
-Curseur SQL simulé (aucune vraie BDD) : santé, auth (succès/échec), protection 401,
+
-RBAC 403, 404, validation des rôles par enum (422), refus de suppression d'un
+**82 tests**, curseur SQL simulé (aucune vraie base requise) :
-référentiel rattaché (409), désactivation logique d'un monitoring.
+
+| Fichier | Domaine testé |
+|---------|---------------|
+| `test_endpoints.py` | santé, référentiels, authentification, filtres du dashboard, rôles par enum |
+| `test_historique.py` | bornes de dates, validation du format, agrégation des courbes par monitoring et par service |
+| `test_securite.py` | protection 401 de chaque route, RBAC 403, jeton forgé rejeté, en-têtes de sécurité, rate-limit 429, compte désactivé, droits RGPD |
+| `test_referentiels.py` | CRUD complet, refus 409 sur rattachement, modification partielle, paramétrage des requêtes de recherche |
+
+Le compteur anti brute-force est remis à zéro entre les tests (fixture
+`limiteur_vierge`) : c'est un état global, et sans cela les tests de connexion
+se comptabilisent entre eux et échouent selon l'ordre d'exécution.
 
 ## Docker & CI/CD
 

RUNBOOK.md

@@ -77,4 +77,4 @@ docker compose up -d datasentinel-api
 ## Contacts
 
 - Hébergement / infra : administrateur homelab (neckfire).
-- Application / code : A. Coyaud (auteur, dépôts GitHub).
+- Application / code : A. Coyaud (auteur, dépôts Gitea — git.nfteam.ovh).

api.env.example

@@ -0,0 +1,51 @@
+# ============================================================
+#  Data Sentinel — modèle de configuration de l'API
+#
+#  Copier ce fichier en api.env (à côté du docker-compose.yml)
+#  et renseigner les valeurs. api.env n'est JAMAIS versionné :
+#  il est exclu par .gitignore.
+#
+#      cp Api-DataSentinel/api.env.example api.env
+# ============================================================
+
+# ---- Base de données ---------------------------------------
+# Hôte SQL Server. En conteneur, le nom du service Docker
+# (ex. dev-mssql) ; en local, localhost ou HOTE\INSTANCE.
+DB_SERVER=dev-mssql
+
+# Port. Laisser VIDE pour une instance nommée (HOTE\INSTANCE).
+DB_PORT=1433
+
+DB_NAME=DataSentinel
+
+# Compte applicatif dédié — surtout pas « sa ».
+# Création : voir README.md §3.3 (db_datareader + db_datawriter
+# sur la seule base DataSentinel).
+DB_USER=datasentinel_app
+DB_PASSWORD=
+
+# Pilote ODBC embarqué dans l'image.
+DB_DRIVER=ODBC Driver 18 for SQL Server
+
+# Mettre à yes pour forcer l'authentification Windows
+# (poste de développement uniquement, sans DB_USER).
+DB_TRUSTED_CONNECTION=
+
+# ---- Sécurité ----------------------------------------------
+# OBLIGATOIRE en déploiement. Sans cette variable, l'API démarre
+# mais tire une clé aléatoire à chaque redémarrage : toutes les
+# sessions sont alors invalidées au moindre restart.
+# Générer :  python -c "import secrets; print(secrets.token_urlsafe(64))"
+JWT_SECRET=
+
+JWT_ALGORITHM=HS256
+JWT_EXPIRE_MINUTES=60
+
+# ---- CORS ---------------------------------------------------
+# Origines autorisées, séparées par des virgules. Doit contenir
+# l'URL exacte du front (schéma + hôte, sans slash final).
+CORS_ORIGINS=https://datasentinel.nfteam.ovh
+
+# ---- Build --------------------------------------------------
+# Injecté par la CI (SHA court du commit) et exposé par GET /health.
+APP_BUILD=local

config.py

@@ -3,10 +3,13 @@
 #  Data Sentinel | COYAUD Anthony | 2026
 # ============================================================
 
+import logging
 import os
-import pyodbc
+import secrets
 from contextlib import contextmanager
 
+import pyodbc
+
 
 def _build_connection_string() -> str:
     """
@@ -45,17 +48,48 @@ def _build_connection_string() -> str:
     )
 
 
+def _resolve_jwt_secret() -> str:
+    """
+    Clé de signature des jetons.
+
+    Elle doit venir de JWT_SECRET. À défaut, on tire une clé aléatoire au
+    démarrage plutôt que de retomber sur une valeur écrite dans le dépôt :
+    un secret public permettrait à quiconque lit le code de forger un jeton
+    d'administrateur.
+
+    Conséquence assumée du repli : la clé change à chaque redémarrage, donc
+    les sessions en cours sont invalidées. C'est visible et sans gravité en
+    développement, et le message ci-dessous dit quoi faire en déploiement.
+    On ne bloque volontairement pas le démarrage, pour ne pas transformer un
+    oubli de configuration en indisponibilité totale du service.
+    """
+    secret = os.getenv("JWT_SECRET")
+    if secret:
+        return secret
+
+    logging.getLogger("uvicorn.error").warning(
+        "JWT_SECRET n'est pas defini : une cle aleatoire est generee pour cette "
+        "execution. Les sessions seront perdues a chaque redemarrage. "
+        "Definir JWT_SECRET dans l'environnement (api.env) pour un deploiement."
+    )
+    return secrets.token_urlsafe(64)
+
+
 class Config:
     # Chaîne de connexion SQL Server (env en prod, Windows en local)
     DB_CONNECTION_STRING = _build_connection_string()
 
     # Paramètres API
     API_TITLE       = "Data Sentinel API"
-    API_VERSION     = "1.0.0"
+    API_VERSION     = "1.1.0"
     API_DESCRIPTION = "API de monitoring de la qualité des données — XEFI"
 
+    # Identifiant de build injecté par la CI : permet de vérifier quelle
+    # version tourne réellement après un déploiement (voir GET /health).
+    BUILD           = os.getenv("APP_BUILD", "local")
+
     # Sécurité JWT
-    SECRET_KEY      = os.getenv("JWT_SECRET", "data-sentinel-secret-change-in-prod")
+    SECRET_KEY      = _resolve_jwt_secret()
     ALGORITHM       = os.getenv("JWT_ALGORITHM", "HS256")
     TOKEN_EXPIRE_MINUTES = int(os.getenv("JWT_EXPIRE_MINUTES", "60"))
 

main.py

@@ -81,7 +81,13 @@ async def security_headers(request: Request, call_next):
 
 @app.get("/health", tags=["Système"])
 def health_check():
-    """Ping API + test connexion SQL Server."""
+    """
+    Ping API + test connexion SQL Server.
+
+    `build` porte le SHA du commit dont l'image a été construite (injecté par
+    la CI). C'est le seul moyen fiable de vérifier qu'un déploiement a bien
+    pris : comparer ce champ au dernier commit poussé.
+    """
     try:
         with get_cursor() as cursor:
             cursor.execute("SELECT 1")
@@ -94,6 +100,7 @@ def health_check():
         "api"            : "ok",
         "database"       : db_status,
         "version"        : Config.API_VERSION,
+        "build"          : Config.BUILD,
         "nb_monitorings" : len(MONITO_TABLES),
     }
 

testAPI.py

@@ -1,47 +0,0 @@
-import requests
-import json
-
-# Configuration
-BASE_URL = "http://127.0.0.1:8000"
-
-def test_get_dashboard():
-    print(f"--- Test de l'endpoint: {BASE_URL}/dashboard ---")
-    
-    try:
-        # 1. Envoi de la requête
-        response = requests.get(f"{BASE_URL}/dashboard")
-        
-        # 2. Vérification du code statut
-        if response.status_code == 200:
-            data = response.json()
-            
-            if not data:
-                print("L'API a répondu avec succès, mais le tableau est vide (VUE_CONSO ne contient rien).")
-                return
-
-            # 3. Affichage formaté du tableau
-            print(f"{'ID':<5} | {'NOM DU MONITORING':<40} | {'ERREURS':<8} | {'SERVICE':<15}")
-            print("-" * 75)
-            
-            for item in data:
-                print(f"{item['id_monito']:<5} | {item['nom_monito'][:38]:<40} | {item['nb_erreurs']:<8} | {item['service']:<15}")
-            
-            print(f"\nTotal de lignes récupérées : {len(data)}")
-            
-        else:
-            print(f"Erreur lors de la requête : {response.status_code}")
-            print(response.text)
-            
-    except requests.exceptions.ConnectionError:
-        print("Erreur : Impossible de se connecter à l'API. Est-ce qu'uvicorn est lancé ?")
-
-if __name__ == "__main__":
-    # Petit check de santé avant de tester les données
-    print("Vérification de l'état de l'API...")
-    health = requests.get(f"{BASE_URL}/health").json()
-    print(f"Statut API : {health['api']} | Statut DB : {health['database']}\n")
-    
-    if health['database'] == "ok":
-        test_get_dashboard()
-    else:
-        print("Abandon du test : La base de données n'est pas accessible.")

tests/conftest.py

@@ -9,6 +9,7 @@ from fastapi.testclient import TestClient
 
 import main
 from auth import create_access_token
+from rate_limit import limiter
 from routers import (
     admin,
     authentification,
@@ -62,6 +63,21 @@ def cur(monkeypatch):
     return c
 
 
+@pytest.fixture(autouse=True)
+def limiteur_vierge():
+    """
+    Remet le compteur anti brute-force à zéro entre les tests.
+
+    Le limiteur est un état global partagé par toute l'application : sans
+    remise à zéro, les tests de connexion se comptabilisent entre eux et
+    finissent par recevoir un 429, avec des échecs qui dépendent de l'ordre
+    d'exécution et ne se reproduisent pas fichier par fichier.
+    """
+    limiter.reset()
+    yield
+    limiter.reset()
+
+
 @pytest.fixture
 def client():
     return TestClient(main.app)

tests/test_historique.py

@@ -0,0 +1,111 @@
+"""Historique et courbes d'évolution (curseur SQL simulé via la fixture `cur`)."""
+
+
+def _requetes(cur) -> str:
+    """Concatène les requêtes exécutées, pour vérifier le SQL construit."""
+    return " ".join(q for q, _ in cur.executed)
+
+
+def _parametres(cur) -> list:
+    return [p for _, args in cur.executed for p in args]
+
+
+def test_historique_exige_authentification(client):
+    assert client.get("/historique").status_code == 401
+
+
+def test_historique_sans_filtre_ne_restreint_rien(client, cur, auth_headers):
+    cur.description = [("id_monito",), ("nb_erreurs",)]
+    cur._rows = [(1, 5)]
+    r = client.get("/historique", headers=auth_headers)
+    assert r.status_code == 200
+    requete = _requetes(cur)
+    assert "date_sauvegarde >=" not in requete
+    assert "date_sauvegarde <=" not in requete
+
+
+def test_historique_applique_les_bornes_de_dates(client, cur, auth_headers):
+    cur.description = [("id_monito",)]
+    cur._rows = []
+    r = client.get(
+        "/historique",
+        params={"date_debut": "2026-01-01", "date_fin": "2026-08-15"},
+        headers=auth_headers,
+    )
+    assert r.status_code == 200
+    requete = _requetes(cur)
+    assert "date_sauvegarde >= ?" in requete
+    assert "date_sauvegarde <= ?" in requete
+    # Les dates sont passées en paramètres, jamais concaténées dans le SQL.
+    assert "2026-01-01" in _parametres(cur)
+    assert "2026-08-15" in _parametres(cur)
+
+
+def test_historique_refuse_une_date_mal_formee(client, cur, auth_headers):
+    r = client.get("/historique", params={"date_debut": "15/01/2026"}, headers=auth_headers)
+    assert r.status_code == 422
+
+
+def test_historique_filtre_par_monitoring(client, cur, auth_headers):
+    cur.description = [("id_monito",)]
+    cur._rows = []
+    client.get("/historique", params={"id_monito": 3}, headers=auth_headers)
+    assert "id_monito = ?" in _requetes(cur)
+    assert 3 in _parametres(cur)
+
+
+def test_evolution_monitoring_inconnu_renvoie_404(client, cur, auth_headers):
+    # 999 n'est pas dans MONITO_TABLES : on refuse avant d'interroger la base.
+    r = client.get("/historique/999/evolution", headers=auth_headers)
+    assert r.status_code == 404
+
+
+def test_evolution_formate_les_points_pour_le_graphique(client, cur, auth_headers):
+    cur._rows = [("2026-01-01", 12), ("2026-01-02", 9)]
+    r = client.get("/historique/1/evolution", headers=auth_headers)
+    assert r.status_code == 200
+    corps = r.json()
+    assert corps["points"] == 2
+    assert corps["evolution"][0] == {"date": "2026-01-01", "nb_erreurs": 12}
+
+
+def test_evolution_globale_renvoie_une_serie(client, cur, auth_headers):
+    cur._rows = [("2026-01-01", 56), ("2026-01-02", 54)]
+    r = client.get("/evolution/global", headers=auth_headers)
+    assert r.status_code == 200
+    corps = r.json()
+    assert corps["type"] == "GLOBAL"
+    assert corps["points"] == 2
+    assert corps["series"][0]["nb_erreurs"] == 56
+
+
+def test_evolution_par_monitoring_regroupe_par_courbe(client, cur, auth_headers):
+    # Lignes à plat en base -> une série par monitoring côté API.
+    cur._rows = [
+        ("2026-01-01", 1, "Tiers payeurs", 5),
+        ("2026-01-02", 1, "Tiers payeurs", 4),
+        ("2026-01-01", 2, "Sans RIB", 6),
+    ]
+    r = client.get("/evolution/par-monitoring", headers=auth_headers)
+    assert r.status_code == 200
+    corps = r.json()
+    assert corps["nb_series"] == 2
+    series = {m["id_monito"]: m for m in corps["monitorings"]}
+    assert len(series[1]["serie"]) == 2
+    assert len(series[2]["serie"]) == 1
+    assert series[1]["nom_monito"] == "Tiers payeurs"
+
+
+def test_evolution_par_service_regroupe_par_service(client, cur, auth_headers):
+    cur._rows = [
+        ("2026-01-01", "Contrat", 20),
+        ("2026-01-02", "Contrat", 18),
+        ("2026-01-01", "Fournisseur", 5),
+    ]
+    r = client.get("/evolution/par-service", headers=auth_headers)
+    assert r.status_code == 200
+    corps = r.json()
+    assert corps["nb_series"] == 2
+    par_service = {s["service"]: s["serie"] for s in corps["services"]}
+    assert len(par_service["Contrat"]) == 2
+    assert len(par_service["Fournisseur"]) == 1

tests/test_referentiels.py

@@ -0,0 +1,143 @@
+"""CRUD des référentiels : catégories, contacts et nomenclature."""
+
+
+def _requetes(cur) -> str:
+    return " ".join(q for q, _ in cur.executed)
+
+
+def _parametres(cur) -> list:
+    return [p for _, args in cur.executed for p in args]
+
+
+# ------------------------------------------------------------
+#  Catégories
+# ------------------------------------------------------------
+
+def test_creation_categorie(client, cur, auth_headers):
+    cur._one = (7,)
+    r = client.post("/categories", headers=auth_headers,
+                    json={"intitule_categorie": "Litiges"})
+    assert r.status_code == 201
+    assert r.json() == {"id_categorie": 7, "intitule_categorie": "Litiges"}
+
+
+def test_categorie_intitule_vide_refuse(client, cur, auth_headers):
+    # Contrainte min_length : on ne crée pas un référentiel sans libellé.
+    r = client.post("/categories", headers=auth_headers, json={"intitule_categorie": ""})
+    assert r.status_code == 422
+
+
+def test_suppression_categorie_refusee_si_monitorings_rattaches(client, cur, auth_headers):
+    cur._one = (3,)
+    r = client.delete("/categories/1", headers=auth_headers)
+    assert r.status_code == 409
+    assert "3 monitoring(s)" in r.json()["detail"]
+    # Aucune suppression ne doit avoir été tentée.
+    assert "DELETE FROM CATEGORIE" not in _requetes(cur)
+
+
+def test_suppression_categorie_libre_autorisee(client, cur, auth_headers):
+    cur._one = (0,)
+    r = client.delete("/categories/1", headers=auth_headers)
+    assert r.status_code == 200
+    assert "DELETE FROM CATEGORIE" in _requetes(cur)
+
+
+def test_modification_categorie_inexistante_renvoie_404(client, cur, auth_headers):
+    cur.rowcount = 0
+    r = client.put("/categories/999", headers=auth_headers,
+                   json={"intitule_categorie": "X"})
+    assert r.status_code == 404
+
+
+# ------------------------------------------------------------
+#  Contacts
+# ------------------------------------------------------------
+
+def test_creation_contact(client, cur, auth_headers):
+    cur._one = (12,)
+    r = client.post("/contacts", headers=auth_headers, json={
+        "id_service": 1, "intitule_contact": "Référent",
+        "nom": "Dupont", "prenom": "Marie", "mail": "marie@xefi-fictif.fr",
+    })
+    assert r.status_code == 201
+    assert r.json()["id_contact"] == 12
+    assert r.json()["nom"] == "Dupont"
+
+
+def test_contact_champ_manquant_refuse(client, cur, auth_headers):
+    r = client.post("/contacts", headers=auth_headers, json={"nom": "Dupont"})
+    assert r.status_code == 422
+
+
+def test_suppression_contact_directe(client, cur, auth_headers):
+    # Aucune table ne référence un contact : suppression sans garde-fou.
+    r = client.delete("/contacts/1", headers=auth_headers)
+    assert r.status_code == 200
+    assert "DELETE FROM CONTACT" in _requetes(cur)
+
+
+def test_suppression_contact_inexistant_renvoie_404(client, cur, auth_headers):
+    cur.rowcount = 0
+    assert client.delete("/contacts/999", headers=auth_headers).status_code == 404
+
+
+# ------------------------------------------------------------
+#  Nomenclature des monitorings
+# ------------------------------------------------------------
+
+def test_creation_monitoring_actif_par_defaut(client, cur, auth_headers):
+    cur._one = (8,)
+    r = client.post("/monitorings", headers=auth_headers, json={
+        "monito_intitule": "Nouveau", "id_service": 1, "id_categorie": 1,
+        "table_source": "MONITO_TEST", "bdd_source": "Sage",
+    })
+    assert r.status_code == 201
+    assert "actif" in _requetes(cur)
+
+
+def test_modification_partielle_ne_touche_que_les_champs_fournis(client, cur, auth_headers):
+    r = client.put("/monitorings/1", headers=auth_headers,
+                   json={"monito_intitule": "Renommé"})
+    assert r.status_code == 200
+    requete = _requetes(cur)
+    assert "monito_intitule = ?" in requete
+    assert "bdd_source = ?" not in requete
+
+
+def test_modification_sans_aucun_champ_refusee(client, cur, auth_headers):
+    r = client.put("/monitorings/1", headers=auth_headers, json={})
+    assert r.status_code == 400
+
+
+def test_reactivation_possible_par_modification(client, cur, auth_headers):
+    # Permet de revenir sur une désactivation.
+    r = client.put("/monitorings/7", headers=auth_headers, json={"actif": True})
+    assert r.status_code == 200
+    assert "actif = ?" in _requetes(cur)
+    assert 1 in _parametres(cur)
+
+
+def test_desactivation_monitoring_inexistant_renvoie_404(client, cur, auth_headers):
+    cur.rowcount = 0
+    assert client.delete("/monitorings/999", headers=auth_headers).status_code == 404
+
+
+def test_details_monitoring_recherche_sur_les_colonnes_declarees(client, cur, auth_headers):
+    cur.description = [("id_ligne",)]
+    cur._rows = []
+    r = client.get("/monitorings/1/details", params={"search": "Lyon"}, headers=auth_headers)
+    assert r.status_code == 200
+    requete = _requetes(cur)
+    assert "LIKE ?" in requete
+    # Le terme est passé en paramètre, jamais injecté dans la requête.
+    assert "%Lyon%" in _parametres(cur)
+    assert "Lyon" not in requete
+
+
+def test_details_monitoring_limite_bornee(client, cur, auth_headers):
+    # limit est plafonné à 5000 pour éviter un export non maîtrisé.
+    assert client.get("/monitorings/1/details", params={"limit": 99999},
+                      headers=auth_headers).status_code == 422
+    assert client.get("/monitorings/1/details", params={"limit": 0},
+                      headers=auth_headers).status_code == 422

tests/test_securite.py

@@ -0,0 +1,159 @@
+"""Contrôles d'accès, en-têtes de sécurité et droits RGPD."""
+import pytest
+
+from auth import create_access_token, hash_password
+from domain import UserRole
+
+
+def _token(role: str, username: str = "u", uid: int = 1) -> dict:
+    return {"Authorization": f"Bearer {create_access_token(
+        {'sub': username, 'role': role, 'uid': uid})}"}
+
+
+# ------------------------------------------------------------
+#  Protection des routes de données
+# ------------------------------------------------------------
+
+@pytest.mark.parametrize("route", [
+    "/categories", "/services", "/contacts", "/monitorings",
+    "/dashboard", "/dashboard/filtres", "/dashboard/summary",
+    "/historique", "/evolution/global", "/admin/users", "/me/data-export",
+])
+def test_toutes_les_routes_de_donnees_exigent_un_jeton(client, route):
+    assert client.get(route).status_code == 401
+
+
+def test_health_reste_public(client, cur):
+    cur._one = (1,)
+    assert client.get("/health").status_code == 200
+
+
+def test_jeton_invalide_rejete(client):
+    r = client.get("/services", headers={"Authorization": "Bearer pas-un-vrai-jeton"})
+    assert r.status_code == 401
+
+
+def test_jeton_signe_avec_une_autre_cle_rejete(client):
+    # Un jeton forgé avec un autre secret ne doit pas ouvrir l'accès :
+    # c'est ce que garantissait mal un secret de repli écrit dans le dépôt.
+    from jose import jwt
+    faux = jwt.encode({"sub": "pirate", "role": "Admin", "uid": 99},
+                      "mauvaise-cle", algorithm="HS256")
+    r = client.get("/admin/users", headers={"Authorization": f"Bearer {faux}"})
+    assert r.status_code == 401
+
+
+# ------------------------------------------------------------
+#  RBAC : seuls les administrateurs écrivent
+# ------------------------------------------------------------
+
+@pytest.mark.parametrize("role", [UserRole.SUPERVISEUR.value, UserRole.CONSULTANT.value])
+@pytest.mark.parametrize("methode,route,corps", [
+    ("post",   "/services",     {"nom_service": "X"}),
+    ("put",    "/services/1",   {"nom_service": "X"}),
+    ("delete", "/services/1",   None),
+    ("post",   "/categories",   {"intitule_categorie": "X"}),
+    ("post",   "/monitorings",  {"monito_intitule": "X", "id_service": 1,
+                                 "id_categorie": 1, "table_source": "T", "bdd_source": "Sage"}),
+    ("get",    "/admin/users",  None),
+    ("get",    "/admin/journal", None),
+])
+def test_ecriture_interdite_hors_admin(client, cur, role, methode, route, corps):
+    appel = getattr(client, methode)
+    reponse = appel(route, headers=_token(role), json=corps) if corps else appel(route, headers=_token(role))
+    assert reponse.status_code == 403
+
+
+def test_lecture_autorisee_pour_un_consultant(client, cur):
+    cur.description = [("id_service",), ("nom_service",)]
+    cur._rows = [(1, "Contrat")]
+    r = client.get("/services", headers=_token(UserRole.CONSULTANT.value))
+    assert r.status_code == 200
+
+
+# ------------------------------------------------------------
+#  En-têtes de sécurité
+# ------------------------------------------------------------
+
+@pytest.mark.parametrize("entete,valeur", [
+    ("X-Content-Type-Options", "nosniff"),
+    ("X-Frame-Options", "DENY"),
+    ("Referrer-Policy", "strict-origin-when-cross-origin"),
+])
+def test_entetes_de_securite_presents(client, cur, entete, valeur):
+    cur._one = (1,)
+    assert client.get("/health").headers[entete] == valeur
+
+
+def test_strict_transport_security_present(client, cur):
+    cur._one = (1,)
+    assert "max-age=" in client.get("/health").headers["Strict-Transport-Security"]
+
+
+# ------------------------------------------------------------
+#  Authentification
+# ------------------------------------------------------------
+
+def test_compte_desactive_ne_peut_pas_se_connecter(client, cur):
+    # actif = 0 : le mot de passe est correct mais l'accès doit être refusé.
+    cur._one = (1, "ancien", hash_password("Bon2026!"), "Admin", 0)
+    r = client.post("/auth/login", data={"username": "ancien", "password": "Bon2026!"})
+    assert r.status_code == 401
+
+
+def test_utilisateur_inconnu_renvoie_401(client, cur):
+    cur._one = None
+    r = client.post("/auth/login", data={"username": "fantome", "password": "x"})
+    assert r.status_code == 401
+
+
+def test_le_mot_de_passe_nest_jamais_renvoye(client, cur):
+    cur._one = (1, "admin", hash_password("Admin2026!"), "Admin", 1)
+    r = client.post("/auth/login", data={"username": "admin", "password": "Admin2026!"})
+    assert r.status_code == 200
+    assert "password" not in r.text
+    assert "hash" not in r.text
+
+
+def test_rate_limit_bloque_le_brute_force(client, cur):
+    """La 6e tentative en moins d'une minute doit être refusée (5/minute)."""
+    cur._one = (1, "admin", hash_password("Admin2026!"), "Admin", 1)
+    codes = [
+        client.post("/auth/login", data={"username": "admin", "password": "faux"}).status_code
+        for _ in range(6)
+    ]
+    assert codes[:5] == [401] * 5
+    assert codes[5] == 429
+
+
+def test_connexion_journalisee(client, cur):
+    cur._one = (1, "admin", hash_password("Admin2026!"), "Admin", 1)
+    client.post("/auth/login", data={"username": "admin", "password": "Admin2026!"})
+    requetes = " ".join(q for q, _ in cur.executed)
+    assert "INSERT INTO JOURNAL_AUDIT" in requetes
+    # L'action passe en paramètre, pas concaténée dans la requête.
+    assert "LOGIN" in [p for _, args in cur.executed for p in args]
+
+
+# ------------------------------------------------------------
+#  RGPD
+# ------------------------------------------------------------
+
+def test_export_de_mes_donnees_retourne_profil_et_journal(client, cur, auth_headers):
+    cur.description = [("id_user",), ("username",)]
+    cur._one = (1, "admin")
+    cur._rows = [(1, "admin")]
+    r = client.get("/me/data-export", headers=auth_headers)
+    assert r.status_code == 200
+    assert set(r.json()) == {"profile", "journal"}
+
+
+def test_suppression_de_compte_anonymise_au_lieu_de_supprimer(client, cur, auth_headers):
+    # Droit à l'oubli : le compte est désactivé et anonymisé, la ligne reste
+    # pour préserver l'intégrité du journal d'audit.
+    r = client.delete("/me", headers=auth_headers)
+    assert r.status_code == 200
+    requetes = " ".join(q for q, _ in cur.executed)
+    assert "UPDATE [USER]" in requetes
+    assert "DELETE FROM [USER]" not in requetes
+    assert "JOURNAL_AUDIT SET username = NULL" in requetes