Public Access
109 lines
5.5 KiB
Python
109 lines
5.5 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"
|
|
|
|
# NOUVEAU PROMPT : Immunité absolue pour les jeux vidéo et la fiction
|
|
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 et réellement illégales (pédopornographie, vente de drogues réelles, terrorisme dans le monde réel, discours de haine).
|
|
RÈGLE ANTI-ZÈLE ET IMMUNITÉ GAMING : Tu dois OBLIGATOIREMENT laisser passer et considérer comme LÉGAL tout ce qui relève des jeux vidéo (comme Minecraft, les mods type Draconic Evolution ou Applied Energistics, KubeJS, les serveurs), de la fiction, du roleplay, ou des questions juridiques. La construction de "réacteurs", de "bombes", ou de "pièges" dans un contexte de jeu vidéo est TOTALEMENT AUTORISÉE.
|
|
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 affiné pour ne pas bloquer les mots courants du jeu vidéo
|
|
mots_interdits = ["pédopornographie", "attentat terroriste réel", "vente de drogue illicite"]
|
|
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}") |