from __future__ import annotations import asyncio import io import json import os import discord from discord import app_commands from discord.ext import commands from config import settings, ENV_PATH from storage import storage import security PLACEHOLDER_TOKENS = { "", "PEGA_AQUI_EL_TOKEN_DEL_BOT", "TU_TOKEN_AQUI", "TOKEN_AQUI", } def validate_token_config() -> None: token = settings.token if token in PLACEHOLDER_TOKENS: raise RuntimeError( f"Falta DISCORD_TOKEN válido en:\n{ENV_PATH}\n" "Abre ese archivo y pega el BOT TOKEN nuevo de Discord Developer Portal." ) # Discord bot tokens normally contain dot-separated sections. # This is only a local sanity check; Discord remains the final validator. if len(token) < 40 or "." not in token: raise RuntimeError( f"DISCORD_TOKEN parece incompleto o no es un Bot Token.\n" f"Archivo leído: {ENV_PATH}\n" f"Longitud detectada: {len(token)} caracteres.\n" "Usa Developer Portal > Applications > NovaShield > Bot > Reset Token." ) class NovaShieldLogChannelSelect(discord.ui.ChannelSelect): def __init__(self, mode: str): self.mode = mode label = { "security": "Selecciona canal de logs", "web": "Selecciona canal de logs web", "announcements": "Selecciona canal de novedades", }.get(mode, "Selecciona canal") super().__init__( placeholder=label, channel_types=[discord.ChannelType.text], min_values=1, max_values=1, ) async def callback(self, interaction: discord.Interaction): if not interaction.guild or not isinstance(interaction.user, discord.Member): return await interaction.response.send_message("❌ Solo disponible dentro del servidor.", ephemeral=True) if not (interaction.user.guild_permissions.administrator or interaction.user.guild_permissions.manage_guild): return await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) channel = self.values[0] key = { "security": "log_channel_id", "web": "web_log_channel_id", "announcements": "announcement_channel_id", }.get(self.mode, "log_channel_id") await storage.update_panel_settings(interaction.guild.id, {key: str(channel.id)}) if self.mode == "security": real_channel = interaction.guild.get_channel(channel.id) if isinstance(real_channel, discord.TextChannel): perms = real_channel.permissions_for(interaction.guild.me) if perms.view_channel and perms.send_messages: embed = discord.Embed( title="✅ NovaShield — Canal actualizado", description=f"Los logs de seguridad ahora se enviarán a {real_channel.mention}.", color=0x57F287, ) await real_channel.send(embed=embed) await interaction.response.send_message( f"✅ { {'security':'Logs de Seguridad','web':'Logs de la Web','announcements':'Canal de Novedades'}.get(self.mode,'Canal') } actualizado a {channel.mention}.", ephemeral=True ) class NovaShieldDiscordConfigView(discord.ui.View): def __init__(self): super().__init__(timeout=300) self.add_item(NovaShieldLogChannelSelect("security")) self.add_item(NovaShieldLogChannelSelect("web")) self.add_item(NovaShieldLogChannelSelect("announcements")) @discord.ui.button(label="Ver configuración", style=discord.ButtonStyle.secondary, emoji="⚙️") async def show_config(self, interaction: discord.Interaction, button: discord.ui.Button): cfg = await storage.settings_for(interaction.guild.id) log_id = str(cfg.get("log_channel_id") or "") web_id = str(cfg.get("web_log_channel_id") or "") log_ch = interaction.guild.get_channel(int(log_id)) if log_id.isdigit() else None web_ch = interaction.guild.get_channel(int(web_id)) if web_id.isdigit() else None embed = discord.Embed(title="🛡️ NovaShield • Configuración actual", color=0x5865F2) embed.add_field(name="Logs Seguridad", value=log_ch.mention if log_ch else "Sin configurar", inline=False) embed.add_field(name="Logs Web", value=web_ch.mention if web_ch else "Sin configurar", inline=False) embed.add_field(name="Backups automáticos", value="Sí" if cfg.get("automatic_backups", True) else "No", inline=True) embed.add_field(name="Frecuencia", value=f"{cfg.get('backup_interval_hours', 6)} h", inline=True) await interaction.response.send_message(embed=embed, ephemeral=True) # ================= NovaShield Internal Configuration ================= class NovaShieldConfigGroup(app_commands.Group): def __init__(self): super().__init__(name="novashield", description="Configuración interna de NovaShield") async def _can_manage(self, interaction: discord.Interaction) -> bool: if not interaction.guild or not isinstance(interaction.user, discord.Member): return False return interaction.user.guild_permissions.administrator or interaction.user.guild_permissions.manage_guild @app_commands.command(name="logs", description="Configura el canal de logs de seguridad.") @app_commands.describe(canal="Canal donde NovaShield enviará los logs") async def logs(self, interaction: discord.Interaction, canal: discord.TextChannel): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return await storage.update_panel_settings(interaction.guild.id, {"log_channel_id": str(canal.id)}) perms = canal.permissions_for(interaction.guild.me) missing = [] if not perms.view_channel: missing.append("Ver canal") if not perms.send_messages: missing.append("Enviar mensajes") if not perms.embed_links: missing.append("Insertar enlaces") if missing: await interaction.response.send_message( "❌ Canal guardado, pero faltan permisos: **" + ", ".join(missing) + "**.", ephemeral=True ) return try: embed = discord.Embed( title="✅ NovaShield Security — Logs funcionando", description=f"Este canal quedó configurado para los logs de **{interaction.guild.name}**.", color=0x57F287, ) await canal.send(embed=embed) await interaction.response.send_message( f"✅ Canal configurado y prueba enviada correctamente a {canal.mention}.", ephemeral=True ) print(f"[NovaShield LOG TEST] OK -> {interaction.guild.name} / #{canal.name}") except Exception as exc: print(f"[NovaShield LOG TEST] ERROR: {type(exc).__name__}: {exc}") await interaction.response.send_message( f"❌ Discord rechazó el envío: `{type(exc).__name__}: {str(exc)[:250]}`", ephemeral=True ) @app_commands.command(name="logs_web", description="Configura el canal para acciones realizadas desde la web.") @app_commands.describe(canal="Canal para registrar acciones del panel") async def logs_web(self, interaction: discord.Interaction, canal: discord.TextChannel): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return await storage.update_panel_settings(interaction.guild.id, {"web_log_channel_id": str(canal.id)}) await interaction.response.send_message(f"✅ Canal de logs web configurado: {canal.mention}", ephemeral=True) @app_commands.command(name="canal_novedades", description="Configura el canal donde NovaShield publicará novedades.") @app_commands.describe(canal="Canal donde se publicarán actualizaciones y novedades") async def canal_novedades(self, interaction: discord.Interaction, canal: discord.TextChannel): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return await storage.update_panel_settings(interaction.guild.id, {"announcement_channel_id": str(canal.id)}) await interaction.response.send_message(f"✅ Canal de novedades configurado: {canal.mention}", ephemeral=True) try: await canal.send("📢 **NovaShield Security — Canal de novedades configurado**\nLas próximas actualizaciones podrán publicarse aquí directamente desde el panel.") except Exception: pass @app_commands.command(name="anuncio", description="Publica una novedad de NovaShield en el canal configurado.") @app_commands.describe(titulo="Título del anuncio", mensaje="Contenido de la novedad") async def anuncio(self, interaction: discord.Interaction, titulo: str, mensaje: str): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return cfg = await storage.settings_for(interaction.guild.id) cid = str(cfg.get("announcement_channel_id") or "") channel = interaction.guild.get_channel(int(cid)) if cid.isdigit() else None if not isinstance(channel, discord.TextChannel): await interaction.response.send_message("❌ Primero configura el canal con `/novashield canal_novedades`.", ephemeral=True) return embed = discord.Embed( title=f"📢 {titulo[:200]}", description=mensaje[:4000], color=0x5865F2, ) embed.set_footer(text="NovaShield Security • Actualizaciones") try: await channel.send(embed=embed) await interaction.response.send_message(f"✅ Novedad publicada en {channel.mention}.", ephemeral=True) except Exception as exc: await interaction.response.send_message(f"❌ No se pudo publicar: `{str(exc)[:250]}`", ephemeral=True) @app_commands.command(name="backup_auto", description="Configura los backups automáticos.") @app_commands.describe(horas="Frecuencia en horas (1-24)", conservar="Cantidad de backups a conservar (1-100)") async def backup_auto(self, interaction: discord.Interaction, horas: app_commands.Range[int,1,24], conservar: app_commands.Range[int,1,100]=30): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return await storage.update_panel_settings(interaction.guild.id, { "automatic_backups": True, "backup_interval_hours": int(horas), "keep_backups": int(conservar) }) await interaction.response.send_message( f"✅ Backups automáticos activados cada **{horas} h**. Se conservarán **{conservar}**.", ephemeral=True ) @app_commands.command(name="backup_off", description="Desactiva los backups automáticos.") async def backup_off(self, interaction: discord.Interaction): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return await storage.update_panel_settings(interaction.guild.id, {"automatic_backups": False}) await interaction.response.send_message("✅ Backups automáticos desactivados.", ephemeral=True) @app_commands.command(name="panel", description="Abre el panel interno de NovaShield dentro de Discord.") async def panel(self, interaction: discord.Interaction): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return embed = discord.Embed( title="🛡️ NovaShield Security — Configuración", description=( "Configura NovaShield directamente aquí en Discord.\n\n" "Usa el **primer selector** para Logs de Seguridad y el **segundo selector** para Logs de la Web.\n" "Puedes cambiar de canal todas las veces que quieras." ), color=0x5865F2 ) await interaction.response.send_message(embed=embed, view=NovaShieldDiscordConfigView(), ephemeral=True) @app_commands.command( name="ultimo_backup", description="Envía el último backup como archivo JSON descargable." ) async def ultimo_backup(self, interaction: discord.Interaction): if not await self._can_manage(interaction): await interaction.response.send_message( "❌ Necesitas Administrar servidor o Administrador.", ephemeral=True ) return rows = await storage.backups_for(interaction.guild.id, 1) if not rows: await interaction.response.send_message( "❌ Este servidor todavía no tiene backups.", ephemeral=True ) return row = rows[0] portable = { "novashield_backup": True, "format_version": 1, "backup_id": row.get("id"), "guild_id": str(interaction.guild.id), "guild_name": interaction.guild.name, "reason": row.get("reason", ""), "created_by": row.get("created_by", ""), "created_at": row.get("created_at", ""), "backup": row.get("backup", {}), } raw = json.dumps(portable, ensure_ascii=False, indent=2).encode("utf-8") filename = f"NovaShield_{interaction.guild.id}_backup_{row.get('id')}.json" await interaction.response.send_message( f"💾 **Último backup de {interaction.guild.name}**\n" f"Backup **#{row.get('id')}** · `{row.get('created_at', '')}`\n" "Guarda este archivo. Puedes importarlo desde **Web → Backups → Importar backup**.", file=discord.File(io.BytesIO(raw), filename=filename), ephemeral=False ) @app_commands.command(name="emergencia", description="Activa inmediatamente el protocolo completo de emergencia.") async def emergencia(self, interaction: discord.Interaction): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return engine = getattr(interaction.client, "ns_security_engine", None) if engine is None: await interaction.response.send_message("❌ Motor de seguridad no disponible.", ephemeral=True) return await interaction.response.defer(ephemeral=True, thinking=True) try: result = await engine.full_lockdown(interaction.guild, str(interaction.user)) await engine.send_simple_panel_log( interaction.guild, f"🚨 **SERVIDOR BAJO ATAQUE** — protocolo de emergencia activado por **{interaction.user}**." ) await engine.alert_owner_critical( interaction.guild, "LOCKDOWN DE EMERGENCIA ACTIVADO", f"El protocolo completo fue activado desde Discord por **{interaction.user}**." ) await interaction.followup.send( f"🚨 Emergencia activada. Backup previo: **#{result.get('backup_id')}**. " f"Usuarios afectados: **{result.get('users_affected',0)}**.", ephemeral=True ) except Exception as exc: await interaction.followup.send(f"❌ No se pudo activar: `{str(exc)[:300]}`", ephemeral=True) @app_commands.command(name="mantenimiento", description="Activa o desactiva el Modo Mantenimiento de Anti-Nuke.") @app_commands.describe(activar="True para activar; False para volver a proteger normalmente.") async def mantenimiento(self, interaction: discord.Interaction, activar: bool): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return await storage.update_panel_settings(interaction.guild.id, {"maintenance_mode": activar}) engine = getattr(interaction.client, "ns_security_engine", None) if engine: await engine.send_simple_panel_log( interaction.guild, f"🛠️ **Modo Mantenimiento {'ACTIVADO' if activar else 'DESACTIVADO'}** por **{interaction.user}**." ) await interaction.response.send_message( f"🛠️ Modo Mantenimiento **{'ACTIVADO' if activar else 'DESACTIVADO'}**.", ephemeral=True ) @app_commands.command(name="estado", description="Muestra la configuración actual de NovaShield.") async def estado(self, interaction: discord.Interaction): if not await self._can_manage(interaction): await interaction.response.send_message("❌ Necesitas Administrar servidor o Administrador.", ephemeral=True) return cfg=await storage.settings_for(interaction.guild.id) log_id=str(cfg.get("log_channel_id") or "") web_id=str(cfg.get("web_log_channel_id") or "") log_ch=interaction.guild.get_channel(int(log_id)) if log_id.isdigit() else None web_ch=interaction.guild.get_channel(int(web_id)) if web_id.isdigit() else None embed=discord.Embed(title="🛡️ NovaShield • Configuración", color=0x5865F2) embed.add_field(name="Logs seguridad", value=log_ch.mention if log_ch else "Sin configurar", inline=False) embed.add_field(name="Logs web", value=web_ch.mention if web_ch else "Mismo canal / Sin configurar", inline=False) embed.add_field(name="Backups automáticos", value="Sí" if cfg.get("automatic_backups", True) else "No", inline=True) embed.add_field(name="Frecuencia", value=f"{cfg.get('backup_interval_hours',6)} h", inline=True) embed.add_field(name="Conservar", value=str(cfg.get("keep_backups",30)), inline=True) await interaction.response.send_message(embed=embed, ephemeral=True) # =================================================================== async def main() -> None: validate_token_config() intents = discord.Intents.default() intents.guilds = True intents.members = True bot = commands.Bot(command_prefix="!", intents=intents) bot.tree.add_command(NovaShieldConfigGroup()) @bot.event async def on_ready(): print(f"NovaShield conectado como {bot.user}") synced_total = 0 for guild in bot.guilds: try: guild_obj = discord.Object(id=guild.id) bot.tree.copy_global_to(guild=guild_obj) synced = await bot.tree.sync(guild=guild_obj) synced_total += len(synced) print(f"Comandos sincronizados en {guild.name}: {len(synced)}") except Exception as exc: print(f"Error sincronizando comandos en {guild.id}: {exc}") print(f"Comandos internos sincronizados: {synced_total}") @bot.event async def on_guild_join(guild: discord.Guild): try: guild_obj = discord.Object(id=guild.id) bot.tree.copy_global_to(guild=guild_obj) await bot.tree.sync(guild=guild_obj) except Exception as exc: print(f"Error sincronizando comandos al entrar a {guild.id}: {exc}") bot.ns_security_engine = security.setup(bot) print("=" * 64) print("NOVASHIELD SECURITY") print("=" * 64) print(f".env leído desde: {ENV_PATH}") print(f"Token detectado: SI ({len(settings.token)} caracteres)") print(f"Datos JSON: {settings.data_dir}") print("NovaShield: preparando almacenamiento JSON...") await storage.initialize() print("NovaShield: almacenamiento listo.") print("NovaShield: conectando con Discord...") try: await bot.start(settings.token) except discord.LoginFailure as exc: print() print("=" * 64) print("ERROR: DISCORD RECHAZÓ EL TOKEN (401 Unauthorized)") print("=" * 64) print(f"Archivo .env usado: {ENV_PATH}") print("El código está leyendo el .env correctamente.") print("El token configurado NO es válido para Discord.") print() print("Haz exactamente esto:") print("1. Discord Developer Portal") print("2. Applications > NovaShield") print("3. Bot") print("4. Reset Token") print("5. Copia el NUEVO Bot Token") print("6. Pégalo SOLO en DISCORD_TOKEN= dentro de .env") print("7. Guarda .env y vuelve a ejecutar run_bot.bat") print() print("No uses Client Secret, Application ID ni Public Key.") raise SystemExit(2) from exc finally: if not bot.is_closed(): await bot.close() if __name__ == "__main__": asyncio.run(main())