ebafe6567d
Detection heuristics: - Rich-embed abuse (user accounts can't send rich embeds) - Inhuman response speed (<300ms consistently) - Message timing precision analysis - 24/7 activity profiling - Automated command-response pattern matching - Rapid-fire burst detection Configurable per-server: - Toggle on/off (default: off) - Actions: log, warn, mute, kick, ban - Sensitivity: low/medium/high + custom threshold - Log channel, exempt roles/users - Staff notifications, mute duration Commands: [p]sbguard enable/disable/action/sensitivity/threshold/logchannel/scan/flagged/reset Slash: /sbguard-status, /sbguard-scan
1005 lines
44 KiB
Python
1005 lines
44 KiB
Python
"""
|
||
SelfbotGuard v1.0.0 — Advanced Selfbot Detection & Punishment
|
||
================================================================
|
||
Cog for Red-DiscordBot.
|
||
|
||
Detection heuristics:
|
||
1. Rich-embed abuse — User accounts cannot send rich embeds via the client.
|
||
2. Response-time analysis — Inhuman reply speeds (< 300ms consistently).
|
||
3. Timing precision — Messages at exact intervals (variance < 50ms).
|
||
4. Activity profiling — 24/7 activity across all hours without gaps.
|
||
5. Pattern matching — Automated prefix→response command patterns.
|
||
6. Burst detection — Rapid-fire messages exceeding human typing speed.
|
||
|
||
Each heuristic contributes to a cumulative suspicion score. When the score
|
||
exceeds the server's threshold, the configured action is taken.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import math
|
||
import time
|
||
from collections import defaultdict, deque
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, Deque, List, Optional, Set, Tuple
|
||
|
||
import discord
|
||
from discord import app_commands
|
||
from redbot.core import Config, checks, commands
|
||
from redbot.core.bot import Red
|
||
|
||
log = logging.getLogger("red.selfbotguard")
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# Constants
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
# Heuristic weights (how much each detection adds to suspicion score)
|
||
WEIGHT_RICH_EMBED = 40 # Very strong signal
|
||
WEIGHT_RESPONSE_SPEED = 15 # Per fast-response instance
|
||
WEIGHT_TIMING_PRECISION = 20 # Consistent intervals
|
||
WEIGHT_ACTIVITY_247 = 25 # Active across all hour buckets
|
||
WEIGHT_PATTERN_MATCH = 15 # Command-response patterns
|
||
WEIGHT_BURST = 10 # Rapid-fire messages
|
||
|
||
# Thresholds for heuristics
|
||
RESPONSE_TIME_FLOOR_MS = 300 # Faster than this = suspicious (ms)
|
||
TIMING_VARIANCE_CEIL_MS = 50 # Lower variance = more suspicious (ms)
|
||
BURST_MESSAGES = 8 # Messages in burst window
|
||
BURST_WINDOW_SECS = 3 # Window for burst detection (seconds)
|
||
ACTIVITY_HOUR_COVERAGE = 20 # Out of 24 hours — if active in this many, suspicious
|
||
MIN_MESSAGES_FOR_TIMING = 10 # Need this many messages before timing analysis
|
||
MIN_RESPONSES_FOR_SPEED = 5 # Need this many fast responses before flagging
|
||
PATTERN_REPEAT_THRESHOLD= 5 # Same prefix→response pattern this many times
|
||
|
||
# Data retention
|
||
MAX_TIMESTAMPS_PER_USER = 200 # Rolling window of message timestamps
|
||
MAX_RESPONSE_PAIRS = 100 # Rolling window of response-time pairs
|
||
SCORE_DECAY_HOURS = 24 # Score decays fully over this period
|
||
PRUNE_INACTIVE_DAYS = 7 # Prune user data after this many days inactive
|
||
|
||
# Sensitivity presets
|
||
SENSITIVITY_PRESETS = {
|
||
"low": {"score_threshold": 150, "description": "Fewer false positives, only catches obvious selfbots"},
|
||
"medium": {"score_threshold": 100, "description": "Balanced detection — recommended for most servers"},
|
||
"high": {"score_threshold": 60, "description": "Aggressive detection — may flag power users, review logs"},
|
||
}
|
||
|
||
# Action types
|
||
VALID_ACTIONS = ("log", "warn", "mute", "kick", "ban")
|
||
|
||
# Colours
|
||
COLOUR_ALERT = discord.Colour.red()
|
||
COLOUR_WARN = discord.Colour.orange()
|
||
COLOUR_INFO = discord.Colour.blurple()
|
||
COLOUR_OK = discord.Colour.green()
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# User tracking dataclass
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
class UserProfile:
|
||
"""Tracks behavioural data for a single user in a guild."""
|
||
|
||
__slots__ = (
|
||
"user_id", "guild_id", "timestamps", "response_times_ms",
|
||
"hour_buckets", "patterns", "bursts", "embed_strikes",
|
||
"score", "last_scored", "last_message_ts", "flagged",
|
||
"action_taken", "last_active",
|
||
)
|
||
|
||
def __init__(self, user_id: int, guild_id: int):
|
||
self.user_id = user_id
|
||
self.guild_id = guild_id
|
||
self.timestamps: Deque[float] = deque(maxlen=MAX_TIMESTAMPS_PER_USER)
|
||
self.response_times_ms: Deque[float] = deque(maxlen=MAX_RESPONSE_PAIRS)
|
||
self.hour_buckets: Set[int] = set() # Hours (0-23) the user has been active
|
||
self.patterns: Dict[str, int] = defaultdict(int) # "prefix→response" counters
|
||
self.bursts: int = 0 # Number of burst events detected
|
||
self.embed_strikes: int = 0 # Rich embeds sent by non-bot user
|
||
self.score: float = 0.0
|
||
self.last_scored: float = 0.0 # time.time() of last score calc
|
||
self.last_message_ts: float = 0.0 # Timestamp of user's last message
|
||
self.flagged: bool = False
|
||
self.action_taken: Optional[str] = None
|
||
self.last_active: float = time.time()
|
||
|
||
def decay_score(self) -> float:
|
||
"""Apply time-based decay to the score."""
|
||
if self.last_scored <= 0:
|
||
return self.score
|
||
elapsed_h = (time.time() - self.last_scored) / 3600
|
||
if elapsed_h >= SCORE_DECAY_HOURS:
|
||
self.score = 0.0
|
||
else:
|
||
decay = 1.0 - (elapsed_h / SCORE_DECAY_HOURS)
|
||
self.score *= decay
|
||
self.last_scored = time.time()
|
||
return self.score
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# The Cog
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
class SelfbotGuard(commands.Cog):
|
||
"""Advanced selfbot detection with per-server configurable punishments."""
|
||
|
||
__version__ = "1.0.0"
|
||
__author__ = ["everestmcarthur"]
|
||
|
||
def __init__(self, bot: Red):
|
||
self.bot = bot
|
||
self.config = Config.get_conf(self, identifier=58_310_627_41, force_registration=True)
|
||
|
||
# Per-guild defaults
|
||
self.config.register_guild(
|
||
enabled=False,
|
||
action="log", # log | warn | mute | kick | ban
|
||
sensitivity="medium", # low | medium | high
|
||
score_threshold=100, # Override, or auto from sensitivity
|
||
log_channel=None, # Channel ID for alerts
|
||
exempt_roles=[], # Role IDs that bypass detection
|
||
exempt_users=[], # User IDs that bypass detection
|
||
mute_duration=600, # Seconds for mute/timeout (default 10min)
|
||
flagged_users={}, # user_id -> {score, reason, timestamp}
|
||
notify_staff=True, # DM the guild owner on detection
|
||
auto_prune=True, # Auto-prune old tracking data
|
||
)
|
||
|
||
# In-memory tracking: guild_id -> user_id -> UserProfile
|
||
self._profiles: Dict[int, Dict[int, UserProfile]] = defaultdict(dict)
|
||
|
||
# Recent messages cache for response-time analysis
|
||
# guild_id -> channel_id -> (author_id, timestamp, content_prefix)
|
||
self._recent_messages: Dict[int, Dict[int, Tuple[int, float, str]]] = defaultdict(dict)
|
||
|
||
# Background task
|
||
self._prune_task: Optional[asyncio.Task] = None
|
||
|
||
async def cog_load(self):
|
||
"""Start background tasks."""
|
||
self._prune_task = self.bot.loop.create_task(self._prune_loop())
|
||
log.info("SelfbotGuard v%s loaded.", self.__version__)
|
||
|
||
async def cog_unload(self):
|
||
"""Clean up."""
|
||
if self._prune_task and not self._prune_task.done():
|
||
self._prune_task.cancel()
|
||
log.info("SelfbotGuard unloaded.")
|
||
|
||
# ── Helpers ──────────────────────────────────────────────
|
||
|
||
def _get_profile(self, guild_id: int, user_id: int) -> UserProfile:
|
||
"""Get or create a UserProfile."""
|
||
guild_profiles = self._profiles[guild_id]
|
||
if user_id not in guild_profiles:
|
||
guild_profiles[user_id] = UserProfile(user_id, guild_id)
|
||
return guild_profiles[user_id]
|
||
|
||
async def _is_exempt(self, member: discord.Member) -> bool:
|
||
"""Check if a member is exempt from detection."""
|
||
if member.bot:
|
||
return True # Actual bots are not selfbots
|
||
if member.guild_permissions.administrator:
|
||
return True
|
||
if member.id == member.guild.owner_id:
|
||
return True
|
||
|
||
guild_conf = await self.config.guild(member.guild).all()
|
||
if member.id in guild_conf["exempt_users"]:
|
||
return True
|
||
member_role_ids = {r.id for r in member.roles}
|
||
if member_role_ids & set(guild_conf["exempt_roles"]):
|
||
return True
|
||
return False
|
||
|
||
async def _get_log_channel(self, guild: discord.Guild) -> Optional[discord.TextChannel]:
|
||
"""Get the configured log channel."""
|
||
ch_id = await self.config.guild(guild).log_channel()
|
||
if ch_id:
|
||
return guild.get_channel(ch_id)
|
||
return None
|
||
|
||
# ── Detection heuristics ─────────────────────────────────
|
||
|
||
def _check_rich_embeds(self, message: discord.Message, profile: UserProfile) -> float:
|
||
"""
|
||
Heuristic 1: Rich embeds from non-bot user accounts.
|
||
|
||
The Discord client does NOT allow regular users to send rich embeds.
|
||
URL unfurls (type='rich' with url field matching message content) are excluded.
|
||
If a user account sends a genuine rich embed, it's almost certainly a selfbot.
|
||
"""
|
||
if not message.embeds:
|
||
return 0.0
|
||
|
||
score = 0.0
|
||
for embed in message.embeds:
|
||
# Skip URL unfurls — these are normal
|
||
if embed.type in ("image", "video", "gifv", "link", "article"):
|
||
continue
|
||
# If it's a "rich" embed, check if it's just a URL preview
|
||
if embed.type == "rich":
|
||
# URL unfurls typically have a url that matches something in message content
|
||
if embed.url and embed.url in (message.content or ""):
|
||
continue
|
||
# If embed has no title/description/fields, it's likely a provider embed
|
||
if not embed.title and not embed.description and not embed.fields:
|
||
continue
|
||
# This is a real rich embed from a user account — strong selfbot signal
|
||
profile.embed_strikes += 1
|
||
score += WEIGHT_RICH_EMBED
|
||
log.debug(
|
||
"Rich embed detected from user %s in guild %s (strike %d)",
|
||
message.author.id, message.guild.id, profile.embed_strikes
|
||
)
|
||
return score
|
||
|
||
def _check_response_speed(self, message: discord.Message, profile: UserProfile) -> float:
|
||
"""
|
||
Heuristic 2: Inhuman response speed.
|
||
|
||
If a user consistently responds to other users' messages in < 300ms,
|
||
they're likely automated. We track response times per channel.
|
||
"""
|
||
guild_id = message.guild.id
|
||
channel_id = message.channel.id
|
||
now = time.time()
|
||
|
||
# Check if there's a recent message in this channel from someone else
|
||
last = self._recent_messages.get(guild_id, {}).get(channel_id)
|
||
if last and last[0] != message.author.id:
|
||
response_ms = (now - last[1]) * 1000
|
||
if response_ms < RESPONSE_TIME_FLOOR_MS:
|
||
profile.response_times_ms.append(response_ms)
|
||
|
||
# Update the recent message cache
|
||
if guild_id not in self._recent_messages:
|
||
self._recent_messages[guild_id] = {}
|
||
content_prefix = (message.content or "")[:50]
|
||
self._recent_messages[guild_id][channel_id] = (message.author.id, now, content_prefix)
|
||
|
||
# Score based on accumulated fast responses
|
||
fast_count = sum(1 for t in profile.response_times_ms if t < RESPONSE_TIME_FLOOR_MS)
|
||
if fast_count >= MIN_RESPONSES_FOR_SPEED:
|
||
return WEIGHT_RESPONSE_SPEED * (fast_count / MIN_RESPONSES_FOR_SPEED)
|
||
return 0.0
|
||
|
||
def _check_timing_precision(self, profile: UserProfile) -> float:
|
||
"""
|
||
Heuristic 3: Message timing precision.
|
||
|
||
Selfbots often send messages at very precise intervals (e.g., exactly
|
||
every 5.000 seconds). We measure the variance of inter-message delays.
|
||
Low variance = likely automated.
|
||
"""
|
||
if len(profile.timestamps) < MIN_MESSAGES_FOR_TIMING:
|
||
return 0.0
|
||
|
||
timestamps = sorted(profile.timestamps)
|
||
deltas = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps) - 1)]
|
||
|
||
if not deltas:
|
||
return 0.0
|
||
|
||
# Filter out very large gaps (> 60s) — those are natural pauses
|
||
active_deltas = [d for d in deltas if d < 60.0]
|
||
if len(active_deltas) < 5:
|
||
return 0.0
|
||
|
||
mean = sum(active_deltas) / len(active_deltas)
|
||
if mean <= 0:
|
||
return 0.0
|
||
variance_ms = (sum((d - mean) ** 2 for d in active_deltas) / len(active_deltas)) * 1000
|
||
|
||
if variance_ms < TIMING_VARIANCE_CEIL_MS:
|
||
# Very precise timing — scale score by how precise
|
||
precision_factor = 1.0 - (variance_ms / TIMING_VARIANCE_CEIL_MS)
|
||
return WEIGHT_TIMING_PRECISION * (1.0 + precision_factor)
|
||
return 0.0
|
||
|
||
def _check_activity_247(self, profile: UserProfile) -> float:
|
||
"""
|
||
Heuristic 4: 24/7 activity profiling.
|
||
|
||
Normal humans sleep. If a user has sent messages across 20+ distinct
|
||
hours of the day, they're likely automated.
|
||
"""
|
||
if len(profile.hour_buckets) >= ACTIVITY_HOUR_COVERAGE:
|
||
coverage = len(profile.hour_buckets) / 24.0
|
||
return WEIGHT_ACTIVITY_247 * coverage
|
||
return 0.0
|
||
|
||
def _check_patterns(self, message: discord.Message, profile: UserProfile) -> float:
|
||
"""
|
||
Heuristic 5: Automated command-response patterns.
|
||
|
||
Selfbots often respond with the same pattern to the same triggers.
|
||
We track content prefixes and look for repetitive automated responses.
|
||
"""
|
||
content = (message.content or "").strip()
|
||
if not content or len(content) < 2:
|
||
return 0.0
|
||
|
||
# Build a simplified pattern key (first 30 chars, lowered)
|
||
pattern_key = content[:30].lower()
|
||
profile.patterns[pattern_key] += 1
|
||
|
||
# Check for repeated patterns
|
||
max_repeats = max(profile.patterns.values()) if profile.patterns else 0
|
||
if max_repeats >= PATTERN_REPEAT_THRESHOLD:
|
||
return WEIGHT_PATTERN_MATCH * (max_repeats / PATTERN_REPEAT_THRESHOLD)
|
||
return 0.0
|
||
|
||
def _check_burst(self, profile: UserProfile) -> float:
|
||
"""
|
||
Heuristic 6: Rapid-fire burst detection.
|
||
|
||
If a user sends BURST_MESSAGES+ messages within BURST_WINDOW_SECS,
|
||
that exceeds normal human typing speed.
|
||
"""
|
||
if len(profile.timestamps) < BURST_MESSAGES:
|
||
return 0.0
|
||
|
||
recent = sorted(profile.timestamps)[-BURST_MESSAGES:]
|
||
window = recent[-1] - recent[0]
|
||
if window <= BURST_WINDOW_SECS:
|
||
profile.bursts += 1
|
||
return WEIGHT_BURST * profile.bursts
|
||
return 0.0
|
||
|
||
async def _compute_score(self, message: discord.Message, profile: UserProfile) -> Tuple[float, List[str]]:
|
||
"""Run all heuristics and return (total_score, list_of_triggered_reasons)."""
|
||
reasons = []
|
||
|
||
# Apply decay first
|
||
profile.decay_score()
|
||
|
||
# Heuristic 1: Rich embeds
|
||
s1 = self._check_rich_embeds(message, profile)
|
||
if s1 > 0:
|
||
reasons.append(f"Rich embed from user account (+{s1:.0f})")
|
||
profile.score += s1
|
||
|
||
# Heuristic 2: Response speed
|
||
s2 = self._check_response_speed(message, profile)
|
||
if s2 > 0:
|
||
reasons.append(f"Inhuman response speed (<{RESPONSE_TIME_FLOOR_MS}ms) (+{s2:.0f})")
|
||
profile.score += s2
|
||
|
||
# Heuristic 3: Timing precision
|
||
s3 = self._check_timing_precision(profile)
|
||
if s3 > 0:
|
||
reasons.append(f"Precise message timing (low variance) (+{s3:.0f})")
|
||
profile.score += s3
|
||
|
||
# Heuristic 4: 24/7 activity
|
||
s4 = self._check_activity_247(profile)
|
||
if s4 > 0:
|
||
reasons.append(f"24/7 activity ({len(profile.hour_buckets)}/24 hours) (+{s4:.0f})")
|
||
profile.score += s4
|
||
|
||
# Heuristic 5: Pattern matching
|
||
s5 = self._check_patterns(message, profile)
|
||
if s5 > 0:
|
||
reasons.append(f"Repetitive automated patterns (+{s5:.0f})")
|
||
profile.score += s5
|
||
|
||
# Heuristic 6: Burst detection
|
||
s6 = self._check_burst(profile)
|
||
if s6 > 0:
|
||
reasons.append(f"Rapid-fire burst ({BURST_MESSAGES}+ msgs in {BURST_WINDOW_SECS}s) (+{s6:.0f})")
|
||
profile.score += s6
|
||
|
||
profile.last_scored = time.time()
|
||
return profile.score, reasons
|
||
|
||
# ── Punishment execution ─────────────────────────────────
|
||
|
||
async def _execute_action(
|
||
self, member: discord.Member, action: str, score: float,
|
||
reasons: List[str], mute_duration: int
|
||
) -> str:
|
||
"""Execute the configured punishment. Returns a status string."""
|
||
reason_text = f"SelfbotGuard — score {score:.0f}: {'; '.join(reasons[:3])}"
|
||
|
||
try:
|
||
if action == "log":
|
||
return "Logged (no action taken)"
|
||
|
||
elif action == "warn":
|
||
try:
|
||
embed = discord.Embed(
|
||
title="⚠️ SelfbotGuard Warning",
|
||
description=(
|
||
f"Your account has been flagged for automated behaviour in "
|
||
f"**{member.guild.name}**.\n\n"
|
||
f"If you believe this is an error, please contact the server staff."
|
||
),
|
||
colour=COLOUR_WARN,
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
embed.add_field(name="Suspicion Score", value=f"{score:.0f}", inline=True)
|
||
await member.send(embed=embed)
|
||
except discord.Forbidden:
|
||
pass
|
||
return "Warned via DM"
|
||
|
||
elif action == "mute":
|
||
duration = min(mute_duration, 2419200) # Discord max: 28 days
|
||
await member.timeout(
|
||
discord.utils.utcnow() + __import__("datetime").timedelta(seconds=duration),
|
||
reason=reason_text,
|
||
)
|
||
return f"Timed out for {duration}s"
|
||
|
||
elif action == "kick":
|
||
try:
|
||
embed = discord.Embed(
|
||
title="🚫 SelfbotGuard — Kicked",
|
||
description=(
|
||
f"You were kicked from **{member.guild.name}** for suspected "
|
||
f"selfbot activity.\nScore: {score:.0f}"
|
||
),
|
||
colour=COLOUR_ALERT,
|
||
)
|
||
await member.send(embed=embed)
|
||
except discord.Forbidden:
|
||
pass
|
||
await member.kick(reason=reason_text)
|
||
return "Kicked"
|
||
|
||
elif action == "ban":
|
||
try:
|
||
embed = discord.Embed(
|
||
title="🔨 SelfbotGuard — Banned",
|
||
description=(
|
||
f"You were banned from **{member.guild.name}** for suspected "
|
||
f"selfbot activity.\nScore: {score:.0f}"
|
||
),
|
||
colour=COLOUR_ALERT,
|
||
)
|
||
await member.send(embed=embed)
|
||
except discord.Forbidden:
|
||
pass
|
||
await member.ban(reason=reason_text, delete_message_days=0)
|
||
return "Banned"
|
||
|
||
except discord.Forbidden:
|
||
return f"Failed (missing permissions for {action})"
|
||
except discord.HTTPException as e:
|
||
return f"Failed ({e})"
|
||
|
||
return "Unknown action"
|
||
|
||
# ── Alert / logging ──────────────────────────────────────
|
||
|
||
async def _send_alert(
|
||
self, guild: discord.Guild, member: discord.Member,
|
||
score: float, reasons: List[str], action_result: str
|
||
):
|
||
"""Send an alert to the log channel."""
|
||
log_channel = await self._get_log_channel(guild)
|
||
if not log_channel:
|
||
return
|
||
|
||
embed = discord.Embed(
|
||
title="🛡️ SelfbotGuard Detection",
|
||
colour=COLOUR_ALERT if "Ban" in action_result or "Kick" in action_result else COLOUR_WARN,
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
embed.set_thumbnail(url=member.display_avatar.url if member.display_avatar else None)
|
||
embed.add_field(name="User", value=f"{member.mention} (`{member.id}`)", inline=True)
|
||
embed.add_field(name="Score", value=f"**{score:.0f}**", inline=True)
|
||
embed.add_field(name="Action", value=action_result, inline=True)
|
||
embed.add_field(
|
||
name="Triggered Heuristics",
|
||
value="\n".join(f"• {r}" for r in reasons) or "None",
|
||
inline=False,
|
||
)
|
||
embed.add_field(name="Account Created", value=discord.utils.format_dt(member.created_at, "R"), inline=True)
|
||
embed.add_field(name="Joined Server", value=discord.utils.format_dt(member.joined_at, "R") if member.joined_at else "Unknown", inline=True)
|
||
embed.set_footer(text=f"SelfbotGuard v{self.__version__}")
|
||
|
||
try:
|
||
await log_channel.send(embed=embed)
|
||
except discord.HTTPException:
|
||
pass
|
||
|
||
# Optionally DM guild owner
|
||
conf = await self.config.guild(guild).all()
|
||
if conf["notify_staff"] and guild.owner:
|
||
try:
|
||
owner_embed = discord.Embed(
|
||
title="🛡️ SelfbotGuard Alert",
|
||
description=(
|
||
f"A potential selfbot was detected in **{guild.name}**:\n"
|
||
f"**{member}** (`{member.id}`) — Score: {score:.0f}\n"
|
||
f"Action taken: {action_result}"
|
||
),
|
||
colour=COLOUR_WARN,
|
||
)
|
||
await guild.owner.send(embed=owner_embed)
|
||
except discord.Forbidden:
|
||
pass
|
||
|
||
# ── Persistence ──────────────────────────────────────────
|
||
|
||
async def _save_flagged_user(self, guild: discord.Guild, profile: UserProfile, reasons: List[str]):
|
||
"""Persist flagged user to config."""
|
||
async with self.config.guild(guild).flagged_users() as flagged:
|
||
flagged[str(profile.user_id)] = {
|
||
"score": round(profile.score, 1),
|
||
"reasons": reasons[:5],
|
||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
"action_taken": profile.action_taken,
|
||
"embed_strikes": profile.embed_strikes,
|
||
"bursts": profile.bursts,
|
||
}
|
||
|
||
# ── Background prune task ────────────────────────────────
|
||
|
||
async def _prune_loop(self):
|
||
"""Periodically prune stale user tracking data."""
|
||
await self.bot.wait_until_ready()
|
||
while True:
|
||
try:
|
||
await asyncio.sleep(3600) # Every hour
|
||
cutoff = time.time() - (PRUNE_INACTIVE_DAYS * 86400)
|
||
pruned = 0
|
||
for guild_id in list(self._profiles.keys()):
|
||
guild_profiles = self._profiles[guild_id]
|
||
stale = [
|
||
uid for uid, p in guild_profiles.items()
|
||
if p.last_active < cutoff and not p.flagged
|
||
]
|
||
for uid in stale:
|
||
del guild_profiles[uid]
|
||
pruned += 1
|
||
if not guild_profiles:
|
||
del self._profiles[guild_id]
|
||
if pruned:
|
||
log.debug("SelfbotGuard pruned %d stale user profiles.", pruned)
|
||
except asyncio.CancelledError:
|
||
break
|
||
except Exception:
|
||
log.exception("Error in SelfbotGuard prune loop")
|
||
await asyncio.sleep(60)
|
||
|
||
# ═════════════════════════════════════════════════════════
|
||
# Event listener — the core detection loop
|
||
# ═════════════════════════════════════════════════════════
|
||
|
||
@commands.Cog.listener()
|
||
async def on_message(self, message: discord.Message):
|
||
"""Analyse every message for selfbot indicators."""
|
||
# Skip DMs, bot messages, messages without a guild
|
||
if not message.guild or message.author.bot:
|
||
return
|
||
|
||
# Check if enabled
|
||
guild = message.guild
|
||
conf = await self.config.guild(guild).all()
|
||
if not conf["enabled"]:
|
||
return
|
||
|
||
member = message.author
|
||
if not isinstance(member, discord.Member):
|
||
return
|
||
|
||
# Check exemptions
|
||
if await self._is_exempt(member):
|
||
return
|
||
|
||
# Get / update profile
|
||
profile = self._get_profile(guild.id, member.id)
|
||
now = time.time()
|
||
profile.timestamps.append(now)
|
||
profile.hour_buckets.add(datetime.now(timezone.utc).hour)
|
||
profile.last_active = now
|
||
profile.last_message_ts = now
|
||
|
||
# Run heuristics
|
||
score, reasons = await self._compute_score(message, profile)
|
||
|
||
# Check threshold
|
||
threshold = conf["score_threshold"]
|
||
if score >= threshold and not profile.flagged:
|
||
profile.flagged = True
|
||
action = conf["action"]
|
||
mute_duration = conf["mute_duration"]
|
||
|
||
# Execute action
|
||
action_result = await self._execute_action(member, action, score, reasons, mute_duration)
|
||
profile.action_taken = action_result
|
||
|
||
# Log and alert
|
||
await self._send_alert(guild, member, score, reasons, action_result)
|
||
await self._save_flagged_user(guild, profile, reasons)
|
||
|
||
log.info(
|
||
"SelfbotGuard flagged user %s in guild %s (score: %.0f, action: %s)",
|
||
member.id, guild.id, score, action_result,
|
||
)
|
||
|
||
# ═════════════════════════════════════════════════════════
|
||
# Commands — Admin configuration
|
||
# ═════════════════════════════════════════════════════════
|
||
|
||
@commands.group(name="sbguard", aliases=["selfbotguard", "sbg"])
|
||
@commands.guild_only()
|
||
@commands.admin_or_permissions(administrator=True)
|
||
async def sbguard(self, ctx: commands.Context):
|
||
"""🛡️ SelfbotGuard — Advanced selfbot detection & punishment.
|
||
|
||
Detects selfbots using behavioural heuristics and takes configurable action.
|
||
Disabled by default — use `[p]sbguard enable` to activate.
|
||
"""
|
||
if ctx.invoked_subcommand is None:
|
||
await ctx.send_help(ctx.command)
|
||
|
||
@sbguard.command(name="enable", aliases=["on"])
|
||
async def sbguard_enable(self, ctx: commands.Context):
|
||
"""Enable SelfbotGuard for this server."""
|
||
await self.config.guild(ctx.guild).enabled.set(True)
|
||
embed = discord.Embed(
|
||
title="🛡️ SelfbotGuard Enabled",
|
||
description=(
|
||
"Selfbot detection is now **active** in this server.\n\n"
|
||
"• Set a log channel: `[p]sbguard logchannel #channel`\n"
|
||
"• Set the action: `[p]sbguard action <log|warn|mute|kick|ban>`\n"
|
||
"• Adjust sensitivity: `[p]sbguard sensitivity <low|medium|high>`"
|
||
),
|
||
colour=COLOUR_OK,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="disable", aliases=["off"])
|
||
async def sbguard_disable(self, ctx: commands.Context):
|
||
"""Disable SelfbotGuard for this server."""
|
||
await self.config.guild(ctx.guild).enabled.set(False)
|
||
embed = discord.Embed(
|
||
title="🛡️ SelfbotGuard Disabled",
|
||
description="Selfbot detection is now **inactive** in this server.",
|
||
colour=COLOUR_INFO,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="action")
|
||
async def sbguard_action(self, ctx: commands.Context, action: str):
|
||
"""Set the punishment action when a selfbot is detected.
|
||
|
||
Valid actions: `log`, `warn`, `mute`, `kick`, `ban`
|
||
"""
|
||
action = action.lower()
|
||
if action not in VALID_ACTIONS:
|
||
return await ctx.send(
|
||
f"❌ Invalid action `{action}`. Choose from: {', '.join(f'`{a}`' for a in VALID_ACTIONS)}"
|
||
)
|
||
await self.config.guild(ctx.guild).action.set(action)
|
||
|
||
action_descriptions = {
|
||
"log": "📝 Detections will be logged but no action taken.",
|
||
"warn": "⚠️ Users will be warned via DM.",
|
||
"mute": "🔇 Users will be timed out.",
|
||
"kick": "👢 Users will be kicked from the server.",
|
||
"ban": "🔨 Users will be banned from the server.",
|
||
}
|
||
embed = discord.Embed(
|
||
title=f"🛡️ Action Set: {action.title()}",
|
||
description=action_descriptions[action],
|
||
colour=COLOUR_OK,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="sensitivity")
|
||
async def sbguard_sensitivity(self, ctx: commands.Context, level: str):
|
||
"""Set detection sensitivity: `low`, `medium`, or `high`.
|
||
|
||
• **Low** — Score threshold 150. Fewer false positives, only catches obvious selfbots.
|
||
• **Medium** — Score threshold 100. Balanced — recommended for most servers.
|
||
• **High** — Score threshold 60. Aggressive — may flag power users, review logs carefully.
|
||
"""
|
||
level = level.lower()
|
||
if level not in SENSITIVITY_PRESETS:
|
||
return await ctx.send(
|
||
f"❌ Invalid level `{level}`. Choose from: {', '.join(f'`{l}`' for l in SENSITIVITY_PRESETS)}"
|
||
)
|
||
preset = SENSITIVITY_PRESETS[level]
|
||
await self.config.guild(ctx.guild).sensitivity.set(level)
|
||
await self.config.guild(ctx.guild).score_threshold.set(preset["score_threshold"])
|
||
|
||
embed = discord.Embed(
|
||
title=f"🛡️ Sensitivity: {level.title()}",
|
||
description=f"{preset['description']}\nScore threshold: **{preset['score_threshold']}**",
|
||
colour=COLOUR_OK,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="threshold")
|
||
async def sbguard_threshold(self, ctx: commands.Context, value: int):
|
||
"""Set a custom score threshold (overrides sensitivity preset).
|
||
|
||
Lower = more aggressive, higher = more lenient.
|
||
Recommended range: 50-200.
|
||
"""
|
||
if value < 10 or value > 500:
|
||
return await ctx.send("❌ Threshold must be between 10 and 500.")
|
||
await self.config.guild(ctx.guild).score_threshold.set(value)
|
||
embed = discord.Embed(
|
||
title="🛡️ Custom Threshold Set",
|
||
description=f"Score threshold: **{value}**",
|
||
colour=COLOUR_OK,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="logchannel", aliases=["log"])
|
||
async def sbguard_logchannel(self, ctx: commands.Context, channel: discord.TextChannel = None):
|
||
"""Set or clear the log channel for SelfbotGuard alerts.
|
||
|
||
Run without a channel to clear.
|
||
"""
|
||
if channel:
|
||
await self.config.guild(ctx.guild).log_channel.set(channel.id)
|
||
embed = discord.Embed(
|
||
title="🛡️ Log Channel Set",
|
||
description=f"Alerts will be sent to {channel.mention}",
|
||
colour=COLOUR_OK,
|
||
)
|
||
else:
|
||
await self.config.guild(ctx.guild).log_channel.set(None)
|
||
embed = discord.Embed(
|
||
title="🛡️ Log Channel Cleared",
|
||
description="No log channel configured. Set one with `[p]sbguard logchannel #channel`.",
|
||
colour=COLOUR_INFO,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="muteduration", aliases=["timeout"])
|
||
async def sbguard_muteduration(self, ctx: commands.Context, seconds: int):
|
||
"""Set the mute/timeout duration in seconds (default: 600 = 10 minutes).
|
||
|
||
Max: 2419200 (28 days — Discord limit).
|
||
"""
|
||
if seconds < 60:
|
||
return await ctx.send("❌ Minimum mute duration is 60 seconds.")
|
||
if seconds > 2419200:
|
||
return await ctx.send("❌ Maximum mute duration is 2,419,200 seconds (28 days).")
|
||
await self.config.guild(ctx.guild).mute_duration.set(seconds)
|
||
mins = seconds // 60
|
||
embed = discord.Embed(
|
||
title="🛡️ Mute Duration Set",
|
||
description=f"Timeout duration: **{seconds}s** (~{mins} minutes)",
|
||
colour=COLOUR_OK,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="exempt")
|
||
async def sbguard_exempt(self, ctx: commands.Context, target: discord.abc.Snowflake):
|
||
"""Add a role or user to the exempt list.
|
||
|
||
Exempt members bypass all selfbot detection.
|
||
Accepts a @role, @user, or raw ID.
|
||
"""
|
||
# Check if it's a role or user
|
||
if isinstance(target, discord.Role):
|
||
async with self.config.guild(ctx.guild).exempt_roles() as roles:
|
||
if target.id not in roles:
|
||
roles.append(target.id)
|
||
await ctx.send(f"✅ Role **{target.name}** is now exempt from SelfbotGuard.")
|
||
elif isinstance(target, (discord.Member, discord.User)):
|
||
async with self.config.guild(ctx.guild).exempt_users() as users:
|
||
if target.id not in users:
|
||
users.append(target.id)
|
||
await ctx.send(f"✅ User **{target}** is now exempt from SelfbotGuard.")
|
||
else:
|
||
await ctx.send("❌ Please provide a valid @role or @user mention.")
|
||
|
||
@sbguard.command(name="unexempt")
|
||
async def sbguard_unexempt(self, ctx: commands.Context, target: discord.abc.Snowflake):
|
||
"""Remove a role or user from the exempt list."""
|
||
if isinstance(target, discord.Role):
|
||
async with self.config.guild(ctx.guild).exempt_roles() as roles:
|
||
if target.id in roles:
|
||
roles.remove(target.id)
|
||
await ctx.send(f"✅ Role **{target.name}** is no longer exempt.")
|
||
elif isinstance(target, (discord.Member, discord.User)):
|
||
async with self.config.guild(ctx.guild).exempt_users() as users:
|
||
if target.id in users:
|
||
users.remove(target.id)
|
||
await ctx.send(f"✅ User **{target}** is no longer exempt.")
|
||
else:
|
||
await ctx.send("❌ Please provide a valid @role or @user mention.")
|
||
|
||
@sbguard.command(name="notifystaff", aliases=["notify"])
|
||
async def sbguard_notifystaff(self, ctx: commands.Context, toggle: bool = None):
|
||
"""Toggle DM notifications to the guild owner on detection.
|
||
|
||
Run without argument to toggle, or specify `true`/`false`.
|
||
"""
|
||
current = await self.config.guild(ctx.guild).notify_staff()
|
||
new_val = not current if toggle is None else toggle
|
||
await self.config.guild(ctx.guild).notify_staff.set(new_val)
|
||
state = "enabled" if new_val else "disabled"
|
||
embed = discord.Embed(
|
||
title=f"🛡️ Staff Notifications {state.title()}",
|
||
description=f"DM notifications to the server owner are now **{state}**.",
|
||
colour=COLOUR_OK,
|
||
)
|
||
await ctx.send(embed=embed)
|
||
|
||
# ── Status / review commands ─────────────────────────────
|
||
|
||
@sbguard.command(name="settings", aliases=["config", "status"])
|
||
async def sbguard_settings(self, ctx: commands.Context):
|
||
"""View the current SelfbotGuard configuration for this server."""
|
||
conf = await self.config.guild(ctx.guild).all()
|
||
|
||
log_ch = ctx.guild.get_channel(conf["log_channel"]) if conf["log_channel"] else None
|
||
exempt_roles = [ctx.guild.get_role(r) for r in conf["exempt_roles"] if ctx.guild.get_role(r)]
|
||
exempt_users_str = ", ".join(f"<@{u}>" for u in conf["exempt_users"][:10]) or "None"
|
||
|
||
embed = discord.Embed(
|
||
title="🛡️ SelfbotGuard Settings",
|
||
colour=COLOUR_INFO,
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
embed.add_field(name="Status", value="✅ Enabled" if conf["enabled"] else "❌ Disabled", inline=True)
|
||
embed.add_field(name="Action", value=conf["action"].title(), inline=True)
|
||
embed.add_field(name="Sensitivity", value=conf["sensitivity"].title(), inline=True)
|
||
embed.add_field(name="Score Threshold", value=str(conf["score_threshold"]), inline=True)
|
||
embed.add_field(name="Mute Duration", value=f"{conf['mute_duration']}s", inline=True)
|
||
embed.add_field(name="Notify Staff", value="✅" if conf["notify_staff"] else "❌", inline=True)
|
||
embed.add_field(name="Log Channel", value=log_ch.mention if log_ch else "Not set", inline=True)
|
||
embed.add_field(
|
||
name="Exempt Roles",
|
||
value=", ".join(r.mention for r in exempt_roles) or "None",
|
||
inline=False,
|
||
)
|
||
embed.add_field(name="Exempt Users", value=exempt_users_str, inline=False)
|
||
|
||
flagged_count = len(conf.get("flagged_users", {}))
|
||
embed.add_field(name="Flagged Users", value=str(flagged_count), inline=True)
|
||
|
||
# In-memory tracking stats
|
||
guild_profiles = self._profiles.get(ctx.guild.id, {})
|
||
embed.add_field(name="Tracking", value=f"{len(guild_profiles)} users", inline=True)
|
||
|
||
embed.set_footer(text=f"SelfbotGuard v{self.__version__}")
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="flagged", aliases=["list", "suspects"])
|
||
async def sbguard_flagged(self, ctx: commands.Context):
|
||
"""View all flagged users in this server."""
|
||
flagged = await self.config.guild(ctx.guild).flagged_users()
|
||
|
||
if not flagged:
|
||
return await ctx.send("✅ No flagged users — SelfbotGuard hasn't detected any selfbots.")
|
||
|
||
embed = discord.Embed(
|
||
title="🛡️ Flagged Users",
|
||
colour=COLOUR_WARN,
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
|
||
for uid, data in list(flagged.items())[:25]:
|
||
reasons_str = "\n".join(f" • {r}" for r in data.get("reasons", [])[:3])
|
||
embed.add_field(
|
||
name=f"<@{uid}> (`{uid}`)",
|
||
value=(
|
||
f"**Score:** {data.get('score', 0)}\n"
|
||
f"**Action:** {data.get('action_taken', 'N/A')}\n"
|
||
f"**When:** {data.get('timestamp', 'N/A')[:19]}\n"
|
||
f"{reasons_str}"
|
||
),
|
||
inline=False,
|
||
)
|
||
|
||
if len(flagged) > 25:
|
||
embed.set_footer(text=f"Showing 25 of {len(flagged)} flagged users")
|
||
else:
|
||
embed.set_footer(text=f"SelfbotGuard v{self.__version__}")
|
||
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="unflag", aliases=["clear"])
|
||
async def sbguard_unflag(self, ctx: commands.Context, user: discord.User):
|
||
"""Remove a user from the flagged list and reset their tracking data."""
|
||
async with self.config.guild(ctx.guild).flagged_users() as flagged:
|
||
if str(user.id) in flagged:
|
||
del flagged[str(user.id)]
|
||
else:
|
||
return await ctx.send(f"❌ {user} is not in the flagged list.")
|
||
|
||
# Reset in-memory profile
|
||
guild_profiles = self._profiles.get(ctx.guild.id, {})
|
||
if user.id in guild_profiles:
|
||
del guild_profiles[user.id]
|
||
|
||
await ctx.send(f"✅ **{user}** has been unflagged and their tracking data reset.")
|
||
|
||
@sbguard.command(name="scan")
|
||
async def sbguard_scan(self, ctx: commands.Context, user: discord.Member):
|
||
"""Manually check a user's current suspicion profile.
|
||
|
||
Shows the live heuristic breakdown without triggering any action.
|
||
"""
|
||
if user.bot:
|
||
return await ctx.send("ℹ️ That's an actual bot account — SelfbotGuard only tracks user accounts.")
|
||
|
||
profile = self._profiles.get(ctx.guild.id, {}).get(user.id)
|
||
if not profile:
|
||
return await ctx.send(f"ℹ️ No tracking data for **{user}** yet.")
|
||
|
||
profile.decay_score()
|
||
|
||
embed = discord.Embed(
|
||
title=f"🔍 SelfbotGuard Scan — {user}",
|
||
colour=COLOUR_INFO,
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
embed.set_thumbnail(url=user.display_avatar.url if user.display_avatar else None)
|
||
embed.add_field(name="Current Score", value=f"**{profile.score:.1f}**", inline=True)
|
||
embed.add_field(name="Flagged", value="⚠️ Yes" if profile.flagged else "✅ No", inline=True)
|
||
embed.add_field(name="Action Taken", value=profile.action_taken or "None", inline=True)
|
||
embed.add_field(name="Messages Tracked", value=str(len(profile.timestamps)), inline=True)
|
||
embed.add_field(name="Hour Coverage", value=f"{len(profile.hour_buckets)}/24", inline=True)
|
||
embed.add_field(name="Embed Strikes", value=str(profile.embed_strikes), inline=True)
|
||
embed.add_field(name="Fast Responses", value=str(len(profile.response_times_ms)), inline=True)
|
||
embed.add_field(name="Burst Events", value=str(profile.bursts), inline=True)
|
||
embed.add_field(name="Unique Patterns", value=str(len(profile.patterns)), inline=True)
|
||
|
||
threshold = await self.config.guild(ctx.guild).score_threshold()
|
||
bar_len = 20
|
||
fill = min(int((profile.score / threshold) * bar_len), bar_len)
|
||
bar = "█" * fill + "░" * (bar_len - fill)
|
||
embed.add_field(
|
||
name="Threshold Progress",
|
||
value=f"`[{bar}]` {profile.score:.0f}/{threshold}",
|
||
inline=False,
|
||
)
|
||
|
||
embed.set_footer(text=f"SelfbotGuard v{self.__version__}")
|
||
await ctx.send(embed=embed)
|
||
|
||
@sbguard.command(name="reset")
|
||
@commands.is_owner()
|
||
async def sbguard_reset(self, ctx: commands.Context, confirm: bool = False):
|
||
"""⚠️ Reset ALL SelfbotGuard data for this server (owner only).
|
||
|
||
Pass `True` to confirm: `[p]sbguard reset True`
|
||
"""
|
||
if not confirm:
|
||
return await ctx.send(
|
||
"⚠️ This will clear ALL flagged users and tracking data for this server.\n"
|
||
"Run `[p]sbguard reset True` to confirm."
|
||
)
|
||
await self.config.guild(ctx.guild).flagged_users.set({})
|
||
if ctx.guild.id in self._profiles:
|
||
del self._profiles[ctx.guild.id]
|
||
if ctx.guild.id in self._recent_messages:
|
||
del self._recent_messages[ctx.guild.id]
|
||
await ctx.send("✅ All SelfbotGuard data for this server has been reset.")
|
||
|
||
# ── Hybrid slash support ─────────────────────────────────
|
||
|
||
@app_commands.command(name="sbguard-status", description="Check SelfbotGuard status")
|
||
@app_commands.guild_only()
|
||
@app_commands.default_permissions(administrator=True)
|
||
async def slash_sbguard_status(self, interaction: discord.Interaction):
|
||
"""Slash command: view SelfbotGuard status."""
|
||
ctx = await commands.Context.from_interaction(interaction)
|
||
await self.sbguard_settings(ctx)
|
||
|
||
@app_commands.command(name="sbguard-scan", description="Scan a user for selfbot indicators")
|
||
@app_commands.guild_only()
|
||
@app_commands.default_permissions(administrator=True)
|
||
@app_commands.describe(user="The user to scan")
|
||
async def slash_sbguard_scan(self, interaction: discord.Interaction, user: discord.Member):
|
||
"""Slash command: scan a specific user."""
|
||
ctx = await commands.Context.from_interaction(interaction)
|
||
await self.sbguard_scan(ctx, user)
|