gestionnaire_specialites.py — v3 (sécurité noms, raison de rejet)

[Mario Da Conceicao]

Version corrigée du 01/08/2026, après une 2e relecture croisée. Correction de sécurité importante : protection contre les noms de spécialités malveillants (traversée de répertoire type "../../"). Ajout : normalisation des noms, raison de rejet archivée.

import json
import re
import shutil
import unicodedata
from pathlib import Path
from datetime import datetime

BASE_DIR = Path(__file__).resolve().parent
SPECIALITES_DIR = BASE_DIR / "specialites"
TELECHARGEMENTS_DIR = BASE_DIR / "telechargements"
ARCHIVES_DIR = BASE_DIR / "archives"
REGISTRE_FILE = SPECIALITES_DIR / "manifest.json"

def valider_nom_specialite(nom):
    """
    CORRECTION DE SÉCURITÉ : normalise et valide un nom de spécialité.
    - Rejette tout nom contenant '..', '/', '\\' (protection contre les
      chemins malveillants type traversée de répertoire).
    - Normalise en minuscules, sans accents, espaces -> tirets, pour éviter
      les doublons déguisés ("Médical" vs "medical").
    Lève ValueError si le nom est invalide.
    """
    if not nom or not isinstance(nom, str):
        raise ValueError("Nom de spécialité invalide (vide).")
    if ".." in nom or "/" in nom or "\\" in nom:
        raise ValueError(f"Nom de spécialité refusé (caractères interdits) : '{nom}'.")

    nom_normalise = unicodedata.normalize("NFKD", nom).encode("ascii", "ignore").decode("ascii")
    nom_normalise = nom_normalise.lower().strip()
    nom_normalise = re.sub(r"\s+", "-", nom_normalise)
    nom_normalise = re.sub(r"[^a-z0-9\-_]", "", nom_normalise)

    if not nom_normalise:
        raise ValueError(f"Nom de spécialité invalide après normalisation : '{nom}'.")

    # Double vérification : le chemin final doit rester SOUS le dossier prévu
    chemin_final = (TELECHARGEMENTS_DIR / nom_normalise).resolve()
    if TELECHARGEMENTS_DIR.resolve() not in chemin_final.parents:
        raise ValueError(f"Nom de spécialité refusé (chemin invalide) : '{nom}'.")

    return nom_normalise

class GestionnaireSpecialites:
    def __init__(self):
        SPECIALITES_DIR.mkdir(parents=True, exist_ok=True)
        TELECHARGEMENTS_DIR.mkdir(parents=True, exist_ok=True)
        ARCHIVES_DIR.mkdir(parents=True, exist_ok=True)
        if not REGISTRE_FILE.exists():
            self._sauvegarder_registre({"specialites": []})

    def _charger_registre(self):
        with open(REGISTRE_FILE, "r", encoding="utf-8") as f:
            return json.load(f)

    def _sauvegarder_registre(self, data):
        with open(REGISTRE_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=4, ensure_ascii=False)

    def proposer_nouvelle_specialite(self, nom, description, documents_sources=None):
        """MODIFIÉ : valide et normalise le nom avant toute chose."""
        try:
            nom = valider_nom_specialite(nom)
        except ValueError as e:
            return f"Refusé : {e}"

        noms_existants = [s["nom"] for s in self.lister_actives()] + \
                          [s["nom"] for s in self.lister_en_attente()]
        if nom in noms_existants:
            return f"Une spécialité nommée '{nom}' existe déjà (active ou en attente)."

        dossier_attente = TELECHARGEMENTS_DIR / nom
        dossier_attente.mkdir(parents=True, exist_ok=True)

        fiche = {
            "nom": nom,
            "type": "specialite_logiciel",
            "version": "1.0",
            "description": description,
            "source_documents": documents_sources or [],
            "date_creation": datetime.now().isoformat(timespec="seconds"),
            "valide_par_utilisateur": False
        }

        with open(dossier_attente / "manifest.json", "w", encoding="utf-8") as f:
            json.dump(fiche, f, indent=4, ensure_ascii=False)

        return f"Nouvelle spécialité '{nom}' préparée dans telechargements/. En attente de ta validation."

    def lister_en_attente(self):
        en_attente = []
        for dossier in TELECHARGEMENTS_DIR.iterdir():
            manifest_path = dossier / "manifest.json"
            if manifest_path.exists():
                with open(manifest_path, "r", encoding="utf-8") as f:
                    en_attente.append(json.load(f))
        return en_attente

    def valider_specialite(self, nom):
        dossier_attente = TELECHARGEMENTS_DIR / nom
        manifest_path = dossier_attente / "manifest.json"
        if not manifest_path.exists():
            return f"Aucune spécialité en attente nommée '{nom}'."

        with open(manifest_path, "r", encoding="utf-8") as f:
            fiche = json.load(f)
        fiche["valide_par_utilisateur"] = True
        fiche["date_validation"] = datetime.now().isoformat(timespec="seconds")

        dossier_final = SPECIALITES_DIR / nom
        shutil.move(str(dossier_attente), str(dossier_final))

        with open(dossier_final / "manifest.json", "w", encoding="utf-8") as f:
            json.dump(fiche, f, indent=4, ensure_ascii=False)

        registre = self._charger_registre()
        registre["specialites"].append({"nom": nom, "description": fiche["description"]})
        self._sauvegarder_registre(registre)

        return f"Spécialité '{nom}' validée et activée."

    def rejeter_specialite(self, nom, raison=""):
        """MODIFIÉ : archive avec une raison de rejet, pour se souvenir pourquoi."""
        dossier_attente = TELECHARGEMENTS_DIR / nom
        if dossier_attente.exists():
            manifest_path = dossier_attente / "manifest.json"
            if manifest_path.exists() and raison:
                with open(manifest_path, "r", encoding="utf-8") as f:
                    fiche = json.load(f)
                fiche["raison_rejet"] = raison
                fiche["date_rejet"] = datetime.now().isoformat(timespec="seconds")
                with open(manifest_path, "w", encoding="utf-8") as f:
                    json.dump(fiche, f, indent=4, ensure_ascii=False)

            dossier_archive = ARCHIVES_DIR / nom
            shutil.move(str(dossier_attente), str(dossier_archive))
            return f"Spécialité '{nom}' rejetée et archivée (récupérable dans archives/)."
        return f"Aucune spécialité en attente nommée '{nom}'."

    def lister_actives(self):
        return self._charger_registre()["specialites"]

    def consulter_specialite(self, nom, mot_cle=None):
        dossier = SPECIALITES_DIR / nom / "documents"
        if not dossier.exists():
            return None
        resultats = []
        for fichier in dossier.glob("*.txt"):
            contenu = fichier.read_text(encoding="utf-8", errors="replace")
            if not mot_cle or mot_cle.lower() in contenu.lower():
                resultats.append({"fichier": fichier.name, "extrait": contenu[:500]})
        return resultats

[Auteur]

Messages