Author SHA1 Message Date
Antho e6afc4f1fa chore(sql): compte de visite « Juré » + dump d'authentification versionné
Ajoute sql/create_admin_jure.sql, script idempotent créant/réinitialisant
le compte de démonstration destiné aux personnes qui consultent le site
(rôle Admin). Aucune instruction USE : la base cible vient du paramètre
-d de la connexion, avec un garde-fou qui arrête le script si les tables
attendues sont absentes.

Copie aussi sql/data_sentinel_auth.sql (déjà présent dans le dump livré
sous RENDU/02_Dump_SQL/) : le README de ce dépôt le référençait déjà,
mais le dossier sql/ n'existait pas encore ici.

Documente le compte dans le tableau des comptes de démo du README.
2026-08-15 13:28:23 +02:00
Antho fd4c085ee4 feat(dashboard): endpoint /dashboard/filtres restreint aux valeurs utilisées
VUE_CONSO ne rattache pas tous les services/catégories des référentiels
SERVICE et CATEGORIE à un monitoring actif (ex. "Business Intelligence").
Les proposer tels quels dans les filtres du dashboard menait à un écran
vide. Le nouvel endpoint renvoie un SELECT DISTINCT service, categorie
sur VUE_CONSO, avec les combinaisons pour permettre un filtrage en
cascade côté frontend.

