diff --git a/antidoxxing/antidoxxing.py b/antidoxxing/antidoxxing.py index 7f79b36..c18dad8 100644 --- a/antidoxxing/antidoxxing.py +++ b/antidoxxing/antidoxxing.py @@ -8,17 +8,25 @@ Features: 2. Strip Discord markdown characters 3. Remove invisible zero-width characters 4. Homoglyph normalization (leet -> digits) - 5. Create numeric-only string by removing spaces, dots, dashes -- Detection for IPv4, IPv6, North American phone numbers, SSNs, and simple GPS coords. -- On detection: delete message and send a silent log to a configured moderation channel. -- Uses Config per guild (setmodchannel command). + 5. Unicode normalization and simple obfuscation replacements + 6. Create numeric-only string by removing spaces, dots, dashes +- Detection for IPv4, IPv6, phone, SSN, emails, and simple GPS coords. +- Scoring: weights per detection; quarantine vs reject thresholds. +- On reject: delete message and send a silent log to a configured moderation channel and/or a global webhook. +- Quarantine: do not delete; send to mod channel/webhook for review. +- Guild-level config: enabled, thresholds, whitelist roles, excluded channels, auto-action. +- Global config: single webhook URL for all guilds. """ import re import ipaddress import logging -from typing import Optional +import unicodedata +import asyncio +import time +from typing import Optional, Dict, List, Tuple +import aiohttp import discord from redbot.core import commands, Config, checks @@ -26,15 +34,32 @@ log = logging.getLogger("red.antidoxxing") class AntiDoxxing(commands.Cog): - """Detect and remove common doxxing attempts (IPv4, IPv6, phone, SSN, GPS).""" + """Detect and remove common doxxing attempts (IPv4, IPv6, phone, SSN, GPS, email). - __version__ = "1.0.0" + Provides guild-level config and a single global webhook for modlogs. + """ + + __version__ = "1.1.0" def __init__(self, bot): self.bot = bot # Unique identifier for Config; change if you fork to avoid collisions self.config = Config.get_conf(self, identifier=1234567890123, force_registration=True) - self.config.register_guild(mod_channel=None) + + # Global config (one webhook for all guilds) + self.config.register_global(webhook_url=None) + + # Guild-specific config + self.config.register_guild( + mod_channel=None, + enabled=True, + filter_level="high", # low|medium|high|paranoid + quarantine_threshold=0.6, + reject_threshold=0.85, + whitelist_role_ids=[], + excluded_channel_ids=[], + auto_action={"violations": 3, "interval_minutes": 60, "action": "temp_ban_24h"}, + ) # Precompute translation maps and regex candidates # Characters to strip for discord markdown formatting @@ -83,9 +108,23 @@ class AntiDoxxing(commands.Cog): self._ipv6_candidate_re = re.compile(r"(?:[0-9A-Fa-f:]{2,})") # Candidate GPS: float pairs with a comma or whitespace separator (simple) self._gps_candidate_re = re.compile(r"([+-]?\d{1,3}\.\d+)[,\s]+([+-]?\d{1,3}\.\d+)") - # Candidate digits for phone/ssn detection will be taken from numeric_text via \d runs + # Email regex (simple and safe) + self._email_re = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}") - # ---------- Commands ---------- + # In-memory recent violation tracking: (guild_id, user_id) -> list[timestamp_seconds] + self._violation_history: Dict[Tuple[int, int], List[float]] = {} + + # Scoring weights (tuneable) + self._weights = { + "IPv4": 1.0, + "IPv6": 1.0, + "GPS": 0.9, + "Phone": 0.9, + "SSN": 1.0, + "Email": 0.8, + } + + # ---------- Administrative Commands ---------- @checks.admin_or_permissions(manage_guild=True) @commands.guild_only() @commands.command(name="setmodchannel", help="Set the moderation channel for antidoxxing logs. Use without args to unset.") @@ -94,7 +133,7 @@ class AntiDoxxing(commands.Cog): guild = ctx.guild if channel is None: await self.config.guild(guild).mod_channel.set(None) - await ctx.send("AntiDoxxing moderation channel unset. Logs will not be posted.") + await ctx.send("AntiDoxxing moderation channel unset. Logs will not be posted to a channel.") return # Save channel ID await self.config.guild(guild).mod_channel.set(channel.id) @@ -116,6 +155,104 @@ class AntiDoxxing(commands.Cog): return await ctx.send("No moderation channel configured for AntiDoxxing.") + @checks.admin_or_permissions(manage_guild=True) + @commands.guild_only() + @commands.command(name="setenabled", help="Enable or disable AntiDoxxing in this guild.") + async def setenabled(self, ctx: commands.Context, enabled: bool): + guild = ctx.guild + await self.config.guild(guild).enabled.set(bool(enabled)) + await ctx.send(f"AntiDoxxing enabled set to {enabled} for this guild.") + + @checks.admin_or_permissions(manage_guild=True) + @commands.guild_only() + @commands.command(name="addwhitelistrole", help="Add a role to the Antidoxxing whitelist (bypasses checks).") + async def addwhitelistrole(self, ctx: commands.Context, role: discord.Role): + guild = ctx.guild + current = await self.config.guild(guild).whitelist_role_ids() + if role.id in current: + await ctx.send("Role already whitelisted.") + return + current.append(role.id) + await self.config.guild(guild).whitelist_role_ids.set(current) + await ctx.send(f"Role {role.name} added to AntiDoxxing whitelist.") + + @checks.admin_or_permissions(manage_guild=True) + @commands.guild_only() + @commands.command(name="removewhitelistrole", help="Remove a role from the Antidoxxing whitelist.") + async def removewhitelistrole(self, ctx: commands.Context, role: discord.Role): + guild = ctx.guild + current = await self.config.guild(guild).whitelist_role_ids() + if role.id not in current: + await ctx.send("Role not in whitelist.") + return + current.remove(role.id) + await self.config.guild(guild).whitelist_role_ids.set(current) + await ctx.send(f"Role {role.name} removed from AntiDoxxing whitelist.") + + @checks.admin_or_permissions(manage_guild=True) + @commands.guild_only() + @commands.command(name="listexcludedchannels", help="List channels excluded from Antidoxxing.") + async def listexcludedchannels(self, ctx: commands.Context): + guild = ctx.guild + current = await self.config.guild(guild).excluded_channel_ids() + if not current: + await ctx.send("No channels excluded.") + return + lines = [] + for cid in current: + ch = guild.get_channel(cid) or self.bot.get_channel(cid) + if ch: + lines.append(f"- {ch.name} (ID: {cid})") + else: + lines.append(f"- Unknown channel ID: {cid}") + await ctx.send("Excluded channels:\n" + "\n".join(lines)) + + @checks.admin_or_permissions(manage_guild=True) + @commands.guild_only() + @commands.command(name="addexcludedchannel", help="Add a channel to the exclusion list for Antidoxxing.") + async def addexcludedchannel(self, ctx: commands.Context, channel: discord.TextChannel): + guild = ctx.guild + current = await self.config.guild(guild).excluded_channel_ids() + if channel.id in current: + await ctx.send("Channel already excluded.") + return + current.append(channel.id) + await self.config.guild(guild).excluded_channel_ids.set(current) + await ctx.send(f"Channel #{channel.name} excluded from AntiDoxxing checks.") + + @checks.admin_or_permissions(manage_guild=True) + @commands.guild_only() + @commands.command(name="removeexcludedchannel", help="Remove a channel from the exclusion list.") + async def removeexcludedchannel(self, ctx: commands.Context, channel: discord.TextChannel): + guild = ctx.guild + current = await self.config.guild(guild).excluded_channel_ids() + if channel.id not in current: + await ctx.send("Channel not in exclusion list.") + return + current.remove(channel.id) + await self.config.guild(guild).excluded_channel_ids.set(current) + await ctx.send(f"Channel #{channel.name} removed from exclusion list.") + + # Global webhook setter - bot owner only + @checks.is_owner() + @commands.command(name="setmodwebhook", help="Set a global webhook URL for AntiDoxxing modlogs. Use without args to unset.") + async def setmodwebhook(self, ctx: commands.Context, webhook_url: Optional[str] = None): + if webhook_url is None: + await self.config.webhook_url.set(None) + await ctx.send("AntiDoxxing global webhook unset.") + return + await self.config.webhook_url.set(webhook_url) + await ctx.send("AntiDoxxing global webhook set.") + + @checks.is_owner() + @commands.command(name="showmodwebhook", help="Show configured global webhook URL for AntiDoxxing (bot owner only).") + async def showmodwebhook(self, ctx: commands.Context): + url = await self.config.webhook_url() + if url: + await ctx.send(f"Global AntiDoxxing webhook: {url}") + else: + await ctx.send("No global webhook configured.") + # ---------- Listeners ---------- @commands.Cog.listener() async def on_message(self, message: discord.Message): @@ -142,6 +279,30 @@ class AntiDoxxing(commands.Cog): if message.guild is None: return + guild = message.guild + + # Load guild config early for short-circuit + guild_cfg = await self.config.guild(guild).all() + enabled = guild_cfg.get("enabled", True) + excluded_channels = guild_cfg.get("excluded_channel_ids", []) or [] + whitelist_roles = set(guild_cfg.get("whitelist_role_ids", []) or []) + mod_channel_id = guild_cfg.get("mod_channel") + quarantine_threshold = guild_cfg.get("quarantine_threshold", 0.6) + reject_threshold = guild_cfg.get("reject_threshold", 0.85) + auto_action_cfg = guild_cfg.get("auto_action", {"violations": 3, "interval_minutes": 60, "action": "temp_ban_24h"}) + + if not enabled: + return + + # Excluded channels + if message.channel.id in excluded_channels: + return + + # Whitelisted roles + author_role_ids = {r.id for r in getattr(message.author, "roles", [])} + if whitelist_roles & author_role_ids: + return + raw = message.content or "" if not raw: return @@ -155,16 +316,48 @@ class AntiDoxxing(commands.Cog): if ch in sanitized: sanitized = sanitized.replace(ch, repl) + # UNICODE NORMALIZATION + sanitized = unicodedata.normalize("NFKC", sanitized) + + # SIMPLE OBFS NORMALIZATION: common human obfuscations -> canonical + lowered = sanitized + replacements = [ + ("[at]", "@"), + ("(at)", "@"), + (" at ", "@"), + ("\\s?\\[dot\\]\\s?", "."), + (" dotcom", ".com"), + (" dot ", "."), + ] + # apply simple replacements safely + for a, b in replacements: + try: + lowered = re.sub(a, b, lowered, flags=re.IGNORECASE) + except re.error: + # fallback to plain replace if replacement pattern fails + lowered = lowered.replace(a, b) + + # REMOVE DISCORD MENTIONS: user/nick/role/channel mentions (including possible missing closing '>') + # Patterns: <@123...>, <@!123...>, <@&123...>, <#123...> + try: + lowered = re.sub(r"<@!?\d+>?", "", lowered) + lowered = re.sub(r"<@&\d+>?", "", lowered) + lowered = re.sub(r"<#\d+>?", "", lowered) + except re.error: + # In the rare case regex fails, fallback to simple replacements for common prefixes + lowered = lowered.replace("<@", "") + lowered = lowered.replace("<#", "") + # HOMOGLYPH NORMALIZATION: map common letters to digits for detection # Create a normalized copy for detection only - normalized = sanitized.translate(self._homoglyph_trans) + normalized = lowered.translate(self._homoglyph_trans) # STRIP SPACES & PUNCTUATION: create numeric_text used for phone/SSN detection # Remove spaces, dots, and dashes (per requirements) - numeric_text = re.sub(r"[ \.\-]", "", normalized) + numeric_text = re.sub(r"[ \\.\-]", "", normalized) # Detection: gather flags and reasons to log - violations = [] + detections: List[Tuple[str, str]] = [] # IPv4 detection: for candidate in self._ipv4_candidate_re.findall(normalized): @@ -182,14 +375,14 @@ class AntiDoxxing(commands.Cog): ipv4_str = ".".join(str(o) for o in octets) try: ipaddress.IPv4Address(ipv4_str) - violations.append(("IPv4", ipv4_str)) + detections.append(("IPv4", ipv4_str)) break # no need to find more IPv4 candidates except Exception: # not a valid IPv4, continue searching continue # IPv6 detection: - if not any(v[0] == "IPv6" for v in violations): + if not any(d[0] == "IPv6" for d in detections): for raw_candidate in self._ipv6_candidate_re.findall(normalized): if ":" not in raw_candidate: continue @@ -198,16 +391,15 @@ class AntiDoxxing(commands.Cog): continue # Try validating via ipaddress try: - # Strip surrounding non-hex/colon characters candidate = raw_candidate.strip() ipaddress.IPv6Address(candidate) - violations.append(("IPv6", candidate)) + detections.append(("IPv6", candidate)) break except Exception: continue # GPS detection (simple float-pair detection) - if not any(v[0] == "GPS" for v in violations): + if not any(d[0] == "GPS" for d in detections): for lat_s, lon_s in self._gps_candidate_re.findall(normalized): try: lat = float(lat_s) @@ -216,22 +408,22 @@ class AntiDoxxing(commands.Cog): continue # Validate ranges: latitude [-90, 90], longitude [-180, 180] if -90.0 <= lat <= 90.0 and -180.0 <= lon <= 180.0: - violations.append(("GPS", f"{lat},{lon}")) + detections.append(("GPS", f"{lat},{lon}")) break # PHONE detection (North American): look for 10-digit or 11-digit numbers starting with 1 - if not any(v[0] == "Phone" for v in violations): + if not any(d[0] == "Phone" for d in detections): for match in re.finditer(r"\d{10,11}", numeric_text): digits = match.group(0) if len(digits) == 10: - violations.append(("Phone", digits)) + detections.append(("Phone", digits)) break elif len(digits) == 11 and digits.startswith("1"): - violations.append(("Phone", digits)) + detections.append(("Phone", digits)) break # SSN detection: 9-digit sequences, exclude invalid ranges - if not any(v[0] == "SSN" for v in violations): + if not any(d[0] == "SSN" for d in detections): for match in re.finditer(r"\d{9}", numeric_text): digits = match.group(0) # Basic SSN invalid checks @@ -249,59 +441,218 @@ class AntiDoxxing(commands.Cog): if group == "00" or serial == "0000": continue # Looks like a valid SSN - violations.append(("SSN", digits)) + detections.append(("SSN", digits)) break - # If anything flagged, take action - if violations: + # EMAIL detection + if not any(d[0] == "Email" for d in detections): + for match in self._email_re.finditer(normalized): + addr = match.group(0) + detections.append(("Email", addr)) + break + + # If nothing detected, return + if not detections: + return + + # Compute a simple score based on weights + score = 0.0 + for dtype, _ in detections: + score += self._weights.get(dtype, 0.5) + # Normalize score (simple): divide by max possible weight (sum of weights of detected types but cap) + # For decision we compare raw score vs thresholds tuned to these weights + + # Decide action based on thresholds + if score >= reject_threshold: + action = "reject" + elif score >= quarantine_threshold: + action = "quarantine" + else: + action = "ignore" + + # Build structured violation list for logs + violation_lines = [f"- {dtype}: `{val}`" for dtype, val in detections] + + # Take actions + if action == "reject": # Attempt to delete the offending message + deleted = False try: await message.delete() + deleted = True except discord.Forbidden: - # Bot lacks permissions to delete; still attempt to log log.exception("Missing permission to delete message in %s", message.guild) except discord.NotFound: # Message already deleted - pass + deleted = True except Exception: log.exception("Unexpected error deleting message") - # Send a silent log to the configured moderation channel (guild-specific) - try: - mod_channel_id = await self.config.guild(message.guild).mod_channel() - if mod_channel_id: - mod_channel = message.guild.get_channel(mod_channel_id) or self.bot.get_channel(mod_channel_id) - if mod_channel and isinstance(mod_channel, discord.TextChannel): - # Build a concise, non-pinging log message - violation_lines = [] - for vtype, val in violations: - violation_lines.append(f"- {vtype}: `{val}`") - author_str = f"{message.author} (ID: {message.author.id})" - jump_url = f"https://discord.com/channels/{message.guild.id}/{message.channel.id}/{message.id}" - # Note: message is deleted already but link gives context for moderators when accessible - log_text = ( - f"AntiDoxxing: message removed in #{message.channel.name} (ID: {message.channel.id})\n" - f"Author: {author_str}\n" - f"Reason(s):\n" + "\n".join(violation_lines) + "\n" - f"Original message content (sanitized copy):\n" - f"```{sanitized[:1900]}```\n" - f"Jump (may be unavailable if message deleted): {jump_url}" - ) - # Send without mentioning the user (no @) - try: - await mod_channel.send(log_text) - except discord.Forbidden: - log.exception("Missing permission to send logs to mod channel %s", mod_channel) - except Exception: - log.exception("Failed to send mod log") - else: - # Configured ID not found or not a text channel - log.warning("Configured mod channel ID %s not found or not a text channel", mod_channel_id) - else: - # No mod channel configured; do nothing (or optionally log to bot owner) - log.debug("No anti-dox mod channel configured for guild %s", message.guild.id) - except Exception: - log.exception("Error while attempting to log anti-dox action") + # Record violation and potentially auto-act + await self._record_violation_and_maybe_autoban(guild.id, message.author.id, auto_action_cfg, message) + + # Send logs to configured targets + await self._send_logs(guild, message, violation_lines, sanitized, deleted, detections) + + elif action == "quarantine": + # Do not delete but send detailed log for moderator review + await self._send_logs(guild, message, violation_lines, sanitized, deleted=False, detections=detections, quarantined=True) + + else: + # Shouldn't get here because we returned earlier if detections empty + log.debug("Detection score below quarantine threshold; ignoring.") + except Exception: # Top-level protection so the bot doesn't crash from unexpected input log.exception("Unhandled error in AntiDoxxing._process_message") + + async def _send_logs(self, guild: discord.Guild, message: discord.Message, violation_lines: List[str], sanitized: str, deleted: bool = False, detections: Optional[List[Tuple[str, str]]] = None, quarantined: bool = False): + """Send logs to mod channel and/or global webhook. Webhook is a single global URL for all guilds.""" + try: + # Prepare human-friendly info + author_str = f"{message.author} (ID: {message.author.id})" + jump_url = f"https://discord.com/channels/{guild.id}/{message.channel.id}/{message.id}" + status = "deleted" if deleted else ("quarantined" if quarantined else "detected") + + log_text = ( + f"AntiDoxxing: message {status} in #{message.channel.name} (ID: {message.channel.id})\n" + f"Guild: {guild.name} (ID: {guild.id})\n" + f"Author: {author_str}\n" + f"Reason(s):\n" + "\n".join(violation_lines) + "\n" + f"Original message content (sanitized copy):\n" + f"```{sanitized[:1900]}```\n" + f"Jump (may be unavailable if message deleted): {jump_url}" + ) + + # Send to configured mod channel if present + try: + mod_channel_id = await self.config.guild(guild).mod_channel() + if mod_channel_id: + mod_channel = guild.get_channel(mod_channel_id) or self.bot.get_channel(mod_channel_id) + if mod_channel and isinstance(mod_channel, discord.TextChannel): + try: + await mod_channel.send(log_text) + except discord.Forbidden: + log.exception("Missing permission to send logs to mod channel %s", mod_channel) + except Exception: + log.exception("Failed to send mod log to channel") + else: + log.warning("Configured mod channel ID %s not found or not a text channel", mod_channel_id) + else: + log.debug("No anti-dox mod channel configured for guild %s", guild.id) + except Exception: + log.exception("Error while attempting to log anti-dox action to channel") + + # Send to global webhook if configured + try: + webhook_url = await self.config.webhook_url() + if webhook_url: + payload = { + "username": "AntiDoxxing", + "content": None, + "embeds": [ + { + "title": f"AntiDoxxing - {status} - {guild.name}", + "description": "\n".join(violation_lines)[:2048], + "fields": [ + {"name": "Guild", "value": f"{guild.name} (ID: {guild.id})", "inline": True}, + {"name": "Channel", "value": f"#{message.channel.name} (ID: {message.channel.id})", "inline": True}, + {"name": "Author", "value": author_str, "inline": False}, + {"name": "Sanitized message", "value": sanitized[:1024], "inline": False}, + {"name": "Jump URL", "value": jump_url, "inline": False}, + ], + "color": 15158332, + "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + } + ], + } + await self._post_webhook(webhook_url, payload) + except Exception: + log.exception("Failed to send global webhook log") + + except Exception: + log.exception("Error while preparing/sending logs") + + async def _post_webhook(self, url: str, payload: dict): + """POST a JSON payload to the configured webhook URL. Fire-and-forget with basic error handling.""" + try: + async with aiohttp.ClientSession() as session: + async with session.post(url, json=payload, timeout=10) as resp: + if resp.status >= 400: + log.warning("Webhook post returned status %s", resp.status) + except Exception: + log.exception("Exception while posting to webhook %s", url) + + async def _record_violation_and_maybe_autoban(self, guild_id: int, user_id: int, auto_action_cfg: dict, message: discord.Message): + """Record a violation and perform auto_action if thresholds are exceeded.""" + try: + key = (guild_id, user_id) + now_ts = time.time() + history = self._violation_history.get(key, []) + # purge old entries + interval = auto_action_cfg.get("interval_minutes", 60) * 60 + history = [t for t in history if now_ts - t <= interval] + history.append(now_ts) + self._violation_history[key] = history + + required = auto_action_cfg.get("violations", 3) + action = auto_action_cfg.get("action", "temp_ban_24h") + + if len(history) >= required: + # attempt action + guild = message.guild + member = guild.get_member(user_id) + if member is None: + # cannot act if member not in cache + return + try: + if action == "kick": + await guild.kick(member) + log.info("Auto-action: kicked member %s in guild %s", user_id, guild_id) + elif action.startswith("temp_ban"): + # parse optional duration in hours e.g. temp_ban_24h + hours = 24 + m = re.search(r"temp_ban_(\d+)h", action) + if m: + hours = int(m.group(1)) + await guild.ban(member, reason="Auto action by AntiDoxxing") + log.info("Auto-action: banned member %s in guild %s for %sh", user_id, guild_id, hours) + + # schedule unban after duration (best-effort; will not survive restarts) + asyncio.create_task(self._schedule_unban(guild, user_id, hours * 3600)) + elif action == "ban": + await guild.ban(member, reason="Auto action by AntiDoxxing") + log.info("Auto-action: banned member %s in guild %s", user_id, guild_id) + else: + log.info("Unknown auto_action %s", action) + except discord.Forbidden: + log.exception("Missing permission to perform auto-action %s in guild %s", action, guild_id) + except Exception: + log.exception("Exception while performing auto-action %s in guild %s", action, guild_id) + + # reset history after action + self._violation_history[key] = [] + + except Exception: + log.exception("Error recording violation or performing auto action") + + async def _schedule_unban(self, guild: discord.Guild, user_id: int, delay_seconds: int): + """Unban a user after delay_seconds (best-effort).""" + try: + await asyncio.sleep(delay_seconds) + # attempt unban + try: + user = await self.bot.fetch_user(user_id) + await guild.unban(user) + log.info("Auto-action: unbanned member %s in guild %s after delay", user_id, guild.id) + except discord.NotFound: + log.warning("Unban: user %s not found in ban list for guild %s", user_id, guild.id) + except discord.Forbidden: + log.exception("Unban: missing permissions to unban %s in guild %s", user_id, guild.id) + except Exception: + log.exception("Unban: unexpected error for %s in guild %s", user_id, guild.id) + except asyncio.CancelledError: + # task cancelled on shutdown + return + except Exception: + log.exception("Error in _schedule_unban")