import os # Nettoyer les variables de certificats SSL corrompues ou obsolètes pour éviter les crashs TLS os.environ.pop("REQUESTS_CA_BUNDLE", None) os.environ.pop("SSL_CERT_FILE", None) import sys from dotenv import load_dotenv # Charger les variables d'environnement AVANT tout le reste load_dotenv() import disnake from commandes.ticket.utils.permissions import can_access_config from commandes.ticket.data.storage import get_storage from disnake.ext import commands import asyncio from src.logger import kuby_logger, log_performance import logging from src.advanced_logger import AdvancedLogger SENSITIVE_PERMISSIONS = { "administrator", "manage_guild", "manage_roles", "manage_channels", "kick_members", "ban_members", "manage_messages", "manage_webhooks", "moderate_members" } class RoleApprovalView(disnake.ui.View): def __init__(self, member: disnake.Member, roles: list, owner: disnake.Member): super().__init__(timeout=86400) # 24 hours self.member = member self.roles = roles self.owner = owner @disnake.ui.button(label="Accepter", style=disnake.ButtonStyle.green, emoji="✅") async def accept(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction): guild = self.member.guild current_member = guild.get_member(self.member.id) if not current_member: await interaction.response.send_message("❌ L'utilisateur n'est plus sur le serveur.", ephemeral=True) self.stop() return try: await current_member.add_roles(*self.roles, reason="Restauration sécurisée approuvée par le propriétaire") await interaction.response.send_message(f"✅ Rôles sensibles restaurés pour {self.member.mention}.", ephemeral=True) components = [ disnake.ui.Container( disnake.ui.TextDisplay(f"✅ Vous avez approuvé la restauration des rôles pour {self.member.mention}."), disnake.ui.TextDisplay(f"🛡️ Rôles : {', '.join([r.name for r in self.roles])}") ) ] await interaction.edit_original_response(embed=None, components=components, view=None) self.stop() except Exception as e: await interaction.response.send_message(f"❌ Erreur lors de l'attribution : {e}", ephemeral=True) @disnake.ui.button(label="Refuser", style=disnake.ButtonStyle.red, emoji="✖️") async def deny(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction): await interaction.response.send_message(f"❌ Vous avez refusé la restauration pour {self.member.mention}.", ephemeral=True) components = [ disnake.ui.Container( disnake.ui.TextDisplay(f"❌ Vous avez refusé la restauration des rôles pour {self.member.mention}."), disnake.ui.TextDisplay(f"🛡️ Rôles : {', '.join([r.name for r in self.roles])}") ) ] await interaction.edit_original_response(embed=None, components=components, view=None) self.stop() # Récupération et validation de l'ID d'application application_id_raw = os.getenv("APPLICATION_ID") kuby_logger.debug(f"Raw APPLICATION_ID: {application_id_raw}") try: application_id = int(application_id_raw) if application_id_raw else None kuby_logger.info(f"✅ APPLICATION_ID loaded successfully: {application_id}") except ValueError: kuby_logger.error(f"❌ APPLICATION_ID invalide : {application_id_raw}") application_id = None # Configuration des intents Disnake intents = disnake.Intents.default() intents.message_content = True intents.members = True intents.voice_states = True kuby_logger.debug(f"Disnake intents configured: {intents}") class MyBot(commands.Bot): def __init__(self): kuby_logger.debug("Initializing MyBot class") super().__init__( command_prefix="!", intents=intents, application_id=application_id, command_sync_flags=commands.CommandSyncFlags.all() ) kuby_logger.info("MyBot instance created successfully") # Initialize advanced logger self.advanced_logger = AdvancedLogger(self) kuby_logger.info("✅ Advanced logger initialized") # Flask est géré par PM2 (kuby-flask), pas besoin de le lancer ici async def do_async_setup(self): kuby_logger.info("🚀 Configuration initiale du bot Kuby...") extensions = [ "commandes.whitelist", "commandes.whitelist_monitor", "commandes.sendbotmessages", "commandes.goodbye", "commandes.welcome", "commandes.security", "commandes.snipe", "commandes.scan", "commandes.no_link", "commandes.blacklist", "commandes.ticket", "commandes.modules_security.antispam", "commandes.modules_security.antiraid", "commandes.modules_security.logs_manager", "commandes.modules_security.rules", "commandes.ticket_whitelist", "commandes.bug_report", "commandes.convocation", "commandes.absence_staff", "commandes.moderation", "commandes.invites", ] for extension in extensions: try: self.load_extension(extension) kuby_logger.info(f"✅ Extension chargée : {extension}") except Exception as e: kuby_logger.error(f"❌ Erreur lors du chargement de {extension} : {e}", exc_info=True) # --- LOGGING DES INTERACTIONS --- @self.listen("on_interaction") async def log_interaction(inter: disnake.Interaction): if inter.type == disnake.InteractionType.application_command: kuby_logger.info(f"⚡ Interaction reçue : /{inter.data.name} par {inter.user}") kuby_logger.info("✅ Configuration terminée") async def on_ready(self): kuby_logger.info(f"🤖 Bot connecté : {self.user} (ID : {self.user.id})") kuby_logger.info(f"📊 Serveurs : {len(self.guilds)}") kuby_logger.info(f"🔍 Intents Status - Members: {self.intents.members}, Presences: {self.intents.presences}, Core: {self.intents.message_content}") if not self.intents.members: kuby_logger.warning("⚠️ L'intent MEMBERS est désactivé !") kuby_logger.info("✅ Bot prêt et opérationnel !") async def on_guild_join(self, guild): kuby_logger.info(f"🆕 Bot joined new guild: {guild.name} (ID: {guild.id})") kuby_logger.audit(f"Bot added to guild: {guild.name}") async def on_guild_remove(self, guild): kuby_logger.info(f"🚪 Bot removed from guild: {guild.name}") kuby_logger.audit(f"Bot removed from guild: {guild.name}") async def on_member_join(self, member): kuby_logger.info(f"👋 {member} joined {member.guild.name}") if hasattr(self, 'advanced_logger'): self.advanced_logger.log_member_join(member) previous_roles = self.advanced_logger.check_previous_roles(member.id) if previous_roles: guild = member.guild roles_to_add = [guild.get_role(rid) for rid in previous_roles.get("role_ids", [])] roles_to_add = [r for r in roles_to_add if r and r < guild.me.top_role] if roles_to_add: safe_roles = [] sensitive_roles = [] for role in roles_to_add: is_sensitive = any(getattr(role.permissions, perm, False) for perm in SENSITIVE_PERMISSIONS) if is_sensitive: sensitive_roles.append(role) else: safe_roles.append(role) if safe_roles: retries = 3 for i in range(retries): try: await member.add_roles(*safe_roles) kuby_logger.info(f"Restored {len(safe_roles)} safe roles for {member}") break except (disnake.HTTPException, disnake.GatewayNotFound, asyncio.TimeoutError) as e: is_transient = isinstance(e, disnake.HTTPException) and e.status in [500, 502, 503, 504] if not isinstance(e, disnake.HTTPException): is_transient = True # 403 with code 50013 = Missing Permissions - don't retry if isinstance(e, disnake.HTTPException) and e.status == 403: error_code = getattr(e, 'code', None) # 50013 = Missing Permissions, 20023 = Cannot modify a role higher than bot's highest role if error_code in (50013, 20023): kuby_logger.warning(f"Missing Manage Roles permission to restore safe roles for {member} in {member.guild.name} - please ensure the bot has the Manage Roles permission") break if is_transient and i < retries - 1: await asyncio.sleep((i + 1) * 2) continue kuby_logger.error(f"Failed to restore safe roles for {member}: {e}") break except Exception as e: kuby_logger.error(f"Error restoring safe roles for {member}: {e}") break if sensitive_roles: try: owner = guild.owner or await guild.fetch_member(guild.owner_id) embed = disnake.Embed( title="⚠️ Restauration de rôles sensibles", description=( f"L'utilisateur **{member}** ({member.id}) vient de rejoindre **{guild.name}**.\n\n" "Certains de ses anciens rôles possèdent des permissions sensibles :\n" f"🛡️ **Rôles en attente :** {', '.join([r.mention for r in sensitive_roles])}\n\n" "Voulez-vous restaurer ces rôles ?" ), color=disnake.Color.yellow() ) embed.set_thumbnail(url=member.display_avatar.url) view = RoleApprovalView(member, sensitive_roles, owner) await owner.send(embed=embed, view=view) kuby_logger.info(f"Sent role approval request for {member} to owner {owner}") except Exception as e: kuby_logger.error(f"Failed to send role approval request to owner: {e}") async def on_member_remove(self, member): kuby_logger.info(f"👋 {member} left {member.guild.name}") if hasattr(self, 'advanced_logger'): self.advanced_logger.log_member_leave(member) try: from commandes.security import load_settings settings = load_settings(member.guild.id) log_channel_id = settings.get("log_channel_id") if log_channel_id: log_channel = member.guild.get_channel(log_channel_id) if log_channel: await self.advanced_logger.send_leave_notification(member, log_channel) except Exception as e: kuby_logger.error(f"Error in leave notification: {e}") async def on_message_delete(self, message): if message.author.bot: return if hasattr(self, 'advanced_logger'): self.advanced_logger.log_message_event(message, "deleted") async def on_message_edit(self, before, after): if before.author.bot or before.content == after.content: return if hasattr(self, 'advanced_logger'): self.advanced_logger.log_message_event(before, "edited", { "before_content": before.content, "after_content": after.content }) async def on_voice_state_update(self, member, before, after): if member.bot: return if before.channel != after.channel and hasattr(self, 'advanced_logger'): if before.channel is None: self.advanced_logger.log_voice_event(member, after.channel, "joined") elif after.channel is None: self.advanced_logger.log_voice_event(member, before.channel, "left") else: self.advanced_logger.log_voice_event(member, before.channel, "moved_from") self.advanced_logger.log_voice_event(member, after.channel, "moved_to") async def on_command_error(self, ctx, error): if isinstance(error, commands.CommandNotFound): return if isinstance(error, commands.CheckFailure): return kuby_logger.error(f"Command error in {ctx.command}: {str(error)}", exc_info=True)