Public Access
108 lines
5.1 KiB
Python
108 lines
5.1 KiB
Python
import os
|
|
import json
|
|
import re
|
|
from datetime import datetime
|
|
from modules.logger import log_event
|
|
from modules.ai_local import interroger_local_ollama
|
|
|
|
# On importe l'alerte mail pour prévenir l'Admin
|
|
try:
|
|
from modules.debug_mailer import envoyer_alerte_debug
|
|
except ImportError:
|
|
envoyer_alerte_debug = None
|
|
|
|
BASE_DATA_DIR = "/DATA/AppData/MULTI-IA-AETHAS38/Utilisateurs"
|
|
|
|
# Liste des Super Admins / Admins à prévenir
|
|
ADMINS_GLOBAUX = ["xavier", "céline"]
|
|
|
|
def analyser_intention_illicite(prompt: str, ai_orchestrateur: str = "qwen2.5-coder:3b") -> tuple[bool, str]:
|
|
"""
|
|
Fait appel à l'IA locale pour analyser l'intention sans faire de zèle.
|
|
Sécurisé contre les valeurs None passées en paramètre.
|
|
"""
|
|
modele_cible = ai_orchestrateur if ai_orchestrateur else "qwen2.5-coder:3b"
|
|
|
|
prompt_juge = f"""Tu es le Juge de Sécurité de la plateforme logicielle Aethas38.
|
|
Ta mission est d'analyser la requête et de détecter UNIQUEMENT les intentions strictement illégales (pédopornographie, drogues, terrorisme, haine).
|
|
RÈGLE ANTI-ZÈLE : Ne bloque QUE l'intention malveillante. Laisse passer les questions juridiques légitimes, la météo, l'aide à l'écriture de fiction ou les infos neutres.
|
|
Renvoie STRICTEMENT un JSON valide : {{"illegal": true ou false, "raison": "Explication très courte ou vide"}}
|
|
Requête à analyser : {prompt}"""
|
|
|
|
try:
|
|
# On interroge le modèle local sécurisé
|
|
reponse, _, _, _ = interroger_local_ollama(prompt_juge, modele_cible)
|
|
|
|
# Extraction du JSON de la réponse
|
|
match = re.search(r'\{.*\}', reponse, re.DOTALL)
|
|
if match:
|
|
decision = json.loads(match.group(0))
|
|
if decision.get("illegal") is True:
|
|
return True, decision.get("raison", "Détection par l'IA locale (Juge Qwen)")
|
|
except Exception as e:
|
|
log_event("WARNING", "MODERATOR", f"Échec de l'analyse IA (Fallback activé) : {e}")
|
|
|
|
# Fallback classique par mots-clés si l'IA locale est indisponible
|
|
mots_interdits = ["pédopornographie", "bombe", "terrorisme"]
|
|
if any(mot in prompt.lower() for mot in mots_interdits):
|
|
return True, "Détection par filtre de mots-clés (Secours)"
|
|
return False, ""
|
|
|
|
def appliquer_sanction_et_archiver(utilisateur_nom: str, prompt: str, raison: str, ip_client: str = "Inconnue") -> str:
|
|
"""
|
|
Gère les avertissements, archive la preuve sur le RAID 1 et alerte les administrateurs.
|
|
"""
|
|
user_dir = os.path.join(BASE_DATA_DIR, utilisateur_nom)
|
|
illegal_dir = os.path.join(user_dir, "Illegal")
|
|
dossier_sanctions = os.path.join(user_dir, "sanctions.json")
|
|
|
|
os.makedirs(illegal_dir, exist_ok=True)
|
|
|
|
strikes = 0
|
|
if os.path.exists(dossier_sanctions):
|
|
try:
|
|
with open(dossier_sanctions, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
strikes = data.get("avertissements", 0)
|
|
except Exception:
|
|
strikes = 0
|
|
|
|
strikes += 1
|
|
sanction_actuelle = "Avertissement"
|
|
if strikes >= 10:
|
|
sanction_actuelle = "BAN DEFINITIF EN ATTENTE DE VALIDATION"
|
|
elif strikes >= 9:
|
|
sanction_actuelle = "Ban 1 mois"
|
|
elif strikes >= 6:
|
|
sanction_actuelle = "Ban 1 semaine"
|
|
elif strikes >= 3:
|
|
sanction_actuelle = "Ban 1 jour"
|
|
|
|
with open(dossier_sanctions, "w", encoding="utf-8") as f:
|
|
json.dump({"avertissements": strikes, "derniere_sanction": sanction_actuelle, "date": str(datetime.now())}, f, indent=4)
|
|
|
|
preuve_path = os.path.join(illegal_dir, f"infraction_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt")
|
|
with open(preuve_path, "w", encoding="utf-8") as f:
|
|
f.write(f"--- INFRACTION REPORTÉE ---\nUtilisateur : {utilisateur_nom}\nIP : {ip_client}\nDate : {datetime.now()}\nRaison : {raison}\n\n[PROMPT INTERCEPTÉ]\n{prompt}")
|
|
|
|
# Exclusion de l'admin s'il est lui-même l'auteur de l'infraction
|
|
admins_a_alerter = [admin for admin in ADMINS_GLOBAUX if admin.lower() != utilisateur_nom.lower()]
|
|
alerte_msg = f"Violation détectée pour {utilisateur_nom} (Avertissement n°{strikes}).\nSanction : {sanction_actuelle}\nPreuve stockée : {preuve_path}"
|
|
log_event("CRITICAL", "MODERATION", alerte_msg)
|
|
|
|
if envoyer_alerte_debug and admins_a_alerter:
|
|
envoyer_alerte_debug("CRITICAL", f"TENTATIVE ILLICITE ({utilisateur_nom})", alerte_msg)
|
|
|
|
return sanction_actuelle
|
|
|
|
def sauvegarder_decision_admin(utilisateur_cible: str, admin_juge: str, decision: str, commentaire: str):
|
|
"""
|
|
Archive la décision prise par un admin avec commentaire libre.
|
|
"""
|
|
dossier = os.path.join(BASE_DATA_DIR, utilisateur_cible, "Illegal")
|
|
os.makedirs(dossier, exist_ok=True)
|
|
fichier = os.path.join(dossier, f"decision_admin_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt")
|
|
|
|
with open(fichier, "w", encoding="utf-8") as f:
|
|
f.write(f"--- DÉCISION ADMINISTRATIVE ---\nCible : {utilisateur_cible}\nJuge Admin : {admin_juge}\nDate : {datetime.now()}\nDécision : {decision}\nCommentaire : {commentaire}\n")
|
|
log_event("INFO", "MODERATION", f"L'admin {admin_juge} a statué sur {utilisateur_cible} : {decision}") |