Ajoute un test dédié et régénère docs/openapi.json.
2026-08-15 13:28:11 +02:00
neckfire 66436629af docs: README complet + RUNBOOK + CHANGELOG + export openapi.json
Build & Deploy / build (push) Successful in 12s
2026-06-20 13:06:59 +02:00
neckfire af00ab36cc test: pytest (auth, RBAC, endpoints) + étape CI build->test->push
Build & Deploy / build (push) Successful in 11s
2026-06-20 12:57:30 +02:00
neckfire 8563e6daac feat(auth): JWT + RBAC, RGPD, sécurité (P0)
Build & Deploy / build (push) Successful in 17s
- auth.py: JWT (python-jose) + bcrypt (passlib), get_current_user, require_admin
- endpoints protégés via APIRouter(dependencies=[get_current_user]); /health public
- /auth/login (rate-limit 5/min) + /auth/me
- /admin/users CRUD + reset-password + /admin/journal (require_admin)
- RGPD: /me/data-export (portabilité) + DELETE /me (oubli/anonymisation)
- en-têtes de sécurité, CORS tous verbes + credentials
- requirements: python-jose, passlib[bcrypt], bcrypt 4.0.1, python-multipart, slowapi
2026-06-20 12:49:01 +02:00
neckfire 51d041c0da ci: dockerise l'API + connexion DB et CORS configurables par env
- config.py: connexion SQL via DB_* (driver 18, auth SQL), fallback Windows local
- main.py: origines CORS via CORS_ORIGINS
- Dockerfile (python 3.12 + msodbcsql18) + workflow Gitea Actions (build/push image)
2026-06-20 12:21:47 +02:00
15 changed files with 931 additions and 49 deletions
+44
View File
@@ -0,0 +1,44 @@
name: Build & Deploy
on:
push:
branches: [main]
workflow_dispatch: {}
env:
IMAGE: git.nfteam.ovh/neckfire/datasentinel-api
jobs:
build:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Build image
run: docker build -t "${IMAGE}:latest" -t "${IMAGE}:${GITHUB_SHA::12}" .
- name: Tests (pytest dans l'image, BDD simulée)
run: |
docker run --rm "${IMAGE}:${GITHUB_SHA::12}" \
sh -c 'pip install -q pytest httpx && pytest -q'
- name: Push image
run: |
set -e
echo "${{ secrets.REGISTRY_PASSWORD }}" | docker login git.nfteam.ovh -u "${{ secrets.REGISTRY_USER }}" --password-stdin
docker push --all-tags "${IMAGE}"
echo "pushed ${IMAGE}"
- name: Notify ntfy (success/failure)
if: always()
run: |
if [ "${{ job.status }}" = "success" ]; then EMOJI="white_check_mark"; PRIO="default"; else EMOJI="rotating_light"; PRIO="high"; fi
curl -s \
-H "Authorization: Bearer ${{ secrets.NTFY_TOKEN }}" \
-H "Title: ${GITHUB_REPOSITORY} — ${{ job.status }}" \
-H "Priority: ${PRIO}" \
-H "Tags: ${EMOJI}" \
-H "Click: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions" \
-d "${GITHUB_WORKFLOW} sur ${GITHUB_REF_NAME} (run #${GITHUB_RUN_NUMBER}) : ${{ job.status }}" \
"${{ secrets.NTFY_URL }}/${{ secrets.NTFY_TOPIC }}" || true
+29
View File
@@ -0,0 +1,29 @@
# Changelog
Format : [Keep a Changelog](https://keepachangelog.com/fr/) · Versioning sémantique.
## [1.1.0] — 2026-06-20
### Ajouté
- Authentification **JWT** (`/auth/login`, `/auth/me`) + hachage bcrypt (`auth.py`).
- **RBAC** : protection de tous les endpoints de données (401 sans token), routes `/admin/*`
réservées au rôle `Admin` (403 sinon).
- Endpoints d'administration : CRUD utilisateurs (`/admin/users`), reset de mot de passe,
journal d'audit (`/admin/journal`).
- **RGPD** : `GET /me/data-export` (portabilité), `DELETE /me` (droit à l'oubli + anonymisation).
- En-têtes de sécurité HTTP + rate-limit `5/min` sur le login (`slowapi`).
- Connexion DB et CORS configurables par variables d'environnement (auth SQL, Driver 18).
- `Dockerfile` + CI Gitea Actions (build → tests pytest → push image) + notifications ntfy.
- Tests `pytest` (auth, RBAC, endpoints) avec curseur SQL simulé.
- Tables `[USER]` + `JOURNAL_AUDIT` et comptes de démo.
### Modifié
- CORS : tous les verbes + credentials (au lieu de `GET` seul).
### Sécurité
- Plus aucun secret en dur côté déploiement (secrets via env / fichiers hors dépôt).
- La base s'exécute sur un SQL Server partagé ; compte applicatif non-`sa`.
## [1.0.0] — 2026-04
- Version initiale (A. Coyaud) : API de lecture sur SQL Server (référentiels, monitorings,
dashboard, historique, évolution).
+21
View File
@@ -0,0 +1,21 @@
# API Data Sentinel — FastAPI + pyodbc (ODBC Driver 18 pour SQL Server).
FROM python:3.12-slim
# Pilote Microsoft ODBC 18 (le 17 n'est plus packagé sur Debian 12 / bookworm).
RUN apt-get update && apt-get install -y --no-install-recommends \
curl gnupg ca-certificates apt-transport-https \
&& curl -sSL https://packages.microsoft.com/keys/microsoft.asc \
| gpg --dearmor -o /usr/share/keyrings/microsoft-prod.gpg \
&& echo "deb [arch=amd64,arm64 signed-by=/usr/share/keyrings/microsoft-prod.gpg] https://packages.microsoft.com/debian/12/prod bookworm main" \
> /etc/apt/sources.list.d/mssql-release.list \
&& apt-get update \
&& ACCEPT_EULA=Y apt-get install -y msodbcsql18 unixodbc-dev \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
+87 -23
View File
@@ -1,34 +1,98 @@
# Api-DataSentinel # Api-DataSentinel
API FastAPI pour Data Sentinel. API **FastAPI** de monitoring de la qualité des données — Data Sentinel (XEFI).
Backend en lecture seule sur SQL Server, sécurisé par authentification JWT.
## Installation ## Stack
1. Ouvrir un terminal dans le dossier du projet : - Python 3.12 · FastAPI 0.136 · uvicorn
```powershell - SQL Server 2022 via `pyodbc` (ODBC Driver 18)
cd "c:\Users\antho\Desktop\Projet de fin d'année\Projet fin d_année\DataSentinel\Api-DataSentinel" - Auth : JWT (`python-jose`) + bcrypt (`passlib`), rate-limit (`slowapi`)
```
2. Créer et activer l'environnement virtuel (si nécessaire) :
```powershell
python -m venv .venv
.\.venv\Scripts\Activate.ps1
```
3. Installer les dépendances :
```powershell
pip install -r requirements.txt
```
## Lancement de l'API ## Lancement en local (dev)
Utiliser la commande suivante pour démarrer le serveur :
```powershell ```powershell
.\.venv\Scripts\python.exe -m uvicorn main:app --reload --port 8000 python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
python -m uvicorn main:app --reload --port 8000
``` ```
> Important : si le lanceur `\.venv\Scripts\uvicorn.exe` est cassé après un déplacement de dossier, utilisez toujours `python -m uvicorn`. - Swagger : `http://127.0.0.1:8000/docs` · Santé : `http://127.0.0.1:8000/health`
## Vérification Sans variables d'environnement, l'API se connecte en authentification Windows
(`LaptopCA\SQLEXPRESS`) — comportement de dev d'origine, inchangé.
- Swagger : `http://127.0.0.1:8000/docs` ## Configuration (variables d'environnement)
- Santé : `http://127.0.0.1:8000/health`
| Variable | Rôle | Défaut |
|----------|------|--------|
| `DB_SERVER` | hôte SQL Server (active l'auth SQL si défini) | — (sinon Windows local) |
| `DB_PORT` / `DB_NAME` | port / base | `1433` / `DataSentinel` |
| `DB_USER` / `DB_PASSWORD` | compte applicatif | — |
| `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_ALGORITHM` / `JWT_EXPIRE_MINUTES` | algo / durée du token | `HS256` / `60` |
## Authentification & rôles
- `POST /auth/login` (form `username`/`password`) → `{access_token, token_type, user}`. Rate-limité **5/min**.
- `GET /auth/me` → profil courant.
- **Tous les endpoints de données exigent** `Authorization: Bearer <token>` (401 sinon). Seul `/health` est public.
- Rôles : `Admin`, `Superviseur`, `Consultant`. Les routes `/admin/*` exigent `Admin` (403 sinon).
### Comptes de démo
| Identifiant | Mot de passe | Rôle | Usage |
|-------------|--------------|------|-------|
| `Juré` (`JURE@NEXA.com`) | `123456` | Admin | **Compte de visite / évaluation** — à communiquer aux personnes qui consultent le site |
| `admin` | `Admin2026!` | Admin | Compte d'administration technique |
| `superviseur` | `Super2026!` | Superviseur | Illustration du RBAC (pas d'accès `/admin/*`) |
| `consultant` | `Conseil2026!` | Consultant | Illustration du RBAC (lecture seule) |
Ces comptes sont créés par `sql/data_sentinel_auth.sql`. Pour (re)créer le seul
compte `Juré` sur une base déjà déployée, sans rejouer tout le script d'auth :
```bash
sqlcmd -S <serveur> -d DataSentinel -U <user> -P <mdp> -i sql/create_admin_jure.sql
```
Les mots de passe ci-dessus sont des identifiants de démonstration : ils sont
stockés hachés (bcrypt, coût 12) et doivent être régénérés avant toute mise en
production réelle (`POST /admin/users/{id}/reset-password`).
## Endpoints (résumé)
- **Données** (protégés) : `/categories`, `/services`, `/contacts`, `/monitorings[...]`,
`/dashboard[...]`, `/historique[...]`, `/evolution/*`.
- **Admin** (`Admin`) : `GET/POST /admin/users`, `PUT/DELETE /admin/users/{id}`,
`POST /admin/users/{id}/reset-password`, `GET /admin/journal`.
- **RGPD** : `GET /me/data-export` (portabilité), `DELETE /me` (droit à l'oubli + anonymisation).
Spécification complète : `GET /openapi.json` (export dans `docs/openapi.json`).
## Sécurité
- 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
n'est pas `sa`. Tables d'auth : `[USER]` + `JOURNAL_AUDIT` (`sql/data_sentinel_auth.sql`,
identique au dump livré dans `RENDU/02_Dump_SQL/`).
- Journal d'audit alimenté à chaque login + action admin.
## Tests
```bash
pip install pytest httpx && pytest -q
```
Curseur SQL simulé (aucune vraie BDD) : santé, auth (succès/échec), protection 401, RBAC 403, 404.
## Docker & CI/CD
- `Dockerfile` : image Python 3.12 + `msodbcsql18`.
- CI Gitea Actions (`.gitea/workflows/build.yml`) : **build → pytest (dans l'image) → push**
vers `git.nfteam.ovh/neckfire/datasentinel-api` + notification ntfy. Déclenché sur push `main`.
- Déploiement, exploitation, sauvegarde : voir `RUNBOOK.md` et le stack d'hébergement
(`homelab/dev/datasentinel/`). La base tourne sur un **SQL Server partagé** (`dev-mssql`),
pas dédiée à l'app.
+80
View File
@@ -0,0 +1,80 @@
# RUNBOOK — Data Sentinel (déploiement & exploitation)
Hébergement sur le homelab `nfteam.ovh`. Stack d'exécution :
`~/Documents/homelab/dev/datasentinel/` (API + Front) et `~/Documents/homelab/dev/mssql/`
(SQL Server partagé `dev-mssql`, réseau externe `dev-shared`).
## Architecture
| Composant | URL publique | Port interne | Conteneur |
|-----------|--------------|--------------|-----------|
| Front (React) | https://datasentinel.nfteam.ovh | 8086 | `datasentinel-front` |
| API (FastAPI) | https://datasentinel-api.nfteam.ovh | 8001 | `datasentinel-api` |
| Base | — (réseau `dev-shared`) | 1433 | `dev-mssql` (partagé) |
Reverse proxy : Nginx Proxy Manager (HTTPS Let's Encrypt). CI : Gitea Actions →
images poussées au registre `git.nfteam.ovh`, déployées par `docker compose` / Watchtower.
## Déploiement / mise à jour
Un push sur `main` (API ou Front) déclenche la CI (build → tests → push image).
Récupérer la dernière image et redéployer :
```bash
cd ~/Documents/homelab/dev/datasentinel
docker compose pull && docker compose up -d
```
Watchtower met aussi à jour automatiquement (~5 min).
## Diagnostic incidents
| Symptôme | Diagnostic | Résolution |
|----------|-----------|------------|
| Front KO | `curl -I https://datasentinel.nfteam.ovh` ; `docker logs datasentinel-front` | `docker compose restart datasentinel-front` |
| API 5xx | `curl https://datasentinel-api.nfteam.ovh/health` → champ `database` | si `error` → voir BDD ci-dessous ; `docker logs datasentinel-api` |
| BDD injoignable | `docker ps | grep dev-mssql` ; tester la connexion (cf. ci-dessous) | `docker compose -f ~/Documents/homelab/dev/mssql/docker-compose.yml up -d` |
| Token invalide / 401 partout | vérifier `JWT_SECRET` dans `api.env` (ne pas le changer à chaud : invalide les sessions) | re-login |
| Login 429 | rate-limit 5/min atteint | attendre 1 min |
Connexion BDD (admin) :
```bash
SA=$(grep ^SA_PASSWORD= ~/Documents/homelab/dev/mssql/.env | cut -d= -f2-)
docker run --rm --network dev-shared mcr.microsoft.com/mssql-tools \
/opt/mssql-tools/bin/sqlcmd -S dev-mssql -U sa -P "$SA" -d DataSentinel -Q "SELECT COUNT(*) FROM [USER];"
```
## (Re)chargement du schéma
```bash
SA=$(grep ^SA_PASSWORD= ~/Documents/homelab/dev/mssql/.env | cut -d= -f2-)
for f in data_sentinel_init.sql data_sentinel_auth.sql; do
docker run --rm --network dev-shared \
-v ~/Documents/homelab/dev/datasentinel/sql:/sql:ro mcr.microsoft.com/mssql-tools \
/opt/mssql-tools/bin/sqlcmd -S dev-mssql -U sa -P "$SA" -i /sql/$f
done
```
## Sauvegarde / restauration
> ⚠️ Stratégie à finaliser avec le disque de sauvegarde dédié du homelab.
Sauvegarde logique recommandée (quotidienne) :
```bash
SA=$(grep ^SA_PASSWORD= ~/Documents/homelab/dev/mssql/.env | cut -d= -f2-)
docker exec dev-mssql /opt/mssql-tools*/bin/sqlcmd -S localhost -U sa -P "$SA" \
-Q "BACKUP DATABASE DataSentinel TO DISK='/var/opt/mssql/backup/DataSentinel.bak' WITH FORMAT, INIT, COMPRESSION"
```
(monter un volume `/var/opt/mssql/backup` vers le disque de sauvegarde). Le volume
Docker `mssql_data` contient les fichiers de la base. RTO visé < 4h, RPO < 24h.
## Rollback
Redéployer une image précise par son tag SHA (au lieu de `latest`) :
```bash
# dans dev/datasentinel/docker-compose.yml : image: .../datasentinel-api:<sha>
docker compose up -d datasentinel-api
```
## Contacts
- Hébergement / infra : administrateur homelab (neckfire).
- Application / code : A. Coyaud (auteur, dépôts GitHub).
+54
View File
@@ -0,0 +1,54 @@
# ============================================================
# auth.py — Authentification JWT + hachage bcrypt
# Data Sentinel
# ============================================================
from datetime import datetime, timedelta, timezone
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from passlib.context import CryptContext
from config import Config
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def hash_password(plain: str) -> str:
return pwd_context.hash(plain)
def create_access_token(data: dict) -> str:
payload = data.copy()
payload["exp"] = datetime.now(timezone.utc) + timedelta(minutes=Config.TOKEN_EXPIRE_MINUTES)
return jwt.encode(payload, Config.SECRET_KEY, algorithm=Config.ALGORITHM)
def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
"""Décode le JWT et retourne l'utilisateur courant, sinon 401."""
creds_exc = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Identifiants invalides",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, Config.SECRET_KEY, algorithms=[Config.ALGORITHM])
username = payload.get("sub")
if not username:
raise creds_exc
return {"username": username, "role": payload.get("role"), "id_user": payload.get("uid")}
except JWTError:
raise creds_exc
def require_admin(user: dict = Depends(get_current_user)) -> dict:
"""Réserve l'accès aux administrateurs (403 sinon)."""
if user.get("role") != "Admin":
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Accès réservé aux administrateurs")
return user
+28 -6
View File
@@ -3,28 +3,50 @@
# Data Sentinel | COYAUD Anthony | 2026 # Data Sentinel | COYAUD Anthony | 2026
# ============================================================ # ============================================================
import os
import pyodbc import pyodbc
from contextlib import contextmanager from contextlib import contextmanager
class Config: def _build_connection_string() -> str:
# Chaîne de connexion SQL Server """
DB_CONNECTION_STRING = ( En déploiement (variables d'environnement présentes), on se connecte en
authentification SQL. Sinon, on garde la connexion Windows locale par défaut
(poste de dev).
"""
server = os.getenv("DB_SERVER")
if server:
driver = os.getenv("DB_DRIVER", "ODBC Driver 18 for SQL Server")
port = os.getenv("DB_PORT", "1433")
return (
f"Driver={{{driver}}};"
f"Server={server},{port};"
f"Database={os.getenv('DB_NAME', 'DataSentinel')};"
f"UID={os.getenv('DB_USER')};"
f"PWD={os.getenv('DB_PASSWORD')};"
"Encrypt=yes;TrustServerCertificate=yes;"
)
return (
"Driver={ODBC Driver 17 for SQL Server};" "Driver={ODBC Driver 17 for SQL Server};"
"Server=LaptopCA\\SQLEXPRESS;" "Server=LaptopCA\\SQLEXPRESS;"
"Database=DataSentinel;" "Database=DataSentinel;"
"Trusted_Connection=yes;" "Trusted_Connection=yes;"
) )
class Config:
# Chaîne de connexion SQL Server (env en prod, Windows en local)
DB_CONNECTION_STRING = _build_connection_string()
# Paramètres API # Paramètres API
API_TITLE = "Data Sentinel API" API_TITLE = "Data Sentinel API"
API_VERSION = "1.0.0" API_VERSION = "1.0.0"
API_DESCRIPTION = "API de monitoring de la qualité des données — XEFI" API_DESCRIPTION = "API de monitoring de la qualité des données — XEFI"
# Sécurité JWT # Sécurité JWT
SECRET_KEY = "data-sentinel-secret-change-in-prod" SECRET_KEY = os.getenv("JWT_SECRET", "data-sentinel-secret-change-in-prod")
ALGORITHM = "HS256" ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")
TOKEN_EXPIRE_MINUTES = 60 TOKEN_EXPIRE_MINUTES = int(os.getenv("JWT_EXPIRE_MINUTES", "60"))
# ---------------------------------------------------------------- # ----------------------------------------------------------------
File diff suppressed because one or more lines are too long
+264 -20
View File
@@ -7,12 +7,32 @@
# Swagger : http://localhost:8000/docs # Swagger : http://localhost:8000/docs
# ============================================================ # ============================================================
from fastapi import FastAPI, HTTPException, Query import os
from fastapi import FastAPI, HTTPException, Query, APIRouter, Depends, Request
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional from typing import Optional
from datetime import date from datetime import date
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from config import Config, get_cursor from config import Config, get_cursor
from auth import (
verify_password, hash_password, create_access_token,
get_current_user, require_admin,
)
# Origines autorisées : depuis CORS_ORIGINS (séparées par des virgules) en prod,
# localhost par défaut en dev.
CORS_ORIGINS = [
o.strip()
for o in os.getenv("CORS_ORIGINS", "http://localhost:5173,http://localhost:3000").split(",")
if o.strip()
]
# ============================================================ # ============================================================
# Initialisation # Initialisation
@@ -24,13 +44,35 @@ app = FastAPI(
description = Config.API_DESCRIPTION, description = Config.API_DESCRIPTION,
) )
# Limiteur de débit (anti brute-force sur /auth/login).
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
allow_origins = ["http://localhost:5173", "http://localhost:3000"], allow_origins = CORS_ORIGINS,
allow_methods = ["GET"], allow_methods = ["GET", "POST", "PUT", "DELETE", "OPTIONS"],
allow_headers = ["*"], allow_headers = ["*"],
allow_credentials = True,
) )
# En-têtes de sécurité sur toutes les réponses.
@app.middleware("http")
async def security_headers(request: Request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response
# Toutes les routes "données" passent par ce router, protégé par JWT.
# /health et /auth/* restent publics (déclarés sur `app`).
router = APIRouter(dependencies=[Depends(get_current_user)])
# ============================================================ # ============================================================
# Mapping id_monito → table SQL dédiée # Mapping id_monito → table SQL dédiée
# Pour ajouter un monitoring : ajouter une entrée ici. # Pour ajouter un monitoring : ajouter une entrée ici.
@@ -102,7 +144,7 @@ def serialize_row(row: dict) -> dict:
# RÉFÉRENTIELS # RÉFÉRENTIELS
# ============================================================ # ============================================================
@app.get("/categories", tags=["Référentiels"]) @router.get("/categories", tags=["Référentiels"])
def get_categories(): def get_categories():
"""Toutes les catégories de monitoring.""" """Toutes les catégories de monitoring."""
with get_cursor() as cursor: with get_cursor() as cursor:
@@ -113,7 +155,7 @@ def get_categories():
return rows_to_list(cursor, cursor.fetchall()) return rows_to_list(cursor, cursor.fetchall())
@app.get("/services", tags=["Référentiels"]) @router.get("/services", tags=["Référentiels"])
def get_services(): def get_services():
"""Tous les services.""" """Tous les services."""
with get_cursor() as cursor: with get_cursor() as cursor:
@@ -121,7 +163,7 @@ def get_services():
return rows_to_list(cursor, cursor.fetchall()) return rows_to_list(cursor, cursor.fetchall())
@app.get("/contacts", tags=["Référentiels"]) @router.get("/contacts", tags=["Référentiels"])
def get_contacts( def get_contacts(
id_service: Optional[int] = Query(None, description="Filtrer par service") id_service: Optional[int] = Query(None, description="Filtrer par service")
): ):
@@ -145,7 +187,7 @@ def get_contacts(
# NOMENCLATURE # NOMENCLATURE
# ============================================================ # ============================================================
@app.get("/monitorings", tags=["Monitorings"]) @router.get("/monitorings", tags=["Monitorings"])
def get_monitorings( def get_monitorings(
id_service : Optional[int] = Query(None, description="Filtrer par service"), id_service : Optional[int] = Query(None, description="Filtrer par service"),
id_categorie : Optional[int] = Query(None, description="Filtrer par catégorie"), id_categorie : Optional[int] = Query(None, description="Filtrer par catégorie"),
@@ -172,7 +214,7 @@ def get_monitorings(
return rows_to_list(cursor, cursor.fetchall()) return rows_to_list(cursor, cursor.fetchall())
@app.get("/monitorings/{id_monito}", tags=["Monitorings"]) @router.get("/monitorings/{id_monito}", tags=["Monitorings"])
def get_monitoring_by_id(id_monito: int): def get_monitoring_by_id(id_monito: int):
"""Détail d'un monitoring.""" """Détail d'un monitoring."""
with get_cursor() as cursor: with get_cursor() as cursor:
@@ -192,7 +234,7 @@ def get_monitoring_by_id(id_monito: int):
# DONNÉES DÉTAILLÉES — table dédiée par monitoring # DONNÉES DÉTAILLÉES — table dédiée par monitoring
# ============================================================ # ============================================================
@app.get("/monitorings/{id_monito}/details", tags=["Monitorings"]) @router.get("/monitorings/{id_monito}/details", tags=["Monitorings"])
def get_monitoring_details( def get_monitoring_details(
id_monito : int, id_monito : int,
search : Optional[str] = Query( search : Optional[str] = Query(
@@ -225,7 +267,7 @@ def get_monitoring_details(
return [serialize_row(r) for r in rows] return [serialize_row(r) for r in rows]
@app.get("/monitorings/{id_monito}/count", tags=["Monitorings"]) @router.get("/monitorings/{id_monito}/count", tags=["Monitorings"])
def get_monitoring_count(id_monito: int): def get_monitoring_count(id_monito: int):
"""Nombre d'erreurs dans la table dédiée du monitoring.""" """Nombre d'erreurs dans la table dédiée du monitoring."""
table = get_table_name(id_monito) table = get_table_name(id_monito)
@@ -238,7 +280,7 @@ def get_monitoring_count(id_monito: int):
} }
@app.get("/monitorings/{id_monito}/columns", tags=["Monitorings"]) @router.get("/monitorings/{id_monito}/columns", tags=["Monitorings"])
def get_monitoring_columns(id_monito: int): def get_monitoring_columns(id_monito: int):
""" """
Retourne les colonnes de la table dédiée. Retourne les colonnes de la table dédiée.
@@ -263,7 +305,7 @@ def get_monitoring_columns(id_monito: int):
# DASHBOARD — VUE_CONSO # DASHBOARD — VUE_CONSO
# ============================================================ # ============================================================
@app.get("/dashboard", tags=["Dashboard"]) @router.get("/dashboard", tags=["Dashboard"])
def get_dashboard( def get_dashboard(
service : Optional[str] = Query(None, description="Filtrer par service"), service : Optional[str] = Query(None, description="Filtrer par service"),
categorie : Optional[str] = Query(None, description="Filtrer par catégorie"), categorie : Optional[str] = Query(None, description="Filtrer par catégorie"),
@@ -289,7 +331,34 @@ def get_dashboard(
return rows_to_list(cursor, cursor.fetchall()) return rows_to_list(cursor, cursor.fetchall())
@app.get("/dashboard/summary", tags=["Dashboard"]) @router.get("/dashboard/filtres", tags=["Dashboard"])
def get_dashboard_filtres():
"""
Valeurs de filtre réellement présentes dans VUE_CONSO.
Les référentiels SERVICE / CATEGORIE contiennent des entrées auxquelles
aucun monitoring actif n'est rattaché (ex. « Business Intelligence »).
Les proposer dans les menus déroulants du dashboard mène à un écran vide :
cet endpoint ne renvoie que les valeurs qui ramènent au moins un monitoring.
`combinaisons` permet au frontend de restreindre les catégories proposées
au service sélectionné.
"""
with get_cursor() as cursor:
cursor.execute(
"SELECT DISTINCT service, categorie FROM VUE_CONSO "
"ORDER BY service, categorie"
)
combinaisons = [{"service": r[0], "categorie": r[1]} for r in cursor.fetchall()]
return {
"services" : sorted({c["service"] for c in combinaisons}),
"categories" : sorted({c["categorie"] for c in combinaisons}),
"combinaisons" : combinaisons,
}
@router.get("/dashboard/summary", tags=["Dashboard"])
def get_dashboard_summary(): def get_dashboard_summary():
"""KPI globaux pour les 4 cartes du dashboard.""" """KPI globaux pour les 4 cartes du dashboard."""
with get_cursor() as cursor: with get_cursor() as cursor:
@@ -318,7 +387,7 @@ def get_dashboard_summary():
# HISTORIQUE — TABLE_FINAL # HISTORIQUE — TABLE_FINAL
# ============================================================ # ============================================================
@app.get("/historique", tags=["Historique"]) @router.get("/historique", tags=["Historique"])
def get_historique( def get_historique(
id_monito : Optional[int] = Query(None), id_monito : Optional[int] = Query(None),
date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"), date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"),
@@ -353,7 +422,7 @@ def get_historique(
return rows_to_list(cursor, cursor.fetchall()) return rows_to_list(cursor, cursor.fetchall())
@app.get("/historique/{id_monito}/evolution", tags=["Historique"]) @router.get("/historique/{id_monito}/evolution", tags=["Historique"])
def get_evolution( def get_evolution(
id_monito : int, id_monito : int,
date_debut : Optional[date] = Query(None), date_debut : Optional[date] = Query(None),
@@ -384,7 +453,7 @@ def get_evolution(
return {"id_monito": id_monito, "points": len(data), "evolution": data} return {"id_monito": id_monito, "points": len(data), "evolution": data}
@app.get("/historique/comparaison", tags=["Historique"]) @router.get("/historique/comparaison", tags=["Historique"])
def get_comparaison( def get_comparaison(
date_debut : Optional[date] = Query(None), date_debut : Optional[date] = Query(None),
date_fin : Optional[date] = Query(None), date_fin : Optional[date] = Query(None),
@@ -422,7 +491,7 @@ def get_comparaison(
# SERVICE → regroupement par service (Contrat / Fournisseur) # SERVICE → regroupement par service (Contrat / Fournisseur)
# ============================================================ # ============================================================
@app.get("/evolution/global", tags=["Évolution globale"]) @router.get("/evolution/global", tags=["Évolution globale"])
def get_evolution_global( def get_evolution_global(
date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"), date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"),
date_fin : Optional[date] = Query(None, description="YYYY-MM-DD"), date_fin : Optional[date] = Query(None, description="YYYY-MM-DD"),
@@ -459,7 +528,7 @@ def get_evolution_global(
} }
@app.get("/evolution/par-monitoring", tags=["Évolution globale"]) @router.get("/evolution/par-monitoring", tags=["Évolution globale"])
def get_evolution_par_monitoring( def get_evolution_par_monitoring(
date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"), date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"),
date_fin : Optional[date] = Query(None, description="YYYY-MM-DD"), date_fin : Optional[date] = Query(None, description="YYYY-MM-DD"),
@@ -513,7 +582,7 @@ def get_evolution_par_monitoring(
} }
@app.get("/evolution/par-service", tags=["Évolution globale"]) @router.get("/evolution/par-service", tags=["Évolution globale"])
def get_evolution_par_service( def get_evolution_par_service(
date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"), date_debut : Optional[date] = Query(None, description="YYYY-MM-DD"),
date_fin : Optional[date] = Query(None, description="YYYY-MM-DD"), date_fin : Optional[date] = Query(None, description="YYYY-MM-DD"),
@@ -583,3 +652,178 @@ def health_check():
"version" : Config.API_VERSION, "version" : Config.API_VERSION,
"nb_monitorings" : len(MONITO_TABLES), "nb_monitorings" : len(MONITO_TABLES),
} }
# ============================================================
# AUTHENTIFICATION
# ============================================================
class UserCreate(BaseModel):
username: str
email: str
password: str
role: str
class UserUpdate(BaseModel):
email: Optional[str] = None
role: Optional[str] = None
actif: Optional[bool] = None
class PasswordReset(BaseModel):
password: str
@app.post("/auth/login", tags=["Auth"])
@limiter.limit("5/minute")
def login(request: Request, form: OAuth2PasswordRequestForm = Depends()):
"""Authentifie un utilisateur et retourne un JWT."""
with get_cursor() as cursor:
cursor.execute(
"SELECT id_user, username, password_hash, role, actif "
"FROM [USER] WHERE username = ?", form.username
)
row = cursor.fetchone()
if not row or not row[4] or not verify_password(form.password, row[2]):
raise HTTPException(status_code=401, detail="Identifiants invalides")
ip = request.client.host if request.client else None
cursor.execute("UPDATE [USER] SET last_login = SYSDATETIME() WHERE id_user = ?", row[0])
cursor.execute(
"INSERT INTO JOURNAL_AUDIT (id_user, username, action, ip) VALUES (?, ?, 'LOGIN', ?)",
row[0], row[1], ip
)
token = create_access_token({"sub": row[1], "role": row[3], "uid": row[0]})
return {"access_token": token, "token_type": "bearer",
"user": {"username": row[1], "role": row[3]}}
@app.get("/auth/me", tags=["Auth"])
def me(user: dict = Depends(get_current_user)):
"""Profil de l'utilisateur connecté."""
return user
# ============================================================
# ADMINISTRATION (réservé Admin)
# ============================================================
@app.get("/admin/users", tags=["Admin"])
def list_users(admin: dict = Depends(require_admin)):
with get_cursor() as cursor:
cursor.execute(
"SELECT id_user, username, email, role, actif, created_at, last_login "
"FROM [USER] ORDER BY id_user"
)
return rows_to_list(cursor, cursor.fetchall())
@app.post("/admin/users", tags=["Admin"], status_code=201)
def create_user(body: UserCreate, admin: dict = Depends(require_admin)):
if body.role not in ("Admin", "Superviseur", "Consultant"):
raise HTTPException(status_code=400, detail="Rôle invalide")
with get_cursor() as cursor:
cursor.execute(
"INSERT INTO [USER] (username, email, password_hash, role) VALUES (?, ?, ?, ?)",
body.username, body.email, hash_password(body.password), body.role
)
cursor.execute(
"INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail) VALUES (?, ?, 'CREATE_USER', ?)",
admin["id_user"], admin["username"], body.username
)
return {"status": "created", "username": body.username}
@app.put("/admin/users/{id_user}", tags=["Admin"])
def update_user(id_user: int, body: UserUpdate, admin: dict = Depends(require_admin)):
sets, params = [], []
if body.email is not None:
sets.append("email = ?"); params.append(body.email)
if body.role is not None:
if body.role not in ("Admin", "Superviseur", "Consultant"):
raise HTTPException(status_code=400, detail="Rôle invalide")
sets.append("role = ?"); params.append(body.role)
if body.actif is not None:
sets.append("actif = ?"); params.append(1 if body.actif else 0)
if not sets:
raise HTTPException(status_code=400, detail="Aucun champ à modifier")
params.append(id_user)
with get_cursor() as cursor:
cursor.execute(f"UPDATE [USER] SET {', '.join(sets)} WHERE id_user = ?", *params)
cursor.execute(
"INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail) VALUES (?, ?, 'UPDATE_USER', ?)",
admin["id_user"], admin["username"], str(id_user)
)
return {"status": "updated", "id_user": id_user}
@app.delete("/admin/users/{id_user}", tags=["Admin"])
def delete_user(id_user: int, admin: dict = Depends(require_admin)):
"""Suppression douce (actif = 0)."""
with get_cursor() as cursor:
cursor.execute("UPDATE [USER] SET actif = 0 WHERE id_user = ?", id_user)
cursor.execute(
"INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail) VALUES (?, ?, 'DELETE_USER', ?)",
admin["id_user"], admin["username"], str(id_user)
)
return {"status": "deactivated", "id_user": id_user}
@app.post("/admin/users/{id_user}/reset-password", tags=["Admin"])
def reset_password(id_user: int, body: PasswordReset, admin: dict = Depends(require_admin)):
with get_cursor() as cursor:
cursor.execute("UPDATE [USER] SET password_hash = ? WHERE id_user = ?",
hash_password(body.password), id_user)
cursor.execute(
"INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail) VALUES (?, ?, 'RESET_PASSWORD', ?)",
admin["id_user"], admin["username"], str(id_user)
)
return {"status": "password_reset", "id_user": id_user}
@app.get("/admin/journal", tags=["Admin"])
def get_journal(limit: int = 200, admin: dict = Depends(require_admin)):
with get_cursor() as cursor:
cursor.execute(
"SELECT TOP (?) date_action, username, action, detail, ip "
"FROM JOURNAL_AUDIT ORDER BY date_action DESC", limit
)
return rows_to_list(cursor, cursor.fetchall())
# ============================================================
# RGPD — droits de la personne
# ============================================================
@app.get("/me/data-export", tags=["RGPD"])
def export_my_data(user: dict = Depends(get_current_user)):
"""Droit à la portabilité (art. 20) : export des données de l'utilisateur."""
with get_cursor() as cursor:
cursor.execute(
"SELECT id_user, username, email, role, actif, created_at, last_login "
"FROM [USER] WHERE username = ?", user["username"]
)
u = cursor.fetchone()
profile = row_to_dict(cursor, u) if u else {}
cursor.execute(
"SELECT date_action, action, detail, ip FROM JOURNAL_AUDIT "
"WHERE username = ? ORDER BY date_action DESC", user["username"]
)
journal = rows_to_list(cursor, cursor.fetchall())
return {"profile": profile, "journal": journal}
@app.delete("/me", tags=["RGPD"])
def delete_my_account(user: dict = Depends(get_current_user)):
"""Droit à l'oubli (art. 17) : désactivation + anonymisation."""
with get_cursor() as cursor:
cursor.execute(
"UPDATE [USER] SET actif = 0, "
"username = CONCAT('deleted_', id_user), "
"email = CONCAT('deleted_', id_user, '@deleted.local') "
"WHERE username = ?", user["username"]
)
cursor.execute("UPDATE JOURNAL_AUDIT SET username = NULL WHERE username = ?", user["username"])
return {"status": "account_deleted"}
# Enregistre les routes "données" protégées par JWT.
app.include_router(router)
+5
View File
@@ -2,3 +2,8 @@ fastapi==0.136.0
uvicorn==0.46.0 uvicorn==0.46.0
pyodbc==5.3.0 pyodbc==5.3.0
requests==2.33.1 requests==2.33.1
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
bcrypt==4.0.1
python-multipart==0.0.9
slowapi==0.1.9
+88
View File
@@ -0,0 +1,88 @@
-- ============================================================
-- DATA SENTINEL — Création / réinitialisation du compte « Juré »
-- Auteur : COYAUD Anthony
-- Version : 1.1 — Août 2026
--
-- Compte Admin destiné aux personnes qui consultent l'application
-- (évaluateurs, jury, démonstration).
--
-- identifiant : Juré
-- mail : JURE@NEXA.com
-- mot de passe : 123456
-- rôle : Admin
--
-- Script idempotent : il peut être rejoué sur une base déjà déployée
-- sans dupliquer le compte ni toucher aux autres utilisateurs.
-- Prérequis : data_sentinel_auth.sql déjà exécuté ([USER] + JOURNAL_AUDIT).
--
-- Pas d'instruction USE : la base cible vient du paramètre -d de la
-- connexion, et le garde-fou ci-dessous interrompt le script si ce
-- n'est pas la bonne. Tout tient dans un seul lot (aucun GO), sans
-- quoi le RETURN du garde-fou n'empêcherait pas la suite de tourner.
--
-- Exécution :
-- sqlcmd -S <serveur> -d <base> -U <user> -P <mdp> -C -b -f 65001 \
-- -i sql/create_admin_jure.sql
-- ============================================================
SET NOCOUNT ON;
SET XACT_ABORT ON;
DECLARE @base NVARCHAR(128) = DB_NAME();
IF OBJECT_ID('[USER]', 'U') IS NULL OR OBJECT_ID('JOURNAL_AUDIT', 'U') IS NULL
BEGIN
RAISERROR(
'ARRET : base incorrecte (%s). [USER] ou JOURNAL_AUDIT est introuvable. Reconnectez-vous sur la base Data Sentinel (option -d).',
16, 1, @base);
RETURN;
END
-- Hachage bcrypt (coût 12) du mot de passe de démonstration.
-- Généré avec passlib.CryptContext(schemes=["bcrypt"]) — même librairie que auth.py.
DECLARE @username NVARCHAR(100) = N'Juré';
DECLARE @email NVARCHAR(200) = N'JURE@NEXA.com';
DECLARE @password_hash NVARCHAR(255) = N'$2b$12$nnWMOCjSiUq4rAyPGbC3W.HGdsvWKu9kEfLst6zBIEIcsgP97W94m';
BEGIN TRY
BEGIN TRANSACTION;
IF EXISTS (SELECT 1 FROM [USER] WHERE username = @username)
BEGIN
UPDATE [USER]
SET email = @email,
password_hash = @password_hash,
role = N'Admin',
actif = 1
WHERE username = @username;
INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail)
SELECT id_user, username, 'RESET_PASSWORD', 'Compte de visite reinitialise (script SQL)'
FROM [USER] WHERE username = @username;
PRINT 'Compte "Jure" deja present -> mis a jour.';
END
ELSE
BEGIN
INSERT INTO [USER] (username, email, password_hash, role, actif)
VALUES (@username, @email, @password_hash, N'Admin', 1);
INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail)
SELECT id_user, username, 'CREATE_USER', 'Compte de visite cree (script SQL)'
FROM [USER] WHERE username = @username;
PRINT 'Compte "Jure" cree.';
END
COMMIT TRANSACTION;
END TRY
BEGIN CATCH
IF @@TRANCOUNT > 0 ROLLBACK TRANSACTION;
PRINT 'ECHEC : ' + ERROR_MESSAGE();
THROW;
END CATCH
-- Vérification
SELECT id_user, username, email, role, actif, created_at, last_login
FROM [USER]
WHERE username = N'Juré';
+105
View File
@@ -0,0 +1,105 @@
-- ============================================================
-- DATA SENTINEL — Script d'authentification SQL Server
-- Tables [USER] + JOURNAL_AUDIT + comptes de démonstration
-- Auteur : COYAUD Anthony
-- Version : 1.1 — Juin 2026
-- À exécuter APRÈS data_sentinel_init.sql (même base DataSentinel)
-- ============================================================
--
-- Ce script crée la couche de sécurité référencée par l'API
-- (main.py / auth.py) :
-- - [USER] : comptes applicatifs (RBAC : Admin / Superviseur / Consultant)
-- - JOURNAL_AUDIT : journal d'audit (connexions + actions d'administration)
--
-- Les mots de passe sont stockés hachés en bcrypt (passlib[bcrypt]).
-- Les hachages ci-dessous correspondent aux comptes de démonstration
-- documentés dans le README (à régénérer en production).
-- ============================================================
USE DataSentinel;
GO
-- ============================================================
-- SUPPRESSION (ordre inverse des dépendances)
-- ============================================================
IF OBJECT_ID('JOURNAL_AUDIT', 'U') IS NOT NULL DROP TABLE JOURNAL_AUDIT;
IF OBJECT_ID('[USER]', 'U') IS NOT NULL DROP TABLE [USER];
GO
-- ============================================================
-- 1. TABLE [USER] — comptes applicatifs
-- ([USER] entre crochets car USER est un mot réservé SQL Server)
-- ============================================================
CREATE TABLE [USER] (
id_user INT NOT NULL IDENTITY(1,1),
username NVARCHAR(100) NOT NULL,
email NVARCHAR(200) NULL,
password_hash NVARCHAR(255) NOT NULL, -- bcrypt
role NVARCHAR(20) NOT NULL, -- Admin | Superviseur | Consultant
actif BIT NOT NULL DEFAULT 1,
created_at DATETIME2 NOT NULL DEFAULT SYSDATETIME(),
last_login DATETIME2 NULL,
CONSTRAINT PK_USER PRIMARY KEY (id_user),
CONSTRAINT UQ_USER_NAME UNIQUE (username),
CONSTRAINT CK_USER_ROLE CHECK (role IN ('Admin', 'Superviseur', 'Consultant'))
);
GO
CREATE INDEX IX_USER_USERNAME ON [USER] (username);
GO
-- ============================================================
-- 2. TABLE JOURNAL_AUDIT — journal d'audit applicatif
-- Alimenté à chaque login + action d'administration.
-- username/id_user nullables (anonymisation RGPD : droit à l'oubli).
-- ============================================================
CREATE TABLE JOURNAL_AUDIT (
id_audit INT NOT NULL IDENTITY(1,1),
id_user INT NULL,
username NVARCHAR(100) NULL,
action NVARCHAR(50) NOT NULL, -- LOGIN | CREATE_USER | UPDATE_USER | DELETE_USER | RESET_PASSWORD
detail NVARCHAR(500) NULL,
ip NVARCHAR(50) NULL,
date_action DATETIME2 NOT NULL DEFAULT SYSDATETIME(),
CONSTRAINT PK_JOURNAL_AUDIT PRIMARY KEY (id_audit)
);
GO
CREATE INDEX IX_AUDIT_DATE ON JOURNAL_AUDIT (date_action DESC);
GO
-- ============================================================
-- 3. COMPTES DE DÉMONSTRATION
-- ⚠️ Mots de passe en clair (à usage de recette / soutenance) :
-- Juré / 123456 → rôle Admin (compte de visite / évaluation)
-- admin / Admin2026! → rôle Admin
-- superviseur / Super2026! → rôle Superviseur
-- consultant / Conseil2026! → rôle Consultant
-- Les hachages bcrypt ci-dessous sont fonctionnels tels quels.
-- ============================================================
INSERT INTO [USER] (username, email, password_hash, role, actif) VALUES
(N'Juré', 'JURE@NEXA.com',
'$2b$12$nnWMOCjSiUq4rAyPGbC3W.HGdsvWKu9kEfLst6zBIEIcsgP97W94m', 'Admin', 1),
('admin', 'admin@xefi-fictif.fr',
'$2b$12$XJobnqM0cHwekv7UYWXUfuxVjMKLZI6VLzpDvRzDdLJgnYHLRNRCy', 'Admin', 1),
('superviseur', 'superviseur@xefi-fictif.fr',
'$2b$12$gMVg2j6wngZZ2KXH2jF6HuvlcAkGIJsbZ05JEQdDYNnIu7fDxOXri', 'Superviseur', 1),
('consultant', 'consultant@xefi-fictif.fr',
'$2b$12$s4Gl3UDq56HOw/GdjeHCrOFjXN/QrV1OEj3wmHhf3T8mFVfijJkty', 'Consultant', 1);
GO
-- Entrée d'audit initiale (création du jeu de comptes)
INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail)
SELECT id_user, username, 'CREATE_USER', 'Compte de démonstration (seed)'
FROM [USER];
GO
-- ============================================================
-- 4. VÉRIFICATIONS
-- ============================================================
SELECT 'USER' AS [Table], COUNT(*) AS [Lignes] FROM [USER]
UNION ALL SELECT 'JOURNAL_AUDIT', COUNT(*) FROM JOURNAL_AUDIT;
GO
SELECT id_user, username, email, role, actif, created_at FROM [USER] ORDER BY id_user;
GO
View File
+53
View File
@@ -0,0 +1,53 @@
"""Fixtures de test : client FastAPI + curseur SQL simulé (aucune vraie BDD)."""
import os
from contextlib import contextmanager
os.environ.setdefault("JWT_SECRET", "test-secret")
import pytest
from fastapi.testclient import TestClient
import main
from auth import create_access_token
class FakeCursor:
"""Curseur pyodbc simulé : on règle description / rows / one par test."""
def __init__(self):
self.description = []
self._rows = []
self._one = None
self.executed = []
def execute(self, query, *args):
self.executed.append((query, args))
return self
def fetchall(self):
return self._rows
def fetchone(self):
return self._one
@pytest.fixture
def cur(monkeypatch):
c = FakeCursor()
@contextmanager
def fake_get_cursor():
yield c
monkeypatch.setattr(main, "get_cursor", fake_get_cursor)
return c
@pytest.fixture
def client():
return TestClient(main.app)
@pytest.fixture
def auth_headers():
token = create_access_token({"sub": "admin", "role": "Admin", "uid": 1})
return {"Authorization": f"Bearer {token}"}
+72
View File
@@ -0,0 +1,72 @@
"""Tests d'endpoints (curseur SQL simulé via la fixture `cur`)."""
from auth import hash_password, create_access_token
def test_health_ok(client, cur):
cur._one = (1,)
r = client.get("/health")
assert r.status_code == 200
assert r.json()["api"] == "ok"
def test_categories_requires_auth(client):
# sans token -> 401
assert client.get("/categories").status_code == 401
def test_services_returns_list(client, cur, auth_headers):
cur.description = [("id_service",), ("nom_service",)]
cur._rows = [(1, "Contrat"), (2, "Fournisseur")]
r = client.get("/services", headers=auth_headers)
assert r.status_code == 200
assert {"id_service": 1, "nom_service": "Contrat"} in r.json()
def test_dashboard_filtres_ne_renvoie_que_les_valeurs_utilisees(client, cur, auth_headers):
# VUE_CONSO ne contient que Contrat / Fournisseur : les services du
# référentiel sans monitoring rattaché ne doivent pas remonter.
cur.description = [("service",), ("categorie",)]
cur._rows = [
("Contrat", "DOM-TOM"),
("Contrat", "Tiers-payeurs"),
("Fournisseur", "Contreparties"),
]
r = client.get("/dashboard/filtres", headers=auth_headers)
assert r.status_code == 200
body = r.json()
assert body["services"] == ["Contrat", "Fournisseur"]
assert "Business Intelligence" not in body["services"]
assert body["categories"] == ["Contreparties", "DOM-TOM", "Tiers-payeurs"]
assert {"service": "Contrat", "categorie": "DOM-TOM"} in body["combinaisons"]
def test_auth_login_success(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
body = r.json()
assert body["token_type"] == "bearer"
assert body["user"]["role"] == "Admin"
def test_auth_login_wrong_password(client, cur):
cur._one = (1, "admin", hash_password("Admin2026!"), "Admin", 1)
r = client.post("/auth/login", data={"username": "admin", "password": "WRONG"})
assert r.status_code == 401
def test_auth_me(client, auth_headers):
r = client.get("/auth/me", headers=auth_headers)
assert r.status_code == 200
assert r.json()["username"] == "admin"
def test_monitoring_unknown_returns_404(client, cur, auth_headers):
r = client.get("/monitorings/999/details", headers=auth_headers)
assert r.status_code == 404
def test_admin_forbidden_for_consultant(client):
token = create_access_token({"sub": "bob", "role": "Consultant", "uid": 2})
r = client.get("/admin/users", headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 403