transfer from private server/git server

This commit is contained in:
M1n-0
2026-08-21 09:12:02 +02:00
parent 45c8982178
commit 6ecb014238
75 changed files with 4344 additions and 274 deletions

22
web/.env.example Normal file
View File

@@ -0,0 +1,22 @@
# Copier ce fichier en .env et renseigner les vraies valeurs.
# .env est exclu de Git (voir .gitignore) — mitigation du risque
# "fuite du CLIENT_SECRET OIDC" (tableau OWASP du dossier technique).
SECRET_KEY=change-me-with-a-random-value
FLASK_ENV=production
FLASK_DEBUG=false
# Pseudonymisation RGPD (droit à l'effacement) — voir routes/rgpd.py.
# Si laissé vide, SECRET_KEY est utilisé comme pepper.
RGPD_ANON_PEPPER=change-me-with-another-random-value
DB_HOST=mariadb
DB_PORT=3306
DB_USER=flaskuser
DB_PASSWORD=change-me
DB_NAME=gesthub
OIDC_CLIENT_ID=flask-app
OIDC_CLIENT_SECRET=change-me
OIDC_ISSUER=https://keycloak.ninolbt.com/realms/gesthub
PUBLIC_BASE_URL=https://dashboard.ninolbt.com

View File

@@ -1,8 +1,21 @@
FROM python:3.11
# syntax=docker/dockerfile:1
FROM python:3.11-slim
WORKDIR /web
COPY . /web
# libmagic1 : bibliothèque système requise par python-magic (détection MIME
# par analyse binaire, étape 3 du module fichiers).
RUN apt-get update \
&& apt-get install -y --no-install-recommends libmagic1 \
&& rm -rf /var/lib/apt/lists/*
RUN pip install flask flask_sqlalchemy pymysql authlib requests
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
CMD ["python", "app.py"]
COPY . .
EXPOSE 5000
# Gunicorn multi-worker en production (Bloc 1 - C6 : workers = 2*CPU + 1).
# GUNICORN_CMD_ARGS permet de surcharger la config sans reconstruire l'image.
CMD ["gunicorn", "--workers=4", "--worker-class=gthread", "--threads=2", \
"--bind=0.0.0.0:5000", "app:app"]

View File

@@ -1,152 +1,87 @@
import os
import uuid
import json
# J'ai ajouté 'request' aux imports
from flask import Flask, redirect, url_for, jsonify, session, render_template, request
from flask_sqlalchemy import SQLAlchemy
"""
app.py — Point d'entrée de l'application GestHub.
Ce fichier ne fait qu'assembler l'application (factory pattern) : création
de l'app Flask, enregistrement des blueprints (couche routes/), du client
OAuth Keycloak, et des gestionnaires d'erreurs globaux. Aucune logique
métier ni aucun accès base de données n'a lieu ici (cf. architecture
5 couches — Bloc 1, C3).
"""
import logging
from authlib.integrations.flask_client import OAuth
from flask import Flask, jsonify, render_template
app = Flask(__name__)
ANNOUNCE_FILE = os.path.join(os.path.dirname(__file__), "annonces.json")
# Ta config DB actuelle
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://flaskuser:flaskpass@mariadb/flaskdb'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.secret_key = os.environ.get("SECRET_KEY", "dev-key")
from config import Config
from db import init_db
db = SQLAlchemy(app)
# --- MODELE DE DONNEES POUR LES BLOCS ---
class Block(db.Model):
id = db.Column(db.Integer, primary_key=True)
block_type = db.Column(db.String(50)) # ex: 'iframe', 'buttons', 'html'
column_name = db.Column(db.String(20)) # ex: 'left', 'center', 'right'
position = db.Column(db.Integer) # pour l'ordre (0, 1, 2...)
data = db.Column(db.Text) # Contenu JSON (url, titre, etc.)
def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
app.secret_key = config_class.SECRET_KEY
def to_dict(self):
return {
"id": self.id,
"type": self.block_type,
"column": self.column_name,
"position": self.position,
"data": json.loads(self.data) if self.data else {}
}
logging.basicConfig(level=logging.INFO)
# Création des tables si elles n'existent pas
with app.app_context():
db.create_all()
# --- Base de données (couche infrastructure) ---
db = init_db(config_class)
app.config["DB"] = db
# ... (Ici, garde ta configuration OAUTH et tes routes Login/Logout/Auth inchangées) ...
# ... (Garde aussi tes fonctions load_announces, save_announces etc.) ...
# --- OAuth / OIDC Keycloak ---
oauth = OAuth(app)
oauth = OAuth(app)
keycloak = oauth.register(
name='keycloak',
client_id='flask-app',
client_secret='T5G5jzCBiphnBNh9uuj0f6YNc9HrP8r4',
server_metadata_url='https://keycloak.ninolbt.com/realms/gesthub/.well-known/openid-configuration',
client_kwargs={'scope': 'openid profile email'}
)
# --- Blueprints (couche routes/) ---
from routes.auth import init_auth_routes
from routes.dashboard import bp as dashboard_bp
from routes.annonces import bp as annonces_bp
from routes.fichiers import bp as fichiers_bp
from routes.evenements import bp as evenements_bp
from routes.rgpd import bp as rgpd_bp
@app.route('/')
def index():
user = session.get('user')
if user:
return render_template('view/index.html', user=user)
return redirect(url_for('login'))
app.register_blueprint(init_auth_routes(oauth))
app.register_blueprint(dashboard_bp)
app.register_blueprint(annonces_bp)
app.register_blueprint(fichiers_bp)
app.register_blueprint(evenements_bp)
app.register_blueprint(rgpd_bp)
@app.route('/login')
def login():
nonce = uuid.uuid4().hex
session['nonce'] = nonce
redirect_uri = url_for('auth', _external=True, _scheme='https')
return keycloak.authorize_redirect(redirect_uri, nonce=nonce)
register_error_handlers(app)
return app
@app.route('/auth')
def auth():
token = keycloak.authorize_access_token()
nonce = session.pop('nonce', None)
userinfo = keycloak.parse_id_token(token, nonce=nonce)
session['user'] = userinfo
session["id_token"] = token.get("id_token")
return redirect('/')
@app.route("/logout")
def logout():
id_token = session.get("id_token")
session.clear()
return redirect(
f"https://keycloak.ninolbt.com/realms/gesthub/protocol/openid-connect/logout"
f"?post_logout_redirect_uri=https://dashboard.ninolbt.com"
f"&id_token_hint={id_token}"
)
def register_error_handlers(app):
"""Gestionnaires d'erreurs globaux (Bloc 1 - C5) : réponses uniformisées,
aucune fuite d'information système sensible (stack trace, chemin serveur...).
"""
# --- API LAYOUT (Gestion des Blocs) ---
@app.errorhandler(403)
def forbidden(_error):
if _wants_json():
return jsonify({"error": "forbidden"}), 403
return render_template("errors/403.html"), 403
@app.route('/api/layout', methods=['GET'])
def get_layout():
# Récupère tous les blocs triés par position
blocks = Block.query.order_by(Block.position).all()
return jsonify([b.to_dict() for b in blocks])
@app.errorhandler(404)
def not_found(_error):
if _wants_json():
return jsonify({"error": "not_found"}), 404
return render_template("errors/404.html"), 404
@app.route('/api/layout/save', methods=['POST'])
def save_layout():
# Sauvegarde l'ordre et la colonne après un drag & drop
user = session.get("user")
if not user or "/admin" not in user.get("groups", []):
return jsonify({"error": "Unauthorized"}), 403
@app.errorhandler(500)
def server_error(error):
app.logger.exception("Erreur interne non gérée: %s", error)
if _wants_json():
return jsonify({"error": "internal_error"}), 500
return render_template("errors/500.html"), 500
layout_data = request.json # Liste de {id, column, position}
for item in layout_data:
block = Block.query.get(item['id'])
if block:
block.column_name = item['column']
block.position = item['position']
db.session.commit()
return jsonify({"status": "saved"})
@app.route('/api/block/add', methods=['POST'])
def add_block():
user = session.get("user")
if not user or "/admin" not in user.get("groups", []):
return jsonify({"error": "Unauthorized"}), 403
def _wants_json():
from flask import request
data = request.json
new_block = Block(
block_type=data.get('type'),
column_name=data.get('column', 'center'),
position=99, # Ajoute à la fin par défaut
data=json.dumps(data.get('data', {}))
)
db.session.add(new_block)
db.session.commit()
return jsonify(new_block.to_dict())
return request.path.startswith("/api/")
@app.route('/api/block/<int:block_id>', methods=['DELETE'])
def delete_block(block_id):
user = session.get("user")
if not user or "/admin" not in user.get("groups", []):
return jsonify({"error": "Unauthorized"}), 403
block = Block.query.get(block_id)
if block:
db.session.delete(block)
db.session.commit()
return jsonify({"status": "deleted"})
return jsonify({"error": "not found"}), 404
app = create_app()
# --- API ANNONCES (Tes routes existantes) ---
# ... (Colle ici tes routes /api/annonces existantes, elles sont très bien) ...
# ... (N'oublie pas la route /api/is_admin) ...
@app.route("/api/is_admin")
def is_admin():
user = session.get("user")
# Sécurité : si pas de user, renvoie false
if not user: return jsonify({"admin": False})
return jsonify({"admin": "/admin" in user.get("groups", [])})
if __name__ == '__main__':
app.run(host='0.0.0.0', debug=True)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=Config.DEBUG)

69
web/config.py Normal file
View File

@@ -0,0 +1,69 @@
"""
config.py — Configuration centralisée de l'application.
Toutes les valeurs sensibles (secrets, mots de passe, client_secret OIDC)
sont lues depuis les variables d'environnement (fichier .env en local,
variables d'environnement Docker en production). Aucun secret n'est en dur
dans le code source, conformément à l'analyse de risques DevSecOps
(Bloc 1 - C8 : "fuite du CLIENT_SECRET OIDC").
"""
import os
class Config:
# --- Application ---
SECRET_KEY = os.environ.get("SECRET_KEY", "dev-key-change-me")
DEBUG = os.environ.get("FLASK_DEBUG", "false").lower() == "true"
ENV = os.environ.get("FLASK_ENV", "production")
# --- Base de données (MariaDB via PyMySQL) ---
DB_HOST = os.environ.get("DB_HOST", "mariadb")
DB_PORT = int(os.environ.get("DB_PORT", "3306"))
DB_USER = os.environ.get("DB_USER", "flaskuser")
DB_PASSWORD = os.environ.get("DB_PASSWORD", "flaskpass")
DB_NAME = os.environ.get("DB_NAME", "gesthub")
# --- Authentification Keycloak (OIDC) ---
OIDC_CLIENT_ID = os.environ.get("OIDC_CLIENT_ID", "flask-app")
OIDC_CLIENT_SECRET = os.environ.get("OIDC_CLIENT_SECRET", "")
OIDC_ISSUER = os.environ.get(
"OIDC_ISSUER", "https://keycloak.ninolbt.com/realms/gesthub"
)
OIDC_DISCOVERY_URL = f"{OIDC_ISSUER}/.well-known/openid-configuration"
OIDC_LOGOUT_URL = f"{OIDC_ISSUER}/protocol/openid-connect/logout"
PUBLIC_BASE_URL = os.environ.get("PUBLIC_BASE_URL", "https://dashboard.ninolbt.com")
# --- Rôles applicatifs (mappés sur les groupes Keycloak) ---
ADMIN_GROUP = "/admin"
# --- RGPD (droit à l'effacement, art. 17) ---
# Secret utilisé pour pseudonymiser irréversiblement un `sub` OIDC lors
# d'une anonymisation (HMAC-SHA256). Distinct de SECRET_KEY par défaut ;
# si absent, retombe sur SECRET_KEY (voir routes/rgpd.py).
RGPD_ANON_PEPPER = os.environ.get("RGPD_ANON_PEPPER", "")
# --- Module fichiers ---
UPLOAD_DIR = os.environ.get(
"UPLOAD_DIR", os.path.join(os.path.dirname(__file__), "static", "assets", "uploads")
)
MAX_UPLOAD_SIZE_BYTES = int(os.environ.get("MAX_UPLOAD_SIZE_BYTES", 20 * 1024 * 1024)) # 20 Mo
ALLOWED_EXTENSIONS = {"pdf", "png", "jpg", "jpeg", "gif", "txt", "docx", "xlsx", "zip"}
ALLOWED_MIME_TYPES = {
"application/pdf", "image/png", "image/jpeg", "image/gif", "text/plain",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"application/zip", "application/x-zip-compressed",
}
class TestConfig(Config):
"""Configuration utilisée par la suite de tests pytest (Bloc 1 - C5, Bloc 3 - C7).
Pointe vers une base MariaDB de test dédiée (voir tests/conftest.py) afin de ne
jamais exécuter les tests contre les données de développement ou de production.
"""
TESTING = True
DB_NAME = os.environ.get("TEST_DB_NAME", "gesthub_test")
SECRET_KEY = "test-secret-key"

0
web/cookie.txt Normal file
View File

93
web/db.py Normal file
View File

@@ -0,0 +1,93 @@
"""
db.py — Module centralisé d'accès à la base de données MariaDB.
Ce module est le SEUL point de contact avec PyMySQL dans l'application.
Il expose un gestionnaire de connexion/curseur réutilisable par tous les
modèles de la couche models/, avec :
- gestion des erreurs et rollback automatique,
- reconnexion en cas de perte de connexion (ping avant chaque emprunt),
- curseurs en mode dictionnaire (résultats accessibles par nom de colonne).
Toutes les requêtes exécutées via ce module DOIVENT utiliser des paramètres
préparés (placeholders %s) — jamais de concaténation de chaînes — afin
d'éliminer tout risque d'injection SQL (Bloc 1 - C4, C8 ; OWASP A03:2021).
"""
import logging
from contextlib import contextmanager
import pymysql
import pymysql.cursors
logger = logging.getLogger("gesthub.db")
class Database:
"""Petite couche d'abstraction au-dessus de PyMySQL (pas d'ORM)."""
def __init__(self, host, port, user, password, database):
self.host = host
self.port = port
self.user = user
self.password = password
self.database = database
def _connect(self):
return pymysql.connect(
host=self.host,
port=self.port,
user=self.user,
password=self.password,
database=self.database,
cursorclass=pymysql.cursors.DictCursor,
autocommit=False,
charset="utf8mb4",
)
@contextmanager
def cursor(self, commit=False):
"""Context manager fournissant un curseur et gérant transaction/erreurs.
Usage :
with db.cursor(commit=True) as cur:
cur.execute("INSERT INTO annonces (titre) VALUES (%s)", (titre,))
"""
conn = self._connect()
try:
with conn.cursor() as cur:
yield cur
if commit:
conn.commit()
except pymysql.MySQLError:
conn.rollback()
logger.exception("Erreur SQL — transaction annulée (rollback)")
raise
finally:
conn.close()
def fetch_all(self, query, params=None):
with self.cursor() as cur:
cur.execute(query, params or ())
return cur.fetchall()
def fetch_one(self, query, params=None):
with self.cursor() as cur:
cur.execute(query, params or ())
return cur.fetchone()
def execute(self, query, params=None):
"""INSERT/UPDATE/DELETE avec commit automatique. Retourne lastrowid."""
with self.cursor(commit=True) as cur:
cur.execute(query, params or ())
return cur.lastrowid
def init_db(config):
"""Fabrique une instance Database à partir d'un objet de configuration."""
return Database(
host=config.DB_HOST,
port=config.DB_PORT,
user=config.DB_USER,
password=config.DB_PASSWORD,
database=config.DB_NAME,
)

101
web/init.sql Normal file
View File

@@ -0,0 +1,101 @@
-- =============================================================================
-- init.sql — Script de création du schéma GestHub (MariaDB / InnoDB)
-- =============================================================================
-- Respecte la 3ème forme normale (3NF) : aucune donnée redondante, toutes les
-- dépendances transitives éliminées. La référence aux utilisateurs se fait
-- uniquement via `sub` (identifiant OIDC Keycloak), sans duplication des
-- données du fournisseur d'identité (Bloc 2 - C4).
--
-- Moteur InnoDB : transactions ACID, contraintes de clé étrangère.
-- Exécuté automatiquement au premier démarrage du conteneur MariaDB
-- (monté sur /docker-entrypoint-initdb.d/init.sql, voir docker-compose.yml).
-- =============================================================================
SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 1;
-- -----------------------------------------------------------------------------
-- Table : annonces
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS annonces (
id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
titre VARCHAR(150) NOT NULL,
contenu TEXT NOT NULL,
epinglee TINYINT(1) NOT NULL DEFAULT 0,
auteur_sub VARCHAR(64) NOT NULL, -- sub OIDC Keycloak (pas de FK vers Keycloak)
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_annonces_auteur (auteur_sub),
INDEX idx_annonces_created (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- -----------------------------------------------------------------------------
-- Table : fichiers
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS fichiers (
id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
nom_original VARCHAR(255) NOT NULL,
nom_stocke VARCHAR(64) NOT NULL UNIQUE, -- nom UUID sur disque (anti path traversal)
mime_type VARCHAR(127) NOT NULL,
taille_octets BIGINT UNSIGNED NOT NULL, -- BIGINT : fichiers volumineux possibles
uploader_sub VARCHAR(64) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
INDEX idx_fichiers_uploader (uploader_sub)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- -----------------------------------------------------------------------------
-- Table : evenements (module planning)
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS evenements (
id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
titre VARCHAR(150) NOT NULL,
description TEXT NULL,
date_debut DATETIME NOT NULL,
date_fin DATETIME NOT NULL,
createur_sub VARCHAR(64) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT chk_evenements_dates CHECK (date_fin >= date_debut),
INDEX idx_evenements_dates (date_debut, date_fin)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- -----------------------------------------------------------------------------
-- Table : blocks (widgets du tableau de bord — module historique, conservé)
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS blocks (
id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
block_type VARCHAR(50) NOT NULL, -- 'iframe' | 'buttons' | 'html'
column_name VARCHAR(20) NOT NULL, -- 'left' | 'center' | 'right'
position INT UNSIGNED NOT NULL DEFAULT 0,
data TEXT NULL, -- JSON sérialisé (url, titre, liens...)
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- -----------------------------------------------------------------------------
-- Table : audit_log (traçabilité des actions sensibles — Bloc 1 - C10)
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS audit_log (
id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
action VARCHAR(50) NOT NULL, -- CREATE_ANNONCE, DELETE_FILE, LOGIN, LOGOUT, REFUSE...
ressource VARCHAR(150) NULL, -- ex. 'annonce:42', 'fichier:uuid'
statut VARCHAR(20) NOT NULL, -- SUCCESS | REFUSED | ERROR
utilisateur_sub VARCHAR(64) NULL,
ip_address VARCHAR(45) NULL, -- IPv4 ou IPv6
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
INDEX idx_audit_action (action),
INDEX idx_audit_created (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- -----------------------------------------------------------------------------
-- Projection méthodologique (non implémentée) : trigger de traçabilité des
-- modifications directes en base, hors application (voir Bloc 1 - C10) :
--
-- DELIMITER $$
-- CREATE TRIGGER trg_annonces_after_update
-- AFTER UPDATE ON annonces
-- FOR EACH ROW
-- BEGIN
-- INSERT INTO audit_log (action, ressource, statut, utilisateur_sub)
-- VALUES ('DIRECT_UPDATE', CONCAT('annonce:', NEW.id), 'SUCCESS', NULL);
-- END$$
-- DELIMITER ;
-- -----------------------------------------------------------------------------

View File

@@ -1,65 +1,60 @@
from app import app, db, Block
import json
"""
init_db.py — Injection des widgets par défaut du tableau de bord au premier
démarrage (les tables elles-mêmes sont créées par init.sql, exécuté
automatiquement par le conteneur MariaDB).
# Configuration des widgets actuels extraits de ton index.html
initial_blocks = [
# 1. Colonne GAUCHE : Le Planning
Block(
block_type='iframe',
column_name='left',
position=0,
data=json.dumps({
Usage : python init_db.py
"""
from config import Config
from db import init_db
from models import block_model
INITIAL_BLOCKS = [
{
"type": "iframe",
"column": "left",
"data": {
"title": "Planning / Agenda",
"url": "https://mattermost.ninolbt.com/boards/team/8xj6d4ukwigk7rznqi3w339x7e/b3kmbqfwd33dmdy9g9g3ezaoxza/vu4nuxhf73ircznrkrcgzonno8a",
"url": "https://chat.ninolbt.com/boards/team/example/planning",
"height": "100%",
"styleClass": "planning"
})
),
# 2. Colonne CENTRALE : Le Trello (Board)
Block(
block_type='iframe',
column_name='center',
position=0,
data=json.dumps({
"styleClass": "planning",
},
},
{
"type": "iframe",
"column": "center",
"data": {
"title": "",
"url": "https://mattermost.ninolbt.com/boards/team/8xj6d4ukwigk7rznqi3w339x7e/b3kmbqfwd33dmdy9g9g3ezaoxza/va5xp53m6spbi8qo6qnng7711me",
"url": "https://chat.ninolbt.com/boards/team/example/kanban",
"height": "680",
"styleClass": "trello"
})
),
# 3. Colonne DROITE : Les boutons
Block(
block_type='buttons',
column_name='right',
position=0,
data=json.dumps({
"styleClass": "trello",
},
},
{
"type": "buttons",
"column": "right",
"data": {
"links": [
{"label": "Projet", "url": "#"},
{"label": "GDD Global", "url": "#"},
{"label": "Bible 3D Art", "url": "#"},
{"label": "Bible GameDev", "url": "#"},
{"label": "Réglement du HUB", "url": "#"}
{"label": "Documentation", "url": "#"},
{"label": "Règlement du HUB", "url": "#"},
]
})
)
},
},
]
def init_data():
with app.app_context():
# Crée les tables si elles n'existent pas encore
db.create_all()
# Vérifie si la DB est déjà remplie pour éviter les doublons
if Block.query.first():
print("La base de données contient déjà des blocs.")
return
print("Injection des widgets par défaut...")
for block in initial_blocks:
db.session.add(block)
db.session.commit()
print("Terminé ! Tes widgets sont en base de données.")
def main():
db = init_db(Config)
if block_model.fetch_all(db):
print("La base de données contient déjà des blocs.")
return
print("Injection des widgets par défaut...")
for block in INITIAL_BLOCKS:
block_model.insert(db, block["type"], block["column"], block["data"])
print("Terminé.")
if __name__ == "__main__":
init_data()
main()

13
web/models/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
"""
Couche models/ — Repository Pattern (Bloc 1 - C4, Bloc 2 - C4).
Chaque module de ce package encapsule TOUT l'accès SQL pour une entité :
les couches routes/ et services/ n'écrivent jamais de SQL elles-mêmes et
ne connaissent pas le schéma de la base. Cela garantit que le changement
de SGBDR (ex. migration vers PostgreSQL) ne nécessite que la réécriture de
ce package — le reste de l'application est inchangé (dépendance vers
l'abstraction, cf. dossier technique section C3/C4).
Chaque module expose une interface CRUD standardisée :
insert(db, ...) / fetch_all(db, ...) / get(db, id) / update(db, ...) / delete(db, id)
"""

View File

@@ -0,0 +1,57 @@
"""
annonce_model.py — Accès aux données pour l'entité "annonce".
Toutes les requêtes utilisent des paramètres préparés (%s) : aucune
concaténation de chaîne SQL n'est utilisée dans ce module (OWASP A03:2021 —
Injection). Voir tests de sécurité TS-01 (tests/test_security.py).
"""
class AnnonceNotFoundError(Exception):
"""Exception métier levée quand une annonce n'existe pas.
La couche services/ traduit les erreurs SQL bas niveau en exceptions
métier compréhensibles, afin que les routes n'aient jamais à connaître
les détails de pymysql (cf. C4 — convention de gestion des erreurs).
"""
def insert(db, titre, contenu, auteur_sub, epinglee=False):
query = (
"INSERT INTO annonces (titre, contenu, epinglee, auteur_sub) "
"VALUES (%s, %s, %s, %s)"
)
return db.execute(query, (titre, contenu, int(epinglee), auteur_sub))
def fetch_all(db):
query = (
"SELECT id, titre, contenu, epinglee, auteur_sub, created_at, updated_at "
"FROM annonces ORDER BY epinglee DESC, created_at DESC"
)
return db.fetch_all(query)
def get(db, annonce_id):
query = (
"SELECT id, titre, contenu, epinglee, auteur_sub, created_at, updated_at "
"FROM annonces WHERE id = %s"
)
return db.fetch_one(query, (annonce_id,))
def update(db, annonce_id, titre, contenu, epinglee=False):
query = (
"UPDATE annonces SET titre = %s, contenu = %s, epinglee = %s "
"WHERE id = %s"
)
with db.cursor(commit=True) as cur:
cur.execute(query, (titre, contenu, int(epinglee), annonce_id))
return cur.rowcount
def delete(db, annonce_id):
query = "DELETE FROM annonces WHERE id = %s"
with db.cursor(commit=True) as cur:
cur.execute(query, (annonce_id,))
return cur.rowcount

17
web/models/audit_model.py Normal file
View File

@@ -0,0 +1,17 @@
"""audit_model.py — Écriture dans la table de traçabilité audit_log (Bloc 1 - C10)."""
def insert(db, action, statut, ressource=None, utilisateur_sub=None, ip_address=None):
query = (
"INSERT INTO audit_log (action, ressource, statut, utilisateur_sub, ip_address) "
"VALUES (%s, %s, %s, %s, %s)"
)
return db.execute(query, (action, ressource, statut, utilisateur_sub, ip_address))
def fetch_recent(db, limit=100):
query = (
"SELECT id, action, ressource, statut, utilisateur_sub, ip_address, created_at "
"FROM audit_log ORDER BY created_at DESC LIMIT %s"
)
return db.fetch_all(query, (limit,))

59
web/models/block_model.py Normal file
View File

@@ -0,0 +1,59 @@
"""
block_model.py — Accès aux données pour l'entité "block" (widgets du dashboard).
Module historique de GestHub, migré de Flask-SQLAlchemy vers PyMySQL pour
rester cohérent avec le reste de la couche models/ (accès direct, sans ORM,
requêtes paramétrées).
"""
import json
def _row_to_dict(row):
if row is None:
return None
return {
"id": row["id"],
"type": row["block_type"],
"column": row["column_name"],
"position": row["position"],
"data": json.loads(row["data"]) if row["data"] else {},
}
def fetch_all(db):
query = (
"SELECT id, block_type, column_name, position, data FROM blocks "
"ORDER BY column_name, position"
)
rows = db.fetch_all(query)
return [_row_to_dict(r) for r in rows]
def insert(db, block_type, column_name, data, position=99):
query = (
"INSERT INTO blocks (block_type, column_name, position, data) "
"VALUES (%s, %s, %s, %s)"
)
new_id = db.execute(query, (block_type, column_name, position, json.dumps(data)))
return _row_to_dict({
"id": new_id,
"block_type": block_type,
"column_name": column_name,
"position": position,
"data": json.dumps(data),
})
def update_position(db, block_id, column_name, position):
query = "UPDATE blocks SET column_name = %s, position = %s WHERE id = %s"
with db.cursor(commit=True) as cur:
cur.execute(query, (column_name, position, block_id))
return cur.rowcount
def delete(db, block_id):
query = "DELETE FROM blocks WHERE id = %s"
with db.cursor(commit=True) as cur:
cur.execute(query, (block_id,))
return cur.rowcount

View File

@@ -0,0 +1,32 @@
"""evenement_model.py — Accès aux données pour l'entité "evenement" (planning)."""
def insert(db, titre, description, date_debut, date_fin, createur_sub):
query = (
"INSERT INTO evenements (titre, description, date_debut, date_fin, createur_sub) "
"VALUES (%s, %s, %s, %s, %s)"
)
return db.execute(query, (titre, description, date_debut, date_fin, createur_sub))
def fetch_all(db):
query = (
"SELECT id, titre, description, date_debut, date_fin, createur_sub, created_at "
"FROM evenements ORDER BY date_debut ASC"
)
return db.fetch_all(query)
def get(db, evenement_id):
query = (
"SELECT id, titre, description, date_debut, date_fin, createur_sub, created_at "
"FROM evenements WHERE id = %s"
)
return db.fetch_one(query, (evenement_id,))
def delete(db, evenement_id):
query = "DELETE FROM evenements WHERE id = %s"
with db.cursor(commit=True) as cur:
cur.execute(query, (evenement_id,))
return cur.rowcount

View File

@@ -0,0 +1,32 @@
"""fichier_model.py — Accès aux données pour l'entité "fichier"."""
def insert(db, nom_original, nom_stocke, mime_type, taille_octets, uploader_sub):
query = (
"INSERT INTO fichiers (nom_original, nom_stocke, mime_type, taille_octets, uploader_sub) "
"VALUES (%s, %s, %s, %s, %s)"
)
return db.execute(query, (nom_original, nom_stocke, mime_type, taille_octets, uploader_sub))
def fetch_all(db):
query = (
"SELECT id, nom_original, nom_stocke, mime_type, taille_octets, uploader_sub, created_at "
"FROM fichiers ORDER BY created_at DESC"
)
return db.fetch_all(query)
def get(db, fichier_id):
query = (
"SELECT id, nom_original, nom_stocke, mime_type, taille_octets, uploader_sub, created_at "
"FROM fichiers WHERE id = %s"
)
return db.fetch_one(query, (fichier_id,))
def delete(db, fichier_id):
query = "DELETE FROM fichiers WHERE id = %s"
with db.cursor(commit=True) as cur:
cur.execute(query, (fichier_id,))
return cur.rowcount

54
web/models/rgpd_model.py Normal file
View File

@@ -0,0 +1,54 @@
"""
rgpd_model.py — Anonymisation des données personnelles (droit à l'effacement,
RGPD art. 17). Bloc 1 - C10 : traçabilité, et Bloc 2 - C4 : la seule donnée
personnelle stockée dans l'application est le `sub` OIDC (Keycloak) — aucune
autre table ne duplique d'attribut d'identité.
`sub` est référencé dans quatre tables : annonces.auteur_sub,
fichiers.uploader_sub, evenements.createur_sub (toutes NOT NULL — la donnée
ne peut donc pas être mise à NULL sans casser la contrainte, d'où le choix
d'un pseudonyme irréversible) et audit_log.utilisateur_sub (NULLable, mais
pseudonymisée elle aussi par cohérence, pour garder la traçabilité du volume
d'actions sans conserver l'identité).
Toutes les mises à jour sont exécutées dans UNE seule transaction InnoDB
(un seul curseur `commit=True`) : soit les quatre tables sont mises à jour,
soit aucune ne l'est en cas d'erreur (cohérence, pas d'anonymisation partielle).
"""
def anonymize(db, target_sub, anon_value):
"""Remplace `target_sub` par `anon_value` dans toutes les tables qui le
référencent. Retourne le nombre de lignes modifiées par table.
"""
with db.cursor(commit=True) as cur:
cur.execute(
"UPDATE annonces SET auteur_sub = %s WHERE auteur_sub = %s",
(anon_value, target_sub),
)
annonces_rows = cur.rowcount
cur.execute(
"UPDATE fichiers SET uploader_sub = %s WHERE uploader_sub = %s",
(anon_value, target_sub),
)
fichiers_rows = cur.rowcount
cur.execute(
"UPDATE evenements SET createur_sub = %s WHERE createur_sub = %s",
(anon_value, target_sub),
)
evenements_rows = cur.rowcount
cur.execute(
"UPDATE audit_log SET utilisateur_sub = %s WHERE utilisateur_sub = %s",
(anon_value, target_sub),
)
audit_rows = cur.rowcount
return {
"annonces": annonces_rows,
"fichiers": fichiers_rows,
"evenements": evenements_rows,
"audit_log": audit_rows,
}

15
web/requirements.txt Normal file
View File

@@ -0,0 +1,15 @@
# Versions figées pour garantir la reproductibilité de l'environnement
# (Bloc 3 - C6 : gestion des dépendances, évite le "dependency hell").
flask==3.0.3
authlib==1.3.2
requests==2.32.3
pymysql==1.1.1
python-magic==0.4.27
python-dotenv==1.0.1
gunicorn==23.0.0
# --- Dépendances de test (Bloc 1 - C5, Bloc 3 - C7) ---
pytest==8.3.3
pytest-cov==5.0.0
flake8==7.1.1
pyyaml==6.0.3

6
web/routes/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""Couche routes/ — orchestration HTTP uniquement (blueprints Flask).
Chaque route : (1) authentifie/autorise via services.auth_service,
(2) délègue la logique métier à la couche services/, (3) retourne une
réponse JSON ou un rendu de template. Aucune route n'exécute de SQL.
"""

62
web/routes/annonces.py Normal file
View File

@@ -0,0 +1,62 @@
"""routes/annonces.py — API CRUD du module Annonces (EXF-01 à EXF-04)."""
from flask import Blueprint, current_app, jsonify, request
from models.annonce_model import AnnonceNotFoundError
from services import annonce_service
from services.auth_service import client_ip, current_user, require_admin, require_login
bp = Blueprint("annonces", __name__, url_prefix="/api/annonces")
def _db():
return current_app.config["DB"]
@bp.route("", methods=["GET"])
@require_login
def list_annonces():
return jsonify(annonce_service.list_annonces(_db()))
@bp.route("", methods=["POST"])
@require_admin
def create_annonce():
data = request.json or {}
user = current_user()
try:
new_id = annonce_service.create_annonce(
_db(), data.get("titre"), data.get("contenu"), user["sub"],
data.get("epinglee", False), client_ip(),
)
except annonce_service.ValidationError as exc:
return jsonify({"error": str(exc)}), 400
return jsonify({"id": new_id}), 201
@bp.route("/<int:annonce_id>", methods=["PUT"])
@require_admin
def update_annonce(annonce_id):
data = request.json or {}
user = current_user()
try:
annonce_service.update_annonce(
_db(), annonce_id, data.get("titre"), data.get("contenu"), user["sub"],
data.get("epinglee", False), client_ip(),
)
except annonce_service.ValidationError as exc:
return jsonify({"error": str(exc)}), 400
except AnnonceNotFoundError as exc:
return jsonify({"error": str(exc)}), 404
return jsonify({"status": "updated"})
@bp.route("/<int:annonce_id>", methods=["DELETE"])
@require_admin
def delete_annonce(annonce_id):
user = current_user()
try:
annonce_service.delete_annonce(_db(), annonce_id, user["sub"], client_ip())
except AnnonceNotFoundError as exc:
return jsonify({"error": str(exc)}), 404
return jsonify({"status": "deleted"})

86
web/routes/auth.py Normal file
View File

@@ -0,0 +1,86 @@
"""
routes/auth.py — Flux d'authentification OIDC (Authorization Code Flow) avec
Keycloak, via Authlib. 7 étapes documentées dans la rétro-documentation
(RD-3) et le journal de bord :
1. Détection de l'absence de session (décorateur / vérification session)
2. Génération de l'URL d'autorisation (+ nonce anti-rejeu)
3. Redirection de l'utilisateur vers Keycloak
4. Retour sur /auth avec le code d'autorisation
5. Échange du code contre un token (authorize_access_token)
6. Vérification du token (signature, nonce, expiration — parse_id_token)
7. Création de la session applicative Flask
"""
import uuid
from flask import Blueprint, redirect, url_for, session
from config import Config
from services import audit_service
from services.auth_service import client_ip
def register_oauth(oauth):
"""Enregistre le client OIDC Keycloak auprès d'Authlib (appelé depuis app.py)."""
return oauth.register(
name="keycloak",
client_id=Config.OIDC_CLIENT_ID,
client_secret=Config.OIDC_CLIENT_SECRET,
server_metadata_url=Config.OIDC_DISCOVERY_URL,
client_kwargs={"scope": "openid profile email"},
)
def init_auth_routes(oauth):
# Le blueprint est créé ICI (et non au niveau du module) afin que
# chaque appel de create_app() (production, ou une fois par test en
# environnement pytest) produise un blueprint neuf. Un blueprint créé
# au niveau module serait décoré une seule fois puis réutilisé tel
# quel, ce que Flask refuse dès le second register_blueprint().
bp = Blueprint("auth", __name__)
keycloak = register_oauth(oauth)
@bp.route("/login")
def login():
# Étape 2 : génération de l'URL d'autorisation avec un nonce
nonce = uuid.uuid4().hex
session["nonce"] = nonce
redirect_uri = url_for("auth.auth_callback", _external=True, _scheme="https")
# Étape 3 : redirection vers Keycloak
return keycloak.authorize_redirect(redirect_uri, nonce=nonce)
@bp.route("/auth")
def auth_callback():
db = _get_db()
# Étape 5 : échange du code contre un token
token = keycloak.authorize_access_token()
nonce = session.pop("nonce", None)
# Étape 6 : vérification (signature + nonce) du token
userinfo = keycloak.parse_id_token(token, nonce=nonce)
# Étape 7 : création de la session applicative
session["user"] = userinfo
session["id_token"] = token.get("id_token")
audit_service.log(db, "LOGIN", None, "SUCCESS", userinfo.get("sub"), client_ip())
return redirect("/")
@bp.route("/logout")
def logout():
db = _get_db()
user = session.get("user") or {}
id_token = session.get("id_token")
audit_service.log(db, "LOGOUT", None, "SUCCESS", user.get("sub"), client_ip())
session.clear()
return redirect(
f"{Config.OIDC_LOGOUT_URL}"
f"?post_logout_redirect_uri={Config.PUBLIC_BASE_URL}"
f"&id_token_hint={id_token}"
)
return bp
def _get_db():
from flask import current_app
return current_app.config["DB"]

55
web/routes/dashboard.py Normal file
View File

@@ -0,0 +1,55 @@
"""routes/dashboard.py — page d'accueil + API des widgets (layout)."""
from flask import Blueprint, current_app, jsonify, redirect, render_template, request, url_for
from services import block_service
from services.auth_service import current_user, is_admin, is_authenticated, require_admin
bp = Blueprint("dashboard", __name__)
def _db():
return current_app.config["DB"]
@bp.route("/")
def index():
if not is_authenticated():
return redirect(url_for("auth.login"))
return render_template("view/index.html", user=current_user())
@bp.route("/api/is_admin")
def api_is_admin():
return jsonify({"admin": is_admin()})
@bp.route("/api/layout", methods=["GET"])
def get_layout():
return jsonify(block_service.list_blocks(_db()))
@bp.route("/api/layout/save", methods=["POST"])
@require_admin
def save_layout():
block_service.save_layout(_db(), request.json or [])
return jsonify({"status": "saved"})
@bp.route("/api/block/add", methods=["POST"])
@require_admin
def add_block():
data = request.json or {}
block = block_service.add_block(
_db(), data.get("type"), data.get("column", "center"), data.get("data", {})
)
return jsonify(block)
@bp.route("/api/block/<int:block_id>", methods=["DELETE"])
@require_admin
def delete_block(block_id):
rows = block_service.delete_block(_db(), block_id)
if rows:
return jsonify({"status": "deleted"})
return jsonify({"error": "not found"}), 404

44
web/routes/evenements.py Normal file
View File

@@ -0,0 +1,44 @@
"""routes/evenements.py — API CRUD du module Planning (EXF-09, EXF-10)."""
from flask import Blueprint, current_app, jsonify, request
from services import evenement_service
from services.auth_service import client_ip, current_user, require_admin, require_login
bp = Blueprint("evenements", __name__, url_prefix="/api/evenements")
def _db():
return current_app.config["DB"]
@bp.route("", methods=["GET"])
@require_login
def list_evenements():
return jsonify(evenement_service.list_evenements(_db()))
@bp.route("", methods=["POST"])
@require_admin
def create_evenement():
data = request.json or {}
user = current_user()
try:
new_id = evenement_service.create_evenement(
_db(), data.get("titre"), data.get("description"),
data.get("date_debut"), data.get("date_fin"), user["sub"], client_ip(),
)
except evenement_service.ValidationError as exc:
return jsonify({"error": str(exc)}), 400
return jsonify({"id": new_id}), 201
@bp.route("/<int:evenement_id>", methods=["DELETE"])
@require_admin
def delete_evenement(evenement_id):
user = current_user()
try:
evenement_service.delete_evenement(_db(), evenement_id, user["sub"], client_ip())
except LookupError as exc:
return jsonify({"error": str(exc)}), 404
return jsonify({"status": "deleted"})

59
web/routes/fichiers.py Normal file
View File

@@ -0,0 +1,59 @@
"""routes/fichiers.py — API upload/download du module Fichiers (EXF-05 à EXF-08)."""
from flask import Blueprint, current_app, jsonify, request, send_file
from services import fichier_service
from services.auth_service import client_ip, current_user, is_admin, require_login
bp = Blueprint("fichiers", __name__, url_prefix="/api/fichiers")
def _db():
return current_app.config["DB"]
@bp.route("", methods=["GET"])
@require_login
def list_fichiers():
return jsonify(fichier_service.list_fichiers(_db()))
@bp.route("/upload", methods=["POST"])
@require_login
def upload():
user = current_user()
try:
fichier_id = fichier_service.handle_upload(
_db(), request.files.get("file"), user["sub"], client_ip()
)
except fichier_service.FileValidationError as exc:
return jsonify({"error": str(exc)}), 400
return jsonify({"id": fichier_id}), 201
@bp.route("/<int:fichier_id>/download", methods=["GET"])
@require_login
def download(fichier_id):
user = current_user()
try:
path, original_name = fichier_service.prepare_download(
_db(), fichier_id, user["sub"], is_admin(user), client_ip()
)
except FileNotFoundError:
return jsonify({"error": "not found"}), 404
except PermissionError:
return jsonify({"error": "forbidden"}), 403
return send_file(path, as_attachment=True, download_name=original_name)
@bp.route("/<int:fichier_id>", methods=["DELETE"])
@require_login
def delete(fichier_id):
user = current_user()
try:
fichier_service.delete_fichier(_db(), fichier_id, user["sub"], is_admin(user), client_ip())
except FileNotFoundError:
return jsonify({"error": "not found"}), 404
except PermissionError:
return jsonify({"error": "forbidden"}), 403
return jsonify({"status": "deleted"})

46
web/routes/rgpd.py Normal file
View File

@@ -0,0 +1,46 @@
"""
routes/rgpd.py — Droit à l'effacement RGPD (art. 17) : anonymisation des
données d'un utilisateur (identifié par son `sub` OIDC Keycloak).
Réservé aux administrateurs (@require_admin), au même titre que les autres
routes de modification (annonces, fichiers, blocks) — cf. auth_service.py.
"""
from flask import Blueprint, current_app, jsonify
from config import Config
from services import rgpd_service
from services.auth_service import client_ip, current_user, require_admin
bp = Blueprint("rgpd", __name__, url_prefix="/api/admin/users")
def _db():
return current_app.config["DB"]
def _pepper():
return getattr(Config, "RGPD_ANON_PEPPER", None) or Config.SECRET_KEY
@bp.route("/<sub>/data", methods=["DELETE"])
@require_admin
def anonymize_user_data(sub):
actor = current_user()
try:
result = rgpd_service.anonymize_user(
_db(), sub, _pepper(), actor.get("sub"), client_ip()
)
except rgpd_service.InvalidSubError as exc:
return jsonify({"error": str(exc)}), 400
if not result["anonymized"]:
return jsonify({"error": "aucune donnée trouvée pour cet utilisateur"}), 404
return jsonify(
{
"status": "anonymized",
"pseudonym": result["pseudonym"],
"rows_updated": result["rows"],
}
)

1
web/services/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Couche services/ — logique métier. Les routes n'appellent jamais les models/ directement."""

View File

@@ -0,0 +1,55 @@
"""
annonce_service.py — Logique métier du module Annonces (EXF-01 à EXF-04 du CDC).
Validation des champs AVANT d'appeler la couche models/, et journalisation
systématique des actions sensibles dans audit_log (Bloc 1 - C10).
"""
from models import annonce_model
from services import audit_service
class ValidationError(Exception):
"""Levée quand les données fournies par l'utilisateur sont invalides."""
TITRE_MAX_LEN = 150
def _validate(titre, contenu):
if not titre or not titre.strip():
raise ValidationError("Le titre est obligatoire.")
if len(titre) > TITRE_MAX_LEN:
raise ValidationError(f"Le titre dépasse {TITRE_MAX_LEN} caractères.")
if not contenu or not contenu.strip():
raise ValidationError("Le contenu est obligatoire.")
def list_annonces(db):
return annonce_model.fetch_all(db)
def create_annonce(db, titre, contenu, auteur_sub, epinglee=False, ip=None):
_validate(titre, contenu)
new_id = annonce_model.insert(db, titre.strip(), contenu.strip(), auteur_sub, epinglee)
audit_service.log(db, "CREATE_ANNONCE", f"annonce:{new_id}", "SUCCESS", auteur_sub, ip)
return new_id
def update_annonce(db, annonce_id, titre, contenu, auteur_sub, epinglee=False, ip=None):
_validate(titre, contenu)
existing = annonce_model.get(db, annonce_id)
if not existing:
raise annonce_model.AnnonceNotFoundError(f"Annonce {annonce_id} introuvable")
rows = annonce_model.update(db, annonce_id, titre.strip(), contenu.strip(), epinglee)
audit_service.log(db, "UPDATE_ANNONCE", f"annonce:{annonce_id}", "SUCCESS", auteur_sub, ip)
return rows
def delete_annonce(db, annonce_id, auteur_sub, ip=None):
existing = annonce_model.get(db, annonce_id)
if not existing:
raise annonce_model.AnnonceNotFoundError(f"Annonce {annonce_id} introuvable")
rows = annonce_model.delete(db, annonce_id)
audit_service.log(db, "DELETE_ANNONCE", f"annonce:{annonce_id}", "SUCCESS", auteur_sub, ip)
return rows

View File

@@ -0,0 +1,11 @@
"""audit_service.py — Point d'entrée unique pour écrire dans audit_log."""
from models import audit_model
def log(db, action, ressource, statut, utilisateur_sub=None, ip_address=None):
return audit_model.insert(db, action, statut, ressource, utilisateur_sub, ip_address)
def recent(db, limit=100):
return audit_model.fetch_recent(db, limit)

View File

@@ -0,0 +1,83 @@
"""
auth_service.py — Fonctions de validation des droits d'accès, centralisées et
réutilisées par toutes les routes (Bloc 1 - C9 : taux de réutilisation > 80 %
via services/ ; Bloc 1 - C8 : "vérification JWT à chaque requête").
Toute route sensible doit passer par current_user() / require_login() /
require_admin() plutôt que de relire directement flask.session, afin que la
règle d'autorisation soit définie à un seul endroit.
"""
from functools import wraps
from flask import session, jsonify, request
from config import Config
def current_user():
"""Retourne l'utilisateur en session (dict issu du token OIDC) ou None."""
return session.get("user")
def is_authenticated():
return current_user() is not None
_UNSET = object()
def is_admin(user=_UNSET):
"""True si l'utilisateur appartient au groupe Keycloak /admin.
Appelé sans argument, lit l'utilisateur courant dans la session Flask
(nécessite un contexte de requête actif). Appelé avec un argument
explicite (y compris None), ce dict/None est utilisé tel quel — utile
pour tester la fonction hors contexte de requête (cf. UT-11).
"""
if user is _UNSET:
user = current_user()
if not user:
return False
return Config.ADMIN_GROUP in user.get("groups", [])
def require_login(view_func):
"""Décorateur : bloque l'accès si aucune session utilisateur valide."""
@wraps(view_func)
def wrapper(*args, **kwargs):
if not is_authenticated():
return jsonify({"error": "unauthorized"}), 401
return view_func(*args, **kwargs)
return wrapper
def require_admin(view_func):
"""Décorateur : bloque l'accès si l'utilisateur n'a pas le groupe /admin.
Utilisé par toutes les routes de modification (annonces, fichiers,
blocks). Aucune route admin n'est accessible sans token valide et sans
appartenance au groupe Keycloak /admin (mitigation du risque "accès non
autorisé à /admin", cf. tableau OWASP du dossier technique).
"""
@wraps(view_func)
def wrapper(*args, **kwargs):
user = current_user()
if not user:
return jsonify({"error": "unauthorized"}), 401
if not is_admin(user):
return jsonify({"error": "forbidden"}), 403
return view_func(*args, **kwargs)
return wrapper
def client_ip():
"""IP cliente, en tenant compte du proxy Caddy (X-Forwarded-For)."""
forwarded = request.headers.get("X-Forwarded-For", "")
if forwarded:
return forwarded.split(",")[0].strip()
return request.remote_addr

View File

@@ -0,0 +1,25 @@
"""block_service.py — Logique métier du tableau de bord (widgets)."""
from models import block_model
def list_blocks(db):
return block_model.fetch_all(db)
def add_block(db, block_type, column, data):
if block_type not in {"iframe", "buttons", "html"}:
raise ValueError(f"Type de bloc inconnu : {block_type}")
if column not in {"left", "center", "right"}:
raise ValueError(f"Colonne inconnue : {column}")
return block_model.insert(db, block_type, column, data)
def save_layout(db, layout):
"""layout : liste de {id, column, position} après un drag & drop."""
for item in layout:
block_model.update_position(db, item["id"], item["column"], item["position"])
def delete_block(db, block_id):
return block_model.delete(db, block_id)

View File

@@ -0,0 +1,42 @@
"""evenement_service.py — Logique métier du module Planning (EXF-09, EXF-10)."""
from datetime import datetime
from models import evenement_model
from services import audit_service
class ValidationError(Exception):
pass
def _parse(dt_str):
try:
return datetime.fromisoformat(dt_str)
except (TypeError, ValueError) as exc:
raise ValidationError(f"Date invalide : {dt_str!r}") from exc
def list_evenements(db):
return evenement_model.fetch_all(db)
def create_evenement(db, titre, description, date_debut, date_fin, createur_sub, ip=None):
if not titre or not titre.strip():
raise ValidationError("Le titre est obligatoire.")
debut = _parse(date_debut)
fin = _parse(date_fin)
if fin < debut:
raise ValidationError("La date de fin doit être postérieure à la date de début.")
new_id = evenement_model.insert(db, titre.strip(), description, debut, fin, createur_sub)
audit_service.log(db, "CREATE_EVENEMENT", f"evenement:{new_id}", "SUCCESS", createur_sub, ip)
return new_id
def delete_evenement(db, evenement_id, createur_sub, ip=None):
existing = evenement_model.get(db, evenement_id)
if not existing:
raise LookupError(f"Événement {evenement_id} introuvable")
rows = evenement_model.delete(db, evenement_id)
audit_service.log(db, "DELETE_EVENEMENT", f"evenement:{evenement_id}", "SUCCESS", createur_sub, ip)
return rows

View File

@@ -0,0 +1,160 @@
"""
fichier_service.py — Module de partage de fichiers (EXF-05 à EXF-08 du CDC).
Upload décomposé en 8 étapes indépendantes et testables unitairement
(cf. dossier technique — Bloc 3, C2 "Décomposition algorithmique") :
1) Vérification de la présence du fichier dans la requête
2) Vérification de la taille (avant lecture complète du flux)
3) Vérification du type MIME réel (python-magic, analyse binaire du contenu)
4) Vérification de l'extension déclarée (liste blanche)
5) Génération d'un nom de fichier sécurisé (UUID4)
6) Sauvegarde sur disque
7) Enregistrement des métadonnées en base de données
8) Journalisation de l'action dans audit_log
Chaque étape peut échouer indépendamment avec un message d'erreur explicite,
ce qui facilite le test (voir tests/test_unit_fichier_service.py, UT-06 à UT-08)
et le débogage (pas de bloc "tout ou rien").
"""
import os
import uuid
try:
import magic # python-magic : détection MIME par analyse binaire (libmagic)
_HAS_MAGIC = True
except ImportError: # pragma: no cover - environnement sans libmagic (dev/CI léger)
_HAS_MAGIC = False
from config import Config
from models import fichier_model
from services import audit_service
class FileValidationError(Exception):
"""Erreur métier explicite pour chaque étape de validation d'upload."""
# --- Étape 1 -----------------------------------------------------------------
def step1_check_presence(file_storage):
if file_storage is None or file_storage.filename == "":
raise FileValidationError("Aucun fichier fourni dans la requête.")
return file_storage
# --- Étape 2 -----------------------------------------------------------------
def step2_check_size(file_storage, max_size=None):
max_size = max_size if max_size is not None else Config.MAX_UPLOAD_SIZE_BYTES
file_storage.stream.seek(0, os.SEEK_END)
size = file_storage.stream.tell()
file_storage.stream.seek(0)
if size == 0:
raise FileValidationError("Le fichier est vide.")
if size > max_size:
raise FileValidationError(
f"Fichier trop volumineux ({size} octets, max {max_size})."
)
return size
# --- Étape 3 -----------------------------------------------------------------
def step3_check_mime(file_storage, allowed_mimes=None):
allowed_mimes = allowed_mimes or Config.ALLOWED_MIME_TYPES
head = file_storage.stream.read(2048)
file_storage.stream.seek(0)
if _HAS_MAGIC:
mime_type = magic.from_buffer(head, mime=True)
else: # repli : type déclaré par le client (moins fiable, signalé dans les logs)
mime_type = file_storage.mimetype or "application/octet-stream"
if mime_type not in allowed_mimes:
raise FileValidationError(f"Type de fichier non autorisé : {mime_type}")
return mime_type
# --- Étape 4 -----------------------------------------------------------------
def step4_check_extension(filename, allowed_ext=None):
allowed_ext = allowed_ext or Config.ALLOWED_EXTENSIONS
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
if ext not in allowed_ext:
raise FileValidationError(f"Extension non autorisée : .{ext}")
return ext
# --- Étape 5 -----------------------------------------------------------------
def step5_generate_safe_name(ext):
return f"{uuid.uuid4().hex}.{ext}"
# --- Étape 6 -----------------------------------------------------------------
def step6_save_to_disk(file_storage, safe_name, upload_dir=None):
# NB : upload_dir n'est PAS résolu en valeur par défaut de paramètre
# (qui serait figée à l'import du module) mais lu dynamiquement dans le
# corps de la fonction, afin de respecter Config.UPLOAD_DIR à l'instant
# de l'appel (important notamment pour les tests, qui redirigent
# UPLOAD_DIR vers un répertoire temporaire via monkeypatch).
upload_dir = upload_dir or Config.UPLOAD_DIR
os.makedirs(upload_dir, exist_ok=True)
destination = os.path.join(upload_dir, safe_name)
file_storage.save(destination)
return destination
# --- Étapes 7 & 8 --------------------------------------------------------------
def step7_persist_metadata(db, original_name, safe_name, mime_type, size, uploader_sub):
return fichier_model.insert(db, original_name, safe_name, mime_type, size, uploader_sub)
def step8_log_audit(db, fichier_id, uploader_sub, ip=None):
audit_service.log(db, "UPLOAD_FILE", f"fichier:{fichier_id}", "SUCCESS", uploader_sub, ip)
def handle_upload(db, file_storage, uploader_sub, ip=None):
"""Orchestre les 8 étapes. Lève FileValidationError au premier échec."""
step1_check_presence(file_storage)
size = step2_check_size(file_storage)
mime_type = step3_check_mime(file_storage)
ext = step4_check_extension(file_storage.filename)
safe_name = step5_generate_safe_name(ext)
step6_save_to_disk(file_storage, safe_name)
fichier_id = step7_persist_metadata(
db, file_storage.filename, safe_name, mime_type, size, uploader_sub
)
step8_log_audit(db, fichier_id, uploader_sub, ip)
return fichier_id
def list_fichiers(db):
return fichier_model.fetch_all(db)
def prepare_download(db, fichier_id, requester_sub, is_admin, ip=None):
"""Vérifie les droits avant de renvoyer le chemin disque du fichier.
Règle métier : un fichier est téléchargeable par son propriétaire ou par
un administrateur (mitigation OWASP A01 — Broken Access Control /
"accès aux ressources d'autrui", cf. tests TS-05).
"""
fichier = fichier_model.get(db, fichier_id)
if not fichier:
raise FileNotFoundError(f"Fichier {fichier_id} introuvable")
if fichier["uploader_sub"] != requester_sub and not is_admin:
audit_service.log(db, "DOWNLOAD_FILE", f"fichier:{fichier_id}", "REFUSED", requester_sub, ip)
raise PermissionError("Accès refusé à ce fichier.")
audit_service.log(db, "DOWNLOAD_FILE", f"fichier:{fichier_id}", "SUCCESS", requester_sub, ip)
return os.path.join(Config.UPLOAD_DIR, fichier["nom_stocke"]), fichier["nom_original"]
def delete_fichier(db, fichier_id, requester_sub, is_admin, ip=None):
fichier = fichier_model.get(db, fichier_id)
if not fichier:
raise FileNotFoundError(f"Fichier {fichier_id} introuvable")
if fichier["uploader_sub"] != requester_sub and not is_admin:
audit_service.log(db, "DELETE_FILE", f"fichier:{fichier_id}", "REFUSED", requester_sub, ip)
raise PermissionError("Accès refusé à ce fichier.")
path = os.path.join(Config.UPLOAD_DIR, fichier["nom_stocke"])
if os.path.exists(path):
os.remove(path)
fichier_model.delete(db, fichier_id)
audit_service.log(db, "DELETE_FILE", f"fichier:{fichier_id}", "SUCCESS", requester_sub, ip)

View File

@@ -0,0 +1,69 @@
"""
rgpd_service.py — Droit à l'effacement / anonymisation RGPD (art. 17).
GestHub ne stocke pas de compte utilisateur : l'unique donnée personnelle
conservée est le `sub` OIDC Keycloak, référencé dans annonces, fichiers,
evenements et audit_log (cf. rgpd_model.py). "Supprimer un utilisateur" au
sens RGPD consiste donc à remplacer ce `sub` partout où il apparaît par un
pseudonyme irréversible, sans toucher au contenu produit (titres, fichiers,
événements) qui reste une donnée d'activité du service, pas une donnée
personnelle une fois l'auteur dé-identifié.
Pseudonyme = HMAC-SHA256(pepper, sub), tronqué et préfixé "anon:" pour tenir
dans la colonne VARCHAR(64) et rester reconnaissable dans les journaux/DB.
Le pepper (RGPD_ANON_PEPPER, ou SECRET_KEY à défaut) n'est jamais stocké
dans la base : sans lui, remonter du pseudonyme au `sub` d'origine est
calculatoirement infaisable (irréversibilité, cf. RGPD art. 17 §1).
Le hash reste déterministe pour un même `sub` : les quatre tables reçoivent
donc le même pseudonyme, ce qui permet de vérifier a posteriori que la purge
a bien été appliquée partout, sans jamais réexposer l'identité d'origine.
"""
import hashlib
import hmac
from models import rgpd_model
from services import audit_service
ANON_PREFIX = "anon:"
_HASH_HEX_LEN = 64 - len(ANON_PREFIX) # tient dans VARCHAR(64)
class InvalidSubError(Exception):
"""Levée quand le `sub` cible est vide ou manifestement invalide."""
def _anon_value(pepper, target_sub):
digest = hmac.new(pepper.encode("utf-8"), target_sub.encode("utf-8"), hashlib.sha256)
return ANON_PREFIX + digest.hexdigest()[:_HASH_HEX_LEN]
def anonymize_user(db, target_sub, pepper, actor_sub=None, ip=None):
"""Anonymise toutes les données rattachées à `target_sub`.
Retourne {"anonymized": bool, "pseudonym": str, "rows": {...}}.
`anonymized` est False si aucune ligne ne référençait ce `sub` (rien à
faire — permet à la route de répondre 404 plutôt que de journaliser une
action fictive).
"""
target_sub = (target_sub or "").strip()
if not target_sub:
raise InvalidSubError("Le sub cible est obligatoire.")
if target_sub.startswith(ANON_PREFIX):
raise InvalidSubError("Cet utilisateur est déjà anonymisé.")
pseudonym = _anon_value(pepper, target_sub)
rows = rgpd_model.anonymize(db, target_sub, pseudonym)
total = sum(rows.values())
if total > 0:
audit_service.log(
db,
"RGPD_ANONYMIZE_USER",
f"user:{pseudonym}",
"SUCCESS",
actor_sub,
ip,
)
return {"anonymized": total > 0, "pseudonym": pseudonym, "rows": rows}

View File

@@ -0,0 +1,15 @@
<!DOCTYPE html>
<html lang="fr">
<head>
<meta charset="UTF-8" />
<title>Accès refusé — GestHub</title>
<link rel="stylesheet" href="{{ url_for('static', filename='assets/css/index.css') }}" />
</head>
<body>
<main role="main" style="text-align:center; padding: 4rem 1rem;">
<h1>403 — Accès refusé</h1>
<p>Vous n'avez pas les droits nécessaires pour accéder à cette ressource.</p>
<a href="{{ url_for('dashboard.index') }}">Retour à l'accueil</a>
</main>
</body>
</html>

View File

@@ -0,0 +1,15 @@
<!DOCTYPE html>
<html lang="fr">
<head>
<meta charset="UTF-8" />
<title>Page introuvable — GestHub</title>
<link rel="stylesheet" href="{{ url_for('static', filename='assets/css/index.css') }}" />
</head>
<body>
<main role="main" style="text-align:center; padding: 4rem 1rem;">
<h1>404 — Page introuvable</h1>
<p>La ressource demandée n'existe pas.</p>
<a href="{{ url_for('dashboard.index') }}">Retour à l'accueil</a>
</main>
</body>
</html>

View File

@@ -0,0 +1,15 @@
<!DOCTYPE html>
<html lang="fr">
<head>
<meta charset="UTF-8" />
<title>Erreur serveur — GestHub</title>
<link rel="stylesheet" href="{{ url_for('static', filename='assets/css/index.css') }}" />
</head>
<body>
<main role="main" style="text-align:center; padding: 4rem 1rem;">
<h1>500 — Erreur interne</h1>
<p>Une erreur inattendue est survenue. L'équipe technique a été notifiée.</p>
<a href="{{ url_for('dashboard.index') }}">Retour à l'accueil</a>
</main>
</body>
</html>

View File

@@ -3,33 +3,62 @@
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
<title>Gesthub</title>
<title>GestHub — Tableau de bord</title>
<link rel="stylesheet" href="{{ url_for('static', filename='assets/css/index.css') }}" />
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.7.2/css/all.min.css" />
<script src="https://cdnjs.cloudflare.com/ajax/libs/Sortable/1.15.0/Sortable.min.js"></script>
</head>
<body>
<button id="toggle-darkmode">🌓</button>
<main>
<aside class="left-column" id="left">
</aside>
<header role="banner">
<button id="toggle-darkmode" aria-label="Basculer le mode sombre / clair">🌓</button>
<span class="user-badge" aria-label="Utilisateur connecté">
{{ user.preferred_username if user else '' }}
</span>
<a href="{{ url_for('auth.logout') }}" class="logout-link" aria-label="Se déconnecter de GestHub">
Déconnexion
</a>
</header>
<section class="center-column" id="center">
<h1 class="main-title">Gesthub</h1>
</section>
<nav role="navigation" aria-label="Navigation principale des modules GestHub">
<a href="#annonces">Annonces</a>
<a href="#fichiers">Fichiers</a>
<a href="#planning">Planning</a>
</nav>
<aside class="right-column" id="right">
</aside>
<main role="main">
<aside class="left-column" id="left" aria-label="Colonne gauche du tableau de bord">
</aside>
<section class="center-column" id="center" aria-label="Colonne centrale du tableau de bord">
<h1 class="main-title">GestHub</h1>
</section>
<aside class="right-column" id="right" aria-label="Colonne droite du tableau de bord">
</aside>
</main>
<div id="admin-panel" style="display:none;">
<section id="annonces" aria-labelledby="annonces-heading">
<h2 id="annonces-heading">Annonces</h2>
<ul id="annonces-list" aria-live="polite"></ul>
</section>
<section id="fichiers" aria-labelledby="fichiers-heading">
<h2 id="fichiers-heading">Partage de fichiers</h2>
<ul id="fichiers-list" aria-live="polite"></ul>
</section>
<section id="planning" aria-labelledby="planning-heading">
<h2 id="planning-heading">Planning</h2>
<ul id="evenements-list" aria-live="polite"></ul>
</section>
<div id="admin-panel" style="display:none;" role="toolbar" aria-label="Barre d'outils d'édition (mode administrateur)">
<div class="admin-toolbar">
<span>🛠 Mode Édition</span>
<button onclick="addNewWidget('iframe')">+ Iframe</button>
<button onclick="addNewWidget('buttons')">+ Boutons</button>
<button onclick="addNewWidget('html')">+ Texte/HTML</button>
<button onclick="addNewWidget('iframe')" aria-label="Ajouter un widget de type iframe">+ Iframe</button>
<button onclick="addNewWidget('buttons')" aria-label="Ajouter un widget de type boutons">+ Boutons</button>
<button onclick="addNewWidget('html')" aria-label="Ajouter un widget de texte ou HTML">+ Texte/HTML</button>
</div>
</div>
@@ -47,8 +76,10 @@
.drag-handle { cursor: move; padding: 5px; background: rgba(0,0,0,0.1); text-align: center; color: #666; font-size: 12px; margin-bottom: 5px; display: none;}
.admin-active .drag-handle { display: block; }
.admin-active .draggable-item { border: 2px dashed #ccc; min-height: 50px; }
header[role="banner"] { display:flex; align-items:center; gap: 1rem; padding: 0.5rem 1rem; }
.logout-link { margin-left: auto; }
</style>
<script src="{{ url_for('static', filename='assets/js/index.js') }}"></script>
</body>
</html>
</html>

0
web/tests/__init__.py Normal file
View File

88
web/tests/conftest.py Normal file
View File

@@ -0,0 +1,88 @@
"""
conftest.py — Fixtures pytest partagées.
La suite de tests s'exécute contre une VRAIE base MariaDB de test
(gesthub_test), jamais contre des mocks de couche SQL : cela permet de
détecter de vraies erreurs de requête (typos de colonnes, contraintes
violées...) que des mocks ne détecteraient pas. Chaque test nettoie les
tables qu'il utilise pour rester indépendant des autres (pas d'ordre imposé).
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import pytest
from config import TestConfig
from db import init_db
@pytest.fixture(scope="session")
def db():
database = init_db(TestConfig)
# Sanity check : la base de test doit être joignable avant de lancer la suite.
database.fetch_one("SELECT 1 AS ok")
return database
@pytest.fixture(autouse=True)
def clean_tables(db):
"""Vide les tables avant CHAQUE test pour garantir l'isolation."""
with db.cursor(commit=True) as cur:
for table in ("audit_log", "fichiers", "evenements", "annonces", "blocks"):
cur.execute(f"DELETE FROM {table}")
yield
@pytest.fixture(scope="session")
def app():
# Portée "session" : Flask interdit de redécorer un blueprint déjà
# enregistré (voir routes/auth.py), donc create_app() n'est appelé
# qu'une seule fois pour toute la suite. L'isolation entre tests est
# assurée par clean_tables (autouse) et par un nouveau test_client()
# (voir fixture client ci-dessous) à chaque test.
from app import create_app
flask_app = create_app(TestConfig)
flask_app.config["DB"] = init_db(TestConfig)
flask_app.config["TESTING"] = True
return flask_app
@pytest.fixture
def client(app):
return app.test_client()
@pytest.fixture
def admin_session(app):
"""Simule une session utilisateur authentifiée avec le groupe /admin.
Utilise SON PROPRE test_client() (indépendant de la fixture `client` et
de `user_session`) : deux clients Flask ont chacun leur propre cookie
jar, donc leurs sessions ne s'écrasent pas quand un test dépend des deux
en même temps (cf. test_it07).
"""
test_client = app.test_client()
with test_client.session_transaction() as sess:
sess["user"] = {
"sub": "admin-sub-0001",
"preferred_username": "admin",
"groups": ["/admin"],
}
return test_client
@pytest.fixture
def user_session(app):
"""Simule une session utilisateur authentifiée SANS le groupe /admin."""
test_client = app.test_client()
with test_client.session_transaction() as sess:
sess["user"] = {
"sub": "user-sub-0002",
"preferred_username": "membre",
"groups": [],
}
return test_client

View File

@@ -0,0 +1,73 @@
"""
Tests d'intégration des routes Flask — IT-01 à IT-10 (client de test Flask).
Vérifient le comportement HTTP complet (routes -> services -> models -> DB
réelle), avec et sans authentification/autorisation.
"""
import json
def test_it01_annonces_sans_auth_401(client):
resp = client.get("/api/annonces")
assert resp.status_code == 401
def test_it02_annonces_avec_auth_200(user_session):
resp = user_session.get("/api/annonces")
assert resp.status_code == 200
assert resp.get_json() == []
def test_it03_creer_annonce_non_admin_403(user_session):
resp = user_session.post("/api/annonces", json={"titre": "T", "contenu": "C"})
assert resp.status_code == 403
def test_it04_creer_annonce_admin_201_puis_visible(admin_session):
resp = admin_session.post("/api/annonces", json={"titre": "Annonce test", "contenu": "Contenu"})
assert resp.status_code == 201
resp_list = admin_session.get("/api/annonces")
titres = [a["titre"] for a in resp_list.get_json()]
assert "Annonce test" in titres
def test_it05_modifier_annonce_admin_200(admin_session):
created = admin_session.post("/api/annonces", json={"titre": "A", "contenu": "B"}).get_json()
resp = admin_session.put(f"/api/annonces/{created['id']}", json={"titre": "A2", "contenu": "B2"})
assert resp.status_code == 200
def test_it06_supprimer_annonce_admin_200(admin_session):
created = admin_session.post("/api/annonces", json={"titre": "A", "contenu": "B"}).get_json()
resp = admin_session.delete(f"/api/annonces/{created['id']}")
assert resp.status_code == 200
def test_it07_is_admin_reflete_groupes(admin_session, user_session):
assert admin_session.get("/api/is_admin").get_json()["admin"] is True
assert user_session.get("/api/is_admin").get_json()["admin"] is False
def test_it08_upload_sans_fichier_400(user_session):
resp = user_session.post("/api/fichiers/upload", data={})
assert resp.status_code == 400
def test_it09_evenement_dates_invalides_400(admin_session):
resp = admin_session.post(
"/api/evenements",
json={"titre": "Réunion", "date_debut": "2026-09-10T10:00:00", "date_fin": "2026-09-09T10:00:00"},
)
assert resp.status_code == 400
def test_it10_accueil_sans_session_redirige_login(client):
resp = client.get("/", follow_redirects=False)
assert resp.status_code == 302
assert "/login" in resp.headers["Location"]
def test_it04b_creer_annonce_titre_manquant_400(admin_session):
resp = admin_session.post("/api/annonces", json={"contenu": "Sans titre"})
assert resp.status_code == 400

97
web/tests/test_rgpd.py Normal file
View File

@@ -0,0 +1,97 @@
"""
Tests RGPD — droit à l'effacement (art. 17) : anonymisation d'un `sub`
OIDC dans toutes les tables qui le référencent (routes/rgpd.py,
services/rgpd_service.py, models/rgpd_model.py).
"""
from models import annonce_model, fichier_model, evenement_model, audit_model
from services import annonce_service, evenement_service, rgpd_service
def _seed_user_data(db, sub):
"""Crée une ligne dans chaque table référençant `sub`, y compris
audit_log (via les services, comme le ferait l'application réelle).
create_annonce() et create_evenement() journalisent déjà chacun une
ligne dans audit_log (CREATE_ANNONCE, CREATE_EVENEMENT) ; on ajoute un
LOGIN explicite pour obtenir 3 lignes audit_log au total pour ce sub.
"""
annonce_id = annonce_service.create_annonce(db, "Titre", "Contenu", sub)
fichier_model.insert(db, "doc.pdf", "uuid-stocke", "application/pdf", 10, sub)
evenement_service.create_evenement(
db, "Réunion", "desc", "2026-09-10T10:00:00", "2026-09-10T11:00:00", sub
)
from services import audit_service
audit_service.log(db, "LOGIN", None, "SUCCESS", sub, "127.0.0.1")
return annonce_id
# --- Accès -------------------------------------------------------------------
def test_rgpd01_sans_auth_401(client):
resp = client.delete("/api/admin/users/sub-cible/data")
assert resp.status_code == 401
def test_rgpd02_non_admin_403(user_session):
resp = user_session.delete("/api/admin/users/sub-cible/data")
assert resp.status_code == 403
def test_rgpd03_sub_inconnu_404(admin_session):
resp = admin_session.delete("/api/admin/users/sub-jamais-vu/data")
assert resp.status_code == 404
# --- Anonymisation effective ---------------------------------------------
def test_rgpd04_anonymisation_remplace_le_sub_partout(db, admin_session):
target_sub = "sub-a-effacer-0001"
annonce_id = _seed_user_data(db, target_sub)
resp = admin_session.delete(f"/api/admin/users/{target_sub}/data")
assert resp.status_code == 200
body = resp.get_json()
assert body["status"] == "anonymized"
pseudonym = body["pseudonym"]
assert pseudonym.startswith("anon:")
assert body["rows_updated"] == {
"annonces": 1, "fichiers": 1, "evenements": 1, "audit_log": 3,
}
# Le sub d'origine n'apparaît plus nulle part...
assert annonce_model.get(db, annonce_id)["auteur_sub"] == pseudonym
fichiers = fichier_model.fetch_all(db)
assert fichiers[0]["uploader_sub"] == pseudonym
evenements = evenement_model.fetch_all(db)
assert evenements[0]["createur_sub"] == pseudonym
recent = audit_model.fetch_recent(db)
subs_restants = [row["utilisateur_sub"] for row in recent]
assert target_sub not in subs_restants
# ... mais une entrée d'audit trace l'action d'anonymisation elle-même,
# journalisée sous l'identité de l'ADMIN qui l'a déclenchée (pas la
# cible, déjà pseudonymisée à ce stade).
actions = {row["action"] for row in recent}
assert "RGPD_ANONYMIZE_USER" in actions
def test_rgpd05_pseudonyme_deterministe_pour_un_meme_sub(db):
target_sub = "sub-deterministe-0002"
_seed_user_data(db, target_sub)
result = rgpd_service.anonymize_user(db, target_sub, pepper="pepper-de-test")
assert result["anonymized"] is True
# Toutes les tables ont reçu EXACTEMENT le même pseudonyme.
pseudo = result["pseudonym"]
assert annonce_model.fetch_all(db)[0]["auteur_sub"] == pseudo
assert fichier_model.fetch_all(db)[0]["uploader_sub"] == pseudo
def test_rgpd06_deja_anonymise_refuse(db, admin_session):
target_sub = "sub-double-appel-0003"
_seed_user_data(db, target_sub)
first = admin_session.delete(f"/api/admin/users/{target_sub}/data")
pseudonym = first.get_json()["pseudonym"]
second = admin_session.delete(f"/api/admin/users/{pseudonym}/data")
assert second.status_code == 400

View File

@@ -0,0 +1,89 @@
"""
Tests de sécurité — TS-01 à TS-06 (Bloc 1 - C8 ; OWASP Top 10 2021).
Approche adversariale : se mettre dans la peau d'un attaquant pour chaque
vecteur identifié dans le tableau de risques du dossier technique.
"""
import io
import os
import yaml
from werkzeug.datastructures import FileStorage
from models import annonce_model
from services import annonce_service, fichier_service
# --- TS-01 : Injection SQL (A03:2021) ----------------------------------------
def test_ts01_injection_sql_dans_titre_est_neutralisee(db):
payload = "Test'; DROP TABLE annonces; --"
new_id = annonce_service.create_annonce(db, payload, "contenu", "sub-attaquant")
# La requête paramétrée traite le payload comme une simple chaîne : la
# table annonces existe toujours et contient bien la ligne créée.
row = annonce_model.get(db, new_id)
assert row["titre"] == payload
all_rows = annonce_model.fetch_all(db)
assert len(all_rows) >= 1 # la table n'a pas été supprimée
# --- TS-02 : Accès sans token (A01:2021 — Broken Access Control) ------------
def test_ts02_acces_sans_token_refuse(client):
resp = client.get("/api/fichiers")
assert resp.status_code == 401
resp2 = client.post("/api/annonces", json={"titre": "x", "contenu": "y"})
assert resp2.status_code == 401
# --- TS-03 : Cookie de session falsifié (token forgé) -----------------------
def test_ts03_cookie_session_falsifie_refuse(client):
client.set_cookie("session", "valeur-totalement-forgee-non-signee")
resp = client.get("/api/annonces")
# Flask ne parvient pas à décoder un cookie non signé par SECRET_KEY :
# la session est traitée comme vide -> 401, pas d'accès accordé.
assert resp.status_code == 401
# --- TS-04 : XSS stocké (A03:2021) ------------------------------------------
def test_ts04_contenu_avec_script_est_echappe_au_rendu(app, db):
payload = "<script>alert('xss')</script>"
annonce_service.create_annonce(db, "Titre", payload, "sub-1")
with app.test_request_context():
from flask import render_template_string
rendered = render_template_string("{{ contenu }}", contenu=payload)
# Jinja2 (autoescape actif par défaut sur .html) transforme les balises
# en entités HTML : le script n'est jamais exécutable tel quel.
assert "<script>" not in rendered
assert "&lt;script&gt;" in rendered
# --- TS-05 : Accès aux ressources d'autrui (A01:2021) -----------------------
def test_ts05_telechargement_fichier_dautrui_refuse(db, tmp_path, monkeypatch):
from config import Config
monkeypatch.setattr(Config, "UPLOAD_DIR", str(tmp_path))
fs = FileStorage(stream=io.BytesIO(b"contenu prive"), filename="prive.pdf", content_type="application/pdf")
fichier_id = fichier_service.handle_upload(db, fs, "sub-proprietaire", ip="127.0.0.1")
try:
fichier_service.prepare_download(db, fichier_id, "sub-attaquant", is_admin=False)
assert False, "le téléchargement aurait dû être refusé"
except PermissionError:
pass
# Le propriétaire, lui, peut télécharger son propre fichier.
path, name = fichier_service.prepare_download(db, fichier_id, "sub-proprietaire", is_admin=False)
assert os.path.exists(path)
# --- TS-06 : Exposition des ports internes (A05:2021 — Security Misconfig) -
def test_ts06_seul_caddy_expose_des_ports_publics():
compose_path = os.path.join(os.path.dirname(__file__), "..", "..", "docker-compose.yml")
with open(compose_path) as f:
compose = yaml.safe_load(f)
for service_name, service_def in compose["services"].items():
if service_name == "caddy":
assert "ports" in service_def
else:
assert "ports" not in service_def, f"{service_name} expose un port public !"

View File

@@ -0,0 +1,41 @@
"""Tests unitaires du service Annonces — UT-01 à UT-05 (Bloc 3 - C7)."""
import pytest
from models import annonce_model
from services import annonce_service
def test_ut01_create_annonce_valide(db):
new_id = annonce_service.create_annonce(db, "Titre test", "Contenu test", "sub-1")
assert new_id is not None
row = annonce_model.get(db, new_id)
assert row["titre"] == "Titre test"
assert row["contenu"] == "Contenu test"
def test_ut02_create_annonce_titre_vide(db):
with pytest.raises(annonce_service.ValidationError):
annonce_service.create_annonce(db, "", "Contenu", "sub-1")
def test_ut03_create_annonce_contenu_vide(db):
with pytest.raises(annonce_service.ValidationError):
annonce_service.create_annonce(db, "Titre", " ", "sub-1")
def test_ut04_update_annonce_existante(db):
new_id = annonce_service.create_annonce(db, "Avant", "Avant contenu", "sub-1")
annonce_service.update_annonce(db, new_id, "Après", "Après contenu", "sub-1")
row = annonce_model.get(db, new_id)
assert row["titre"] == "Après"
def test_ut05_update_annonce_inexistante(db):
with pytest.raises(annonce_model.AnnonceNotFoundError):
annonce_service.update_annonce(db, 999999, "Titre", "Contenu", "sub-1")
def test_ut05b_delete_annonce_inexistante(db):
with pytest.raises(annonce_model.AnnonceNotFoundError):
annonce_service.delete_annonce(db, 999999, "sub-1")

View File

@@ -0,0 +1,22 @@
"""Tests unitaires du service d'authentification — UT-09 à UT-12."""
from services.auth_service import is_admin
def test_ut09_is_admin_avec_groupe_admin():
user = {"sub": "1", "groups": ["/admin", "/staff"]}
assert is_admin(user) is True
def test_ut10_is_admin_sans_groupe_admin():
user = {"sub": "1", "groups": ["/staff"]}
assert is_admin(user) is False
def test_ut11_is_admin_utilisateur_none():
assert is_admin(None) is False
def test_ut12_is_admin_sans_champ_groups():
user = {"sub": "1"}
assert is_admin(user) is False

View File

@@ -0,0 +1,56 @@
"""Tests unitaires du module Fichiers — UT-06 à UT-08 (décomposition en 8 étapes)."""
import io
import pytest
from werkzeug.datastructures import FileStorage
from services import fichier_service
def _file_storage(content=b"contenu de test", filename="rapport.pdf", mimetype="application/pdf"):
return FileStorage(stream=io.BytesIO(content), filename=filename, content_type=mimetype)
def test_ut06_fichier_trop_grand():
fs = _file_storage(content=b"x" * 100)
with pytest.raises(fichier_service.FileValidationError):
fichier_service.step2_check_size(fs, max_size=50)
def test_ut06b_fichier_vide():
fs = _file_storage(content=b"")
with pytest.raises(fichier_service.FileValidationError):
fichier_service.step2_check_size(fs)
def test_ut07_extension_interdite():
with pytest.raises(fichier_service.FileValidationError):
fichier_service.step4_check_extension("script.exe")
def test_ut07b_extension_autorisee():
ext = fichier_service.step4_check_extension("rapport.pdf")
assert ext == "pdf"
def test_ut08_nom_genere_est_unique():
name1 = fichier_service.step5_generate_safe_name("pdf")
name2 = fichier_service.step5_generate_safe_name("pdf")
assert name1 != name2
assert name1.endswith(".pdf")
def test_ut08b_presence_fichier_absent():
with pytest.raises(fichier_service.FileValidationError):
fichier_service.step1_check_presence(None)
def test_upload_complet_end_to_end(db, tmp_path, monkeypatch):
"""Vérifie l'enchaînement complet des 8 étapes avec un fichier valide."""
from config import Config
monkeypatch.setattr(Config, "UPLOAD_DIR", str(tmp_path))
fs = _file_storage(content=b"%PDF-1.4 contenu factice", filename="doc.pdf", mimetype="application/pdf")
fichier_id = fichier_service.handle_upload(db, fs, "sub-1", ip="127.0.0.1")
assert fichier_id is not None