185 lines
No EOL
7.2 KiB
Python
185 lines
No EOL
7.2 KiB
Python
import discord
|
|
from discord.ext import commands
|
|
import json
|
|
import os
|
|
from collections import deque
|
|
from datetime import datetime, timedelta
|
|
|
|
# Chemins des fichiers
|
|
SETTINGS_FILE = "data/security_settings.json"
|
|
WHITELIST_FILE = "data/whitelist.json"
|
|
|
|
# Cache TTL en secondes (5 secondes pour avoir une réactivité acceptable)
|
|
CACHE_TTL = 5
|
|
|
|
# Triggers pour détection de liens
|
|
LINK_TRIGGERS = ("discord.gg/", "http://", "https://", "www.")
|
|
|
|
class AntiSpam(commands.Cog):
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
# Optimisation: deque avec maxlen pour O(1) au lieu de O(n) list filtering
|
|
self.message_counts = {}
|
|
self.spam_warnings = {}
|
|
self.link_warnings = {}
|
|
# Cache avec timestamps pour éviter la lecture fichier à chaque message
|
|
self._settings_cache = {} # {guild_id: (timestamp, settings_dict)}
|
|
self._whitelist_cache = {} # {guild_id: (timestamp, whitelist_set)}
|
|
|
|
def _get_cached_settings(self, guild_id):
|
|
"""Récupère les paramètres avec cache (TTL de 5 secondes)"""
|
|
now = discord.utils.utcnow().timestamp()
|
|
cached = self._settings_cache.get(guild_id)
|
|
|
|
if cached:
|
|
cached_time, settings = cached
|
|
if now - cached_time < CACHE_TTL:
|
|
return settings
|
|
|
|
# Cache expiré ou absent - lire le fichier
|
|
try:
|
|
if os.path.exists(SETTINGS_FILE):
|
|
with open(SETTINGS_FILE, "r", encoding='utf-8') as f:
|
|
all_settings = json.load(f)
|
|
settings = all_settings.get(str(guild_id), {})
|
|
else:
|
|
settings = {}
|
|
except:
|
|
settings = {}
|
|
|
|
self._settings_cache[guild_id] = (now, settings)
|
|
return settings
|
|
|
|
def _get_cached_whitelist(self, guild_id):
|
|
"""Récupère la whitelist avec cache (TTL de 5 secondes)"""
|
|
now = discord.utils.utcnow().timestamp()
|
|
cached = self._whitelist_cache.get(guild_id)
|
|
|
|
if cached:
|
|
cached_time, whitelist_set = cached
|
|
if now - cached_time < CACHE_TTL:
|
|
return whitelist_set
|
|
|
|
# Cache expiré ou absent - lire le fichier
|
|
whitelist_set = set()
|
|
try:
|
|
if os.path.exists(WHITELIST_FILE):
|
|
with open(WHITELIST_FILE, "r", encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
server_id_str = str(guild_id)
|
|
whitelist_ids = data.get(server_id_str, [])
|
|
whitelist_set = set(str(uid) for uid in whitelist_ids)
|
|
except:
|
|
pass
|
|
|
|
self._whitelist_cache[guild_id] = (now, whitelist_set)
|
|
return whitelist_set
|
|
|
|
def is_whitelisted(self, guild_id, user_id):
|
|
"""Vérifie l'immunité (Propriétaire ou Whitelist JSON avec cache)"""
|
|
guild = self.bot.get_guild(guild_id)
|
|
if guild and user_id == guild.owner_id:
|
|
return True
|
|
|
|
# Utilisation du cache whitelist
|
|
whitelist = self._get_cached_whitelist(guild_id)
|
|
return str(user_id) in whitelist
|
|
|
|
@commands.Cog.listener()
|
|
async def on_message(self, message):
|
|
# 1. Filtres de base (early return pour performance)
|
|
if message.author.bot or not message.guild:
|
|
return
|
|
|
|
guild_id = message.guild.id
|
|
user_id = message.author.id
|
|
|
|
# 2. VÉRIFICATION WHITELIST (Immunité totale)
|
|
if self.is_whitelisted(guild_id, user_id):
|
|
return
|
|
|
|
# --- LECTURE CACHÉ (dynamique mais rapide) ---
|
|
settings = self._get_cached_settings(guild_id)
|
|
|
|
# --- TEST ANTI-LIENS ---
|
|
# Bloque tout le monde sauf whitelist ET manage_messages
|
|
anti_link_enabled = settings.get("anti_link", False) if settings else False
|
|
has_perm = message.author.guild_permissions.manage_messages
|
|
if anti_link_enabled and not has_perm:
|
|
content_lower = message.content.lower()
|
|
if any(trigger in content_lower for trigger in LINK_TRIGGERS):
|
|
await self.handle_violation(message, "link")
|
|
return
|
|
|
|
# Sortir si pas de settings
|
|
if not settings:
|
|
return
|
|
|
|
# --- TEST ANTI-SPAM (deque pour O(1)) ---
|
|
if settings.get("anti_spam") is True:
|
|
now = discord.utils.utcnow().timestamp()
|
|
|
|
# Initialiser ou récupérer le deque pour cet utilisateur
|
|
if user_id not in self.message_counts:
|
|
# maxlen=10 suffit pour la fenêtre de 5 secondes avec seuil typique
|
|
self.message_counts[user_id] = deque(maxlen=10)
|
|
|
|
user_deque = self.message_counts[user_id]
|
|
user_deque.append(now)
|
|
|
|
# Nettoyer les timestamps expirés (plus vieux que 5 secondes)
|
|
# Le deque supprime automatiquement les anciens éléments quand il dépasse maxlen
|
|
# Mais on doit aussi filtrer par temps pour être précis
|
|
while user_deque and now - user_deque[0] >= 5:
|
|
user_deque.popleft()
|
|
|
|
threshold = settings.get("spam_threshold", 5)
|
|
if len(user_deque) > threshold:
|
|
await self.handle_violation(message, "spam")
|
|
|
|
async def handle_violation(self, message, type_violation):
|
|
user_id = message.author.id
|
|
|
|
if type_violation == "link":
|
|
try: await message.delete()
|
|
except: pass
|
|
self.link_warnings[user_id] = self.link_warnings.get(user_id, 0) + 1
|
|
current_warns = self.link_warnings[user_id]
|
|
reason = "Envoi de liens interdits"
|
|
else:
|
|
try: await message.channel.purge(limit=5, check=lambda m: m.author == message.author)
|
|
except: pass
|
|
self.spam_warnings[user_id] = self.spam_warnings.get(user_id, 0) + 1
|
|
current_warns = self.spam_warnings[user_id]
|
|
reason = "Spam excessif"
|
|
# Réinitialiser le deque après spam détecté
|
|
self.message_counts[user_id] = deque(maxlen=10)
|
|
|
|
if current_warns <= 3:
|
|
embed = discord.Embed(
|
|
title="🛡️ Sécurité Kuby",
|
|
description=f"{message.author.mention}, {reason}.\n**Avertissement : {current_warns}/3**",
|
|
color=0xFFA500
|
|
)
|
|
await message.channel.send(embed=embed, delete_after=5)
|
|
else:
|
|
await self.apply_mute(message, reason)
|
|
if type_violation == "link": self.link_warnings[user_id] = 0
|
|
else: self.spam_warnings[user_id] = 0
|
|
|
|
async def apply_mute(self, message, reason):
|
|
try:
|
|
until = discord.utils.utcnow() + timedelta(minutes=5)
|
|
await message.author.timeout(until, reason=f"Kuby Security: {reason} (3 Warns)")
|
|
|
|
embed = discord.Embed(
|
|
title="⛔ Exclusion Temporaire",
|
|
description=f"**Membre :** {message.author.mention}\n**Durée :** 5 minutes\n**Raison :** {reason} répétitif.",
|
|
color=0xFF0000
|
|
)
|
|
await message.channel.send(embed=embed)
|
|
except Exception as e:
|
|
print(f"Erreur lors de la sanction : {e}")
|
|
|
|
async def setup(bot):
|
|
await bot.add_cog(AntiSpam(bot)) |