import discord from discord.ext import commands, tasks import os import json import asyncio from datetime import datetime, timedelta from collections import defaultdict, deque from dotenv import load_dotenv import logging load_dotenv() # Logging setup logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('fortress_guard.log'), logging.StreamHandler() ] ) logger = logging.getLogger('FortressGuard') # Bot setup intents = discord.Intents.all() bot = commands.Bot(command_prefix='!', intents=intents, case_insensitive=True) # Global tracking dictionaries join_tracker = defaultdict(lambda: deque(maxlen=50)) leave_tracker = defaultdict(lambda: deque(maxlen=50)) ban_tracker = defaultdict(lambda: deque(maxlen=20)) kick_tracker = defaultdict(lambda: deque(maxlen=20)) channel_delete_tracker = defaultdict(lambda: deque(maxlen=20)) role_delete_tracker = defaultdict(lambda: deque(maxlen=20)) webhook_tracker = defaultdict(lambda: deque(maxlen=20)) message_tracker = defaultdict(lambda: deque(maxlen=100)) permission_changes = defaultdict(lambda: deque(maxlen=20)) # Lockdown state lockdown_state = {} # Load config def load_config(): try: with open('config.json', 'r') as f: return json.load(f) except FileNotFoundError: logger.warning("Config file not found, using defaults") return create_default_config() def save_config(config): with open('config.json', 'w') as f: json.dump(config, f, indent=4) def create_default_config(): config = { "guilds": {}, "default_settings": { "anti_raid": { "enabled": True, "join_threshold": 10, "time_window": 10, "action": "kick" }, "anti_nuke": { "enabled": True, "ban_threshold": 5, "kick_threshold": 5, "channel_delete_threshold": 3, "role_delete_threshold": 3, "time_window": 60, "action": "remove_permissions" }, "anti_spam": { "enabled": True, "message_threshold": 7, "time_window": 5, "action": "timeout" }, "webhook_protection": { "enabled": True, "creation_threshold": 3, "time_window": 30 }, "whitelist": [], "log_channel": None, "quarantine_role": None } } save_config(config) return config config = load_config() def get_guild_config(guild_id): guild_id_str = str(guild_id) if guild_id_str not in config["guilds"]: config["guilds"][guild_id_str] = config["default_settings"].copy() save_config(config) return config["guilds"][guild_id_str] def is_whitelisted(guild_id, user_id): guild_config = get_guild_config(guild_id) return user_id in guild_config.get("whitelist", []) async def log_action(guild, title, description, color=discord.Color.red()): guild_config = get_guild_config(guild.id) log_channel_id = guild_config.get("log_channel") if log_channel_id: channel = guild.get_channel(log_channel_id) if channel: embed = discord.Embed( title=f"🛡️ {title}", description=description, color=color, timestamp=datetime.utcnow() ) embed.set_footer(text="FortressGuard Security System") try: await channel.send(embed=embed) except: logger.error(f"Failed to send log to channel {log_channel_id}") @bot.event async def on_ready(): logger.info(f'✅ FortressGuard is online! Logged in as {bot.user}') logger.info(f'Protecting {len(bot.guilds)} servers') cleanup_old_data.start() await bot.change_presence(activity=discord.Activity(type=discord.ActivityType.watching, name="for threats | !help")) @bot.event async def on_member_join(member): guild = member.guild guild_config = get_guild_config(guild.id) if not guild_config["anti_raid"]["enabled"]: return now = datetime.utcnow() join_tracker[guild.id].append(now) threshold = guild_config["anti_raid"]["join_threshold"] time_window = guild_config["anti_raid"]["time_window"] recent_joins = [t for t in join_tracker[guild.id] if (now - t).total_seconds() <= time_window] if len(recent_joins) >= threshold: logger.warning(f"⚠️ RAID DETECTED in {guild.name}! {len(recent_joins)} joins in {time_window}s") action = guild_config["anti_raid"]["action"] if action == "kick": try: if not is_whitelisted(guild.id, member.id): await member.kick(reason="FortressGuard: Anti-raid protection") await log_action( guild, "Raid Protection Activated", f"**Member Kicked:** {member.mention} ({member.id})\n" f"**Reason:** Mass join detected ({len(recent_joins)} joins in {time_window}s)\n" f"**Action:** Automatic kick" ) except Exception as e: logger.error(f"Failed to kick raider: {e}") elif action == "lockdown": await activate_lockdown(guild) await log_action( guild, "Raid Lockdown Activated", f"**Trigger:** Mass join detected ({len(recent_joins)} joins in {time_window}s)\n" f"**Action:** Server locked down\n" f"**Use:** `!unlock` to restore access" ) @bot.event async def on_member_remove(member): leave_tracker[member.guild.id].append(datetime.utcnow()) @bot.event async def on_member_ban(guild, user): guild_config = get_guild_config(guild.id) if not guild_config["anti_nuke"]["enabled"]: return now = datetime.utcnow() try: async for entry in guild.audit_logs(limit=1, action=discord.AuditLogAction.ban): if entry.target.id == user.id: if is_whitelisted(guild.id, entry.user.id): return ban_tracker[guild.id].append((now, entry.user.id)) threshold = guild_config["anti_nuke"]["ban_threshold"] time_window = guild_config["anti_nuke"]["time_window"] recent_bans = [(t, u) for t, u in ban_tracker[guild.id] if (now - t).total_seconds() <= time_window] if len(recent_bans) >= threshold: logger.critical(f"🚨 MASS BAN DETECTED in {guild.name} by {entry.user}") await handle_nuke_attempt(guild, entry.user, "mass ban", len(recent_bans)) except Exception as e: logger.error(f"Error in on_member_ban: {e}") @bot.event async def on_guild_channel_delete(channel): guild = channel.guild guild_config = get_guild_config(guild.id) if not guild_config["anti_nuke"]["enabled"]: return now = datetime.utcnow() try: async for entry in guild.audit_logs(limit=1, action=discord.AuditLogAction.channel_delete): if entry.target.id == channel.id: if is_whitelisted(guild.id, entry.user.id): return channel_delete_tracker[guild.id].append((now, entry.user.id)) threshold = guild_config["anti_nuke"]["channel_delete_threshold"] time_window = guild_config["anti_nuke"]["time_window"] recent_deletes = [(t, u) for t, u in channel_delete_tracker[guild.id] if (now - t).total_seconds() <= time_window] if len(recent_deletes) >= threshold: logger.critical(f"🚨 MASS CHANNEL DELETE DETECTED in {guild.name} by {entry.user}") await handle_nuke_attempt(guild, entry.user, "mass channel deletion", len(recent_deletes)) except Exception as e: logger.error(f"Error in on_guild_channel_delete: {e}") @bot.event async def on_guild_role_delete(role): guild = role.guild guild_config = get_guild_config(guild.id) if not guild_config["anti_nuke"]["enabled"]: return now = datetime.utcnow() try: async for entry in guild.audit_logs(limit=1, action=discord.AuditLogAction.role_delete): if entry.target.id == role.id: if is_whitelisted(guild.id, entry.user.id): return role_delete_tracker[guild.id].append((now, entry.user.id)) threshold = guild_config["anti_nuke"]["role_delete_threshold"] time_window = guild_config["anti_nuke"]["time_window"] recent_deletes = [(t, u) for t, u in role_delete_tracker[guild.id] if (now - t).total_seconds() <= time_window] if len(recent_deletes) >= threshold: logger.critical(f"🚨 MASS ROLE DELETE DETECTED in {guild.name} by {entry.user}") await handle_nuke_attempt(guild, entry.user, "mass role deletion", len(recent_deletes)) except Exception as e: logger.error(f"Error in on_guild_role_delete: {e}") @bot.event async def on_webhooks_update(channel): guild = channel.guild guild_config = get_guild_config(guild.id) if not guild_config["webhook_protection"]["enabled"]: return now = datetime.utcnow() try: async for entry in guild.audit_logs(limit=1, action=discord.AuditLogAction.webhook_create): if is_whitelisted(guild.id, entry.user.id): return webhook_tracker[guild.id].append((now, entry.user.id)) threshold = guild_config["webhook_protection"]["creation_threshold"] time_window = guild_config["webhook_protection"]["time_window"] recent_webhooks = [(t, u) for t, u in webhook_tracker[guild.id] if (now - t).total_seconds() <= time_window] if len(recent_webhooks) >= threshold: logger.warning(f"⚠️ MASS WEBHOOK CREATION in {guild.name} by {entry.user}") webhooks = await channel.webhooks() for webhook in webhooks: if webhook.user == entry.user: try: await webhook.delete(reason="FortressGuard: Unauthorized mass webhook creation") except: pass await log_action( guild, "Webhook Protection Triggered", f"**User:** {entry.user.mention} ({entry.user.id})\n" f"**Action:** Mass webhook creation detected ({len(recent_webhooks)} in {time_window}s)\n" f"**Response:** Webhooks deleted" ) except Exception as e: logger.error(f"Error in on_webhooks_update: {e}") @bot.event async def on_guild_role_update(before, after): guild = after.guild dangerous_perms = [ 'administrator', 'kick_members', 'ban_members', 'manage_guild', 'manage_roles', 'manage_channels', 'manage_webhooks' ] before_perms = {perm: value for perm, value in before.permissions if value and perm in dangerous_perms} after_perms = {perm: value for perm, value in after.permissions if value and perm in dangerous_perms} new_perms = set(after_perms.keys()) - set(before_perms.keys()) if new_perms: try: async for entry in guild.audit_logs(limit=1, action=discord.AuditLogAction.role_update): if entry.target.id == after.id: if not is_whitelisted(guild.id, entry.user.id): await log_action( guild, "Dangerous Permission Added", f"**Role:** {after.mention}\n" f"**Modified by:** {entry.user.mention} ({entry.user.id})\n" f"**New Permissions:** {', '.join(new_perms)}\n" f"**⚠️ Action Required:** Review this change immediately", color=discord.Color.orange() ) except Exception as e: logger.error(f"Error in on_guild_role_update: {e}") @bot.event async def on_message(message): if message.author.bot: await bot.process_commands(message) return guild = message.guild if not guild: return guild_config = get_guild_config(guild.id) if not guild_config["anti_spam"]["enabled"]: await bot.process_commands(message) return if is_whitelisted(guild.id, message.author.id): await bot.process_commands(message) return now = datetime.utcnow() message_tracker[message.author.id].append(now) threshold = guild_config["anti_spam"]["message_threshold"] time_window = guild_config["anti_spam"]["time_window"] recent_messages = [t for t in message_tracker[message.author.id] if (now - t).total_seconds() <= time_window] if len(recent_messages) >= threshold: logger.warning(f"⚠️ SPAM DETECTED from {message.author} in {guild.name}") try: await message.channel.purge(limit=20, check=lambda m: m.author == message.author) action = guild_config["anti_spam"]["action"] if action == "timeout": timeout_duration = timedelta(minutes=10) await message.author.timeout(timeout_duration, reason="FortressGuard: Spam detected") await log_action( guild, "Anti-Spam Protection", f"**User:** {message.author.mention} ({message.author.id})\n" f"**Action:** Timed out for 10 minutes\n" f"**Reason:** Spam detected ({len(recent_messages)} messages in {time_window}s)" ) elif action == "kick": await message.author.kick(reason="FortressGuard: Spam detected") await log_action( guild, "Anti-Spam Protection", f"**User:** {message.author.mention} ({message.author.id})\n" f"**Action:** Kicked\n" f"**Reason:** Spam detected ({len(recent_messages)} messages in {time_window}s)" ) except Exception as e: logger.error(f"Failed to handle spammer: {e}") await bot.process_commands(message) async def handle_nuke_attempt(guild, attacker, attack_type, count): try: member = guild.get_member(attacker.id) if member: await member.ban(reason=f"FortressGuard: {attack_type} detected ({count} actions)") await log_action( guild, "🚨 NUKE ATTEMPT BLOCKED", f"**Attacker:** {attacker.mention} ({attacker.id})\n" f"**Attack Type:** {attack_type}\n" f"**Actions Detected:** {count}\n" f"**Response:** User banned and permissions removed", color=discord.Color.dark_red() ) except Exception as e: logger.error(f"Failed to handle nuke attempt: {e}") async def activate_lockdown(guild): lockdown_state[guild.id] = [] for channel in guild.text_channels: try: overwrite = channel.overwrites_for(guild.default_role) original_send = overwrite.send_messages lockdown_state[guild.id].append((channel.id, original_send)) overwrite.send_messages = False await channel.set_permissions(guild.default_role, overwrite=overwrite) except Exception as e: logger.error(f"Failed to lock channel {channel.name}: {e}") async def deactivate_lockdown(guild): if guild.id not in lockdown_state: return False for channel_id, original_perm in lockdown_state[guild.id]: channel = guild.get_channel(channel_id) if channel: try: overwrite = channel.overwrites_for(guild.default_role) overwrite.send_messages = original_perm await channel.set_permissions(guild.default_role, overwrite=overwrite) except Exception as e: logger.error(f"Failed to unlock channel {channel.name}: {e}") del lockdown_state[guild.id] return True @tasks.loop(minutes=5) async def cleanup_old_data(): now = datetime.utcnow() for guild_id in list(join_tracker.keys()): join_tracker[guild_id] = deque( [t for t in join_tracker[guild_id] if (now - t).total_seconds() <= 300], maxlen=50 ) for guild_id in list(ban_tracker.keys()): ban_tracker[guild_id] = deque( [(t, u) for t, u in ban_tracker[guild_id] if (now - t).total_seconds() <= 600], maxlen=20 ) @bot.command(name='lockdown', aliases=['lock']) @commands.has_permissions(administrator=True) async def lockdown(ctx): await activate_lockdown(ctx.guild) await ctx.send("🔒 **Server locked down.** Only staff can send messages. Use `!unlock` to restore.") await log_action( ctx.guild, "Manual Lockdown", f"**Initiated by:** {ctx.author.mention}\n**Status:** Server locked", color=discord.Color.gold() ) @bot.command(name='unlock') @commands.has_permissions(administrator=True) async def unlock(ctx): success = await deactivate_lockdown(ctx.guild) if success: await ctx.send("🔓 **Server unlocked.** Normal permissions restored.") await log_action( ctx.guild, "Lockdown Lifted", f"**Initiated by:** {ctx.author.mention}\n**Status:** Server unlocked", color=discord.Color.green() ) else: await ctx.send("❌ Server is not currently locked down.") @bot.command(name='whitelist') @commands.has_permissions(administrator=True) async def whitelist(ctx, user: discord.User): guild_config = get_guild_config(ctx.guild.id) if user.id not in guild_config["whitelist"]: guild_config["whitelist"].append(user.id) save_config(config) await ctx.send(f"✅ {user.mention} has been added to the whitelist.") await log_action( ctx.guild, "User Whitelisted", f"**User:** {user.mention} ({user.id})\n**Added by:** {ctx.author.mention}", color=discord.Color.blue() ) else: await ctx.send(f"ℹ️ {user.mention} is already whitelisted.") @bot.command(name='unwhitelist') @commands.has_permissions(administrator=True) async def unwhitelist(ctx, user: discord.User): guild_config = get_guild_config(ctx.guild.id) if user.id in guild_config["whitelist"]: guild_config["whitelist"].remove(user.id) save_config(config) await ctx.send(f"✅ {user.mention} has been removed from the whitelist.") await log_action( ctx.guild, "User Removed from Whitelist", f"**User:** {user.mention} ({user.id})\n**Removed by:** {ctx.author.mention}", color=discord.Color.blue() ) else: await ctx.send(f"ℹ️ {user.mention} is not whitelisted.") @bot.command(name='setlog') @commands.has_permissions(administrator=True) async def setlog(ctx, channel: discord.TextChannel): guild_config = get_guild_config(ctx.guild.id) guild_config["log_channel"] = channel.id save_config(config) await ctx.send(f"✅ Security logs will now be sent to {channel.mention}") await log_action( ctx.guild, "Log Channel Updated", f"**New Channel:** {channel.mention}\n**Set by:** {ctx.author.mention}", color=discord.Color.blue() ) @bot.command(name='config') @commands.has_permissions(administrator=True) async def show_config(ctx): guild_config = get_guild_config(ctx.guild.id) embed = discord.Embed( title="🛡️ FortressGuard Configuration", color=discord.Color.blue(), timestamp=datetime.utcnow() ) embed.add_field( name="Anti-Raid", value=f"**Enabled:** {guild_config['anti_raid']['enabled']}\n" f"**Threshold:** {guild_config['anti_raid']['join_threshold']} joins in {guild_config['anti_raid']['time_window']}s\n" f"**Action:** {guild_config['anti_raid']['action']}", inline=False ) embed.add_field( name="Anti-Nuke", value=f"**Enabled:** {guild_config['anti_nuke']['enabled']}\n" f"**Ban Threshold:** {guild_config['anti_nuke']['ban_threshold']}\n" f"**Channel Delete Threshold:** {guild_config['anti_nuke']['channel_delete_threshold']}\n" f"**Role Delete Threshold:** {guild_config['anti_nuke']['role_delete_threshold']}", inline=False ) embed.add_field( name="Anti-Spam", value=f"**Enabled:** {guild_config['anti_spam']['enabled']}\n" f"**Threshold:** {guild_config['anti_spam']['message_threshold']} messages in {guild_config['anti_spam']['time_window']}s\n" f"**Action:** {guild_config['anti_spam']['action']}", inline=False ) embed.add_field( name="Webhook Protection", value=f"**Enabled:** {guild_config['webhook_protection']['enabled']}\n" f"**Threshold:** {guild_config['webhook_protection']['creation_threshold']} in {guild_config['webhook_protection']['time_window']}s", inline=False ) whitelist_count = len(guild_config["whitelist"]) embed.add_field(name="Whitelisted Users", value=str(whitelist_count), inline=True) log_channel_id = guild_config.get("log_channel") if log_channel_id: channel = ctx.guild.get_channel(log_channel_id) embed.add_field(name="Log Channel", value=channel.mention if channel else "Not set", inline=True) else: embed.add_field(name="Log Channel", value="Not set", inline=True) embed.set_footer(text="Use !help to see all commands") await ctx.send(embed=embed) @bot.command(name='status') @commands.has_permissions(administrator=True) async def status(ctx): guild = ctx.guild recent_joins = len([t for t in join_tracker[guild.id] if (datetime.utcnow() - t).total_seconds() <= 60]) recent_bans = len([(t, u) for t, u in ban_tracker[guild.id] if (datetime.utcnow() - t).total_seconds() <= 60]) embed = discord.Embed( title="🛡️ FortressGuard Status", color=discord.Color.green(), timestamp=datetime.utcnow() ) embed.add_field(name="Joins (Last 60s)", value=str(recent_joins), inline=True) embed.add_field(name="Bans (Last 60s)", value=str(recent_bans), inline=True) embed.add_field(name="Lockdown Active", value=str(guild.id in lockdown_state), inline=True) embed.add_field(name="Protected Servers", value=str(len(bot.guilds)), inline=True) embed.add_field(name="Latency", value=f"{round(bot.latency * 1000)}ms", inline=True) embed.set_footer(text="FortressGuard Security System") await ctx.send(embed=embed) @bot.command(name='nuke') @commands.has_permissions(administrator=True) async def emergency_nuke(ctx): await ctx.send("⚠️ **EMERGENCY MODE:** React with 🛡️ within 10 seconds to ban all members who joined in the last 5 minutes.") confirm_msg = await ctx.send("React to confirm emergency action!") await confirm_msg.add_reaction("🛡️") def check(reaction, user): return user == ctx.author and str(reaction.emoji) == "🛡️" and reaction.message.id == confirm_msg.id try: reaction, user = await bot.wait_for('reaction_add', timeout=10.0, check=check) now = datetime.utcnow() count = 0 for member in ctx.guild.members: if (now - member.joined_at).total_seconds() <= 300 and not member.bot: try: await member.ban(reason="FortressGuard: Emergency raid response") count += 1 except: pass await ctx.send(f"✅ Emergency action complete. Banned {count} recent members.") await log_action( ctx.guild, "🚨 Emergency Nuke Executed", f"**Initiated by:** {ctx.author.mention}\n**Members Banned:** {count}\n**Criteria:** Joined in last 5 minutes", color=discord.Color.dark_red() ) except asyncio.TimeoutError: await ctx.send("❌ Emergency action cancelled (timeout).") @bot.command(name='purge') @commands.has_permissions(manage_messages=True) async def purge(ctx, amount: int): if amount > 100: await ctx.send("❌ Cannot delete more than 100 messages at once.") return deleted = await ctx.channel.purge(limit=amount + 1) msg = await ctx.send(f"✅ Deleted {len(deleted) - 1} messages.") await asyncio.sleep(3) await msg.delete() @bot.command(name='fortress') async def fortress_info(ctx): embed = discord.Embed( title="🛡️ FortressGuard Security Bot", description="Advanced protection against raids, nukes, and malicious attacks.", color=discord.Color.blue(), timestamp=datetime.utcnow() ) embed.add_field( name="Protection Features", value="• Anti-Raid (mass join detection)\n" "• Anti-Nuke (mass ban/delete detection)\n" "• Anti-Spam (message flood protection)\n" "• Webhook monitoring\n" "• Permission change alerts\n" "• Whitelist system", inline=False ) embed.add_field( name="Key Commands", value="`!help` - View all commands\n" "`!config` - View current settings\n" "`!status` - View security status\n" "`!lockdown` - Lock server\n" "`!whitelist @user` - Add trusted user", inline=False ) embed.set_footer(text="Developed for maximum server protection") await ctx.send(embed=embed) @bot.event async def on_command_error(ctx, error): if isinstance(error, commands.MissingPermissions): await ctx.send("❌ You don't have permission to use this command.") elif isinstance(error, commands.MissingRequiredArgument): await ctx.send(f"❌ Missing required argument: `{error.param.name}`") elif isinstance(error, commands.BadArgument): await ctx.send("❌ Invalid argument provided.") elif isinstance(error, commands.CommandNotFound): return else: logger.error(f"Command error: {error}") await ctx.send("❌ An error occurred while executing the command.") if __name__ == "__main__": token = os.getenv('DISCORD_TOKEN') if not token: logger.error("❌ DISCORD_TOKEN not found in .env file!") exit(1) try: bot.run(token) except Exception as e: logger.error(f"Failed to start bot: {e}")