161 lines
6.7 KiB
Python
161 lines
6.7 KiB
Python
"""
|
|
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)
|