From 8a73f40271d8a83fbb39290d9d6ee5c2787e4b65 Mon Sep 17 00:00:00 2001 From: pushiscool Date: Tue, 9 Jun 2026 19:55:07 -0400 Subject: [PATCH 1/4] Add scam image detector --- requirements.txt | 2 + src/cogs/logs.py | 28 ++ src/cogs/moderation/appeal.py | 7 +- src/cogs/scam_image_detector.py | 773 ++++++++++++++++++++++++++++++ src/config.json.example | 5 + src/database_handler.py | 2 +- src/startup.py | 1 + tests/test_scam_image_detector.py | 373 ++++++++++++++ 8 files changed, 1187 insertions(+), 4 deletions(-) create mode 100644 src/cogs/scam_image_detector.py create mode 100644 tests/test_scam_image_detector.py diff --git a/requirements.txt b/requirements.txt index 30a0b87..1604f43 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,3 +4,5 @@ motor jishaku==2.4.0 python-dotenv pytest +nudenet>=3.4.2 +rapidocr-onnxruntime>=1.4.4 diff --git a/src/cogs/logs.py b/src/cogs/logs.py index 3488566..ee425f2 100644 --- a/src/cogs/logs.py +++ b/src/cogs/logs.py @@ -1,5 +1,11 @@ import nextcord from nextcord.ext import commands +from nextcord.ext.application_checks import ( + ApplicationCheckFailure, + ApplicationMissingAnyRole, + ApplicationMissingPermissions, + ApplicationMissingRole, +) import traceback import sys import json @@ -15,6 +21,13 @@ # Avoid multiple logging instances already_redirected = False +IGNORED_APPLICATION_ERRORS = ( + ApplicationCheckFailure, + ApplicationMissingAnyRole, + ApplicationMissingPermissions, + ApplicationMissingRole, +) + class StreamToDiscord(io.StringIO): def __init__(self, bot: commands.Bot, channel_id: int, stream_name: str): @@ -121,6 +134,21 @@ async def on_command_error(self, ctx: commands.Context, error: Exception): @commands.Cog.listener() async def on_application_command_error(self, interaction: nextcord.Interaction, error: Exception): + original_error = getattr(error, "original", error) + if isinstance(original_error, IGNORED_APPLICATION_ERRORS): + try: + await interaction.response.send_message( + "You do not have permission to use this command.", ephemeral=True + ) + except nextcord.errors.InteractionResponded: + try: + await interaction.followup.send( + "You do not have permission to use this command.", ephemeral=True + ) + except Exception: + pass + return + error_msg = "".join(traceback.format_exception(type(error), error, error.__traceback__)) try: diff --git a/src/cogs/moderation/appeal.py b/src/cogs/moderation/appeal.py index e1f3fe6..df28b1a 100644 --- a/src/cogs/moderation/appeal.py +++ b/src/cogs/moderation/appeal.py @@ -23,7 +23,8 @@ config_path = "config.json" conf = Config(config_path) -APPEAL_GUILD_ID = int(conf.get("ban_appeal_server")) +ban_appeal_server = conf.get("ban_appeal_server") +APPEAL_GUILD_IDS = [int(ban_appeal_server)] if ban_appeal_server is not None else [] REVIEW_DELAY_SECONDS = 60 * 60 * 24 * 14 # 2 weeks REAPPEAL_DELAY_SECONDS = 60 * 60 * 24 * 84 # 12 weeks @@ -302,7 +303,7 @@ async def create_views(self): @slash_command( name="appealbutton", default_member_permissions=0x8, - guild_ids=[APPEAL_GUILD_ID], + guild_ids=APPEAL_GUILD_IDS, ) async def send_appeal_button(self, inter: Interaction): await inter.send( @@ -323,4 +324,4 @@ async def send_appeal_button(self, inter: Interaction): def setup(bot: APBot) -> None: - bot.add_cog(BanAppeal(bot)) \ No newline at end of file + bot.add_cog(BanAppeal(bot)) diff --git a/src/cogs/scam_image_detector.py b/src/cogs/scam_image_detector.py new file mode 100644 index 0000000..4dd6428 --- /dev/null +++ b/src/cogs/scam_image_detector.py @@ -0,0 +1,773 @@ +from __future__ import annotations + +import asyncio +import os +import re +import tempfile +from dataclasses import dataclass +from io import BytesIO +from pathlib import Path +from typing import Iterable, Optional + +import nextcord +from nextcord.ext import commands + +from bot_base import APBot + + +REVIEW_USER_ID = 920819377627099166 +DEFAULT_ENABLED_BOT_IDS = {1508281890820460604} +LOG_CHANNEL_NAMES = ("logs", "server-log", "bot-logs") +FORWARD_CHANNEL_ID_KEYS = ("scam_detector_forward_channel_id", "scam_detector_review_channel_id") +SAFE_MARKER = "(image marked safe)" +DANGEROUS_MARKER = "(image marked dangerous)" +VIDEO_SAFE_MARKER = "(video marked safe)" +VIDEO_DANGEROUS_MARKER = "(video marked dangerous)" +TEXT_DANGEROUS_MARKER = "(message marked dangerous)" +IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"} +VIDEO_EXTENSIONS = {".mp4", ".mov", ".webm", ".m4v", ".mkv"} +MAX_SCAN_BYTES = 8 * 1024 * 1024 +MAX_VIDEO_SCAN_BYTES = 32 * 1024 * 1024 +MAX_OCR_FRAMES = 3 +MAX_VIDEO_FRAMES = 5 +OCR_CONFIDENCE_THRESHOLD = 0.45 +NSFW_SCORE_THRESHOLD = 0.55 + +LOW_RISK = "low" +MEDIUM_RISK = "medium" +HIGH_RISK = "high" +ALERT_THRESHOLD = 70 +REVIEW_THRESHOLD = 45 + +URL_RE = re.compile(r"https?://|discord\.gift|discord\.gg|www\.", re.IGNORECASE) +URL_TOKEN_RE = re.compile(r"https?://\S+", re.IGNORECASE) +SHORTENER_RE = re.compile( + r"\b(?:bit\.ly|tinyurl\.com|t\.co|goo\.gl|is\.gd|cutt\.ly|rebrand\.ly|rb\.gy)\b", + re.IGNORECASE, +) +SUSPICIOUS_DOMAIN_RE = re.compile( + r"\b[\w.-]+\.(?:ru|cn|top|xyz|click|link|zip|mov|work|live|site|online|shop|gift|claim|win)\b", + re.IGNORECASE, +) +MONEY_RE = re.compile(r"(?:\$|usd|cash|money|gift\s*card|nitro|robux|v-?bucks|crypto)", re.IGNORECASE) +HANDLE_RE = re.compile(r"@(?:everyone|here|[\w.-]{2,32})", re.IGNORECASE) + +BRAND_TERMS = ( + "mrbeast", + "mr beast", + "discord", + "nitro", + "steam", + "roblox", + "fortnite", + "cash app", + "paypal", + "apple", + "amazon", + "google", + "youtube", +) +GIVEAWAY_TERMS = ( + "giveaway", + "winner", + "won", + "prize", + "reward", + "free", + "limited time", + "congratulations", + "selected", + "airdrop", +) +STRONG_ACTION_TERMS = ( + "claim", + "verify", + "register", + "login", + "log in", + "sign in", + "sign up", + "redeem", + "promo code", + "scan qr", + "scan the qr", + "connect wallet", + "seed phrase", + "password", + "enter code", + "payment", + "withdrawal", +) +WEAK_ACTION_TERMS = ( + "click", + "tap", + "visit", + "page", + "continue", + "follow", + "subscribe", +) +PARODY_TERMS = ( + "joke", + "meme", + "parody", + "satire", + "fake", + "shitpost", + "sketch", + "drawing", + "drawn", +) +NSFW_CLASSES = { + "ANUS_EXPOSED", + "BUTTOCKS_EXPOSED", + "FEMALE_BREAST_EXPOSED", + "FEMALE_GENITALIA_EXPOSED", + "MALE_GENITALIA_EXPOSED", + "EXPOSED_ANUS", + "EXPOSED_BUTTOCKS", + "EXPOSED_BREAST", + "EXPOSED_BREASTS", + "EXPOSED_VAGINA", + "EXPOSED_PENIS", +} + +_rapid_ocr = None +_rapid_ocr_unavailable = False +_nude_detector = None +_nude_detector_unavailable = False + + +@dataclass(frozen=True) +class ScamAssessment: + score: int + level: str + reasons: tuple[str, ...] + scanned_text: str = "" + content_kind: str = "image" + + @property + def should_alert(self) -> bool: + return self.score >= ALERT_THRESHOLD + + +def normalize_text(value: Optional[str]) -> str: + return re.sub(r"\s+", " ", str(value or "")).strip().lower() + + +def contains_any(text: str, terms: Iterable[str]) -> bool: + return any(term in text for term in terms) + + +def score_to_level(score: int) -> str: + if score >= ALERT_THRESHOLD: + return HIGH_RISK + if score >= REVIEW_THRESHOLD: + return MEDIUM_RISK + return LOW_RISK + + +def assess_scam_text( + text: str, + *, + has_qr: bool = False, + has_visual_attachment: bool = True, + nsfw_detections: Iterable[str] = (), + content_kind: str = "image", +) -> ScamAssessment: + normalized = normalize_text(text) + score = 0 + reasons: list[str] = [] + nsfw_detections = tuple(nsfw_detections) + + if has_visual_attachment: + score += 5 + + if nsfw_detections: + score += 100 + reasons.append("detects explicit or adult visual content") + + has_brand = contains_any(normalized, BRAND_TERMS) + has_giveaway = contains_any(normalized, GIVEAWAY_TERMS) or bool(MONEY_RE.search(normalized)) + has_strong_action = contains_any(normalized, STRONG_ACTION_TERMS) + has_weak_action = contains_any(normalized, WEAK_ACTION_TERMS) + has_url = bool(URL_RE.search(normalized) or SHORTENER_RE.search(normalized) or SUSPICIOUS_DOMAIN_RE.search(normalized)) + has_parody = contains_any(normalized, PARODY_TERMS) + + if has_brand: + score += 15 + reasons.append("mentions an impersonation-prone brand or creator") + + if has_giveaway: + score += 20 + reasons.append("uses giveaway, prize, money, or reward language") + + if has_strong_action: + score += 25 + reasons.append("asks users to claim, verify, log in, scan, or enter sensitive info") + elif has_weak_action: + score += 5 + reasons.append("uses weak call-to-action language") + + if has_url: + score += 30 + reasons.append("contains a link or suspicious domain") + + if has_qr: + score += 30 + reasons.append("contains a QR code") + + if HANDLE_RE.search(normalized) and has_giveaway: + score += 10 + reasons.append("combines a mention with reward language") + + if has_brand and has_giveaway and (has_strong_action or has_url or has_qr): + score += 15 + reasons.append("combines brand impersonation, reward language, and a risky action") + + if has_brand and has_giveaway and has_weak_action and not has_strong_action and not has_url and not has_qr: + score += 25 + reasons.append("looks like a creator giveaway ad asking users to visit or continue") + + if has_parody and not has_url and not has_qr and not has_strong_action: + score = max(0, score - 25) + reasons.append("looks like a joke/parody without a link, QR code, or account action") + + score = min(score, 100) + if score < REVIEW_THRESHOLD and not reasons: + reasons.append("no strong scam signals found") + + return ScamAssessment( + score=score, + level=score_to_level(score), + reasons=tuple(reasons), + scanned_text=clip(text, 500), + content_kind=content_kind, + ) + + +def is_visual_attachment(attachment) -> bool: + content_type = normalize_text(getattr(attachment, "content_type", "")) + filename = normalize_text(getattr(attachment, "filename", "")) + suffix = Path(filename).suffix + return content_type.startswith("image/") or suffix in IMAGE_EXTENSIONS + + +def is_video_attachment(attachment) -> bool: + content_type = normalize_text(getattr(attachment, "content_type", "")) + filename = normalize_text(getattr(attachment, "filename", "")) + suffix = Path(filename).suffix + return content_type.startswith("video/") or suffix in VIDEO_EXTENSIONS + + +def clean_url(value: str) -> str: + return str(value or "").strip("<>()[]{}.,!?") + + +def is_visual_url(value: str) -> bool: + url = clean_url(value).lower() + without_query = url.split("?", 1)[0].split("#", 1)[0] + return ( + without_query.endswith(tuple(IMAGE_EXTENSIONS)) + or "tenor.com/view/" in url + or "giphy.com/gifs/" in url + or "media.discordapp.net/" in url + or "cdn.discordapp.com/" in url + ) + + +def visual_urls_from_message(message) -> list[str]: + urls = [clean_url(match.group(0)) for match in URL_TOKEN_RE.finditer(getattr(message, "content", "") or "")] + + for embed in getattr(message, "embeds", []) or []: + for attr_name in ("image", "thumbnail"): + embed_image = getattr(embed, attr_name, None) + image_url = getattr(embed_image, "url", None) + if image_url: + urls.append(clean_url(image_url)) + + return [url for url in urls if is_visual_url(url)] + + +def attachment_label(attachment) -> str: + filename = getattr(attachment, "filename", None) or "attachment" + url = getattr(attachment, "url", None) or getattr(attachment, "proxy_url", None) + if url: + return f"[{filename}]({url})" + return f"`{filename}`" + + +def attachment_lines(message) -> list[str]: + return [attachment_label(attachment) for attachment in getattr(message, "attachments", [])] + + +def clip(text: str, limit: int = 1000) -> str: + if len(text) <= limit: + return text + return text[: limit - 20] + "\n... *(truncated)*" + + +def optional_ocr_available() -> bool: + if get_rapid_ocr() is not None: + return True + + try: + import PIL.Image # noqa: F401 + import pytesseract # noqa: F401 + except ImportError: + return False + return True + + +def optional_qr_available() -> bool: + try: + import pyzbar.pyzbar # noqa: F401 + import PIL.Image # noqa: F401 + except ImportError: + return False + return True + + +def optional_nudity_available() -> bool: + return get_nude_detector() is not None + + +def get_rapid_ocr(): + global _rapid_ocr, _rapid_ocr_unavailable + if _rapid_ocr_unavailable: + return None + if _rapid_ocr is not None: + return _rapid_ocr + + try: + from rapidocr_onnxruntime import RapidOCR + except ImportError: + _rapid_ocr_unavailable = True + return None + + try: + _rapid_ocr = RapidOCR() + except Exception: + _rapid_ocr_unavailable = True + return None + + return _rapid_ocr + + +def get_nude_detector(): + global _nude_detector, _nude_detector_unavailable + if _nude_detector_unavailable: + return None + if _nude_detector is not None: + return _nude_detector + + try: + from nudenet import NudeDetector + except ImportError: + _nude_detector_unavailable = True + return None + + try: + _nude_detector = NudeDetector() + except Exception: + _nude_detector_unavailable = True + return None + + return _nude_detector + + +def extract_text_with_rapidocr(data: bytes) -> str: + engine = get_rapid_ocr() + if engine is None: + return "" + + try: + result, _ = engine(data) + except Exception: + return "" + + if not result: + return "" + + lines = [] + for item in result: + if len(item) < 3: + continue + text = str(item[1] or "").strip() + try: + confidence = float(item[2]) + except (TypeError, ValueError): + confidence = 0 + if text and confidence >= OCR_CONFIDENCE_THRESHOLD: + lines.append(text) + + return "\n".join(lines) + + +def detect_nsfw_with_nudenet(data: bytes) -> tuple[str, ...]: + detector = get_nude_detector() + if detector is None: + return () + + try: + detections = detector.detect(data) + except Exception: + return () + + unsafe = [] + for detection in detections or []: + label = normalize_text(detection.get("class", "")).upper().replace(" ", "_") + try: + score = float(detection.get("score", 0)) + except (TypeError, ValueError): + score = 0 + + if label in NSFW_CLASSES and score >= NSFW_SCORE_THRESHOLD: + unsafe.append(f"{label} ({score:.2f})") + + return tuple(unsafe) + + +def scan_video_bytes(data: bytes, suffix: str = ".mp4") -> tuple[str, bool, tuple[str, ...]]: + try: + import cv2 + except ImportError: + return "", False, () + + temp_path = None + text_parts: list[str] = [] + has_qr = False + nsfw_detections: list[str] = [] + + try: + with tempfile.NamedTemporaryFile(suffix=suffix or ".mp4", delete=False) as temp_file: + temp_file.write(data) + temp_path = temp_file.name + + capture = cv2.VideoCapture(temp_path) + if not capture.isOpened(): + return "", False, () + + frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT) or 0) + if frame_count > 0: + frame_indexes = [ + int((index + 1) * frame_count / (MAX_VIDEO_FRAMES + 1)) + for index in range(MAX_VIDEO_FRAMES) + ] + else: + frame_indexes = list(range(MAX_VIDEO_FRAMES)) + + for frame_index in frame_indexes: + if frame_index > 0: + capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index) + + ok, frame = capture.read() + if not ok: + continue + + ok, encoded = cv2.imencode(".jpg", frame) + if not ok: + continue + + frame_bytes = encoded.tobytes() + extracted_text, frame_has_qr = extract_text_from_image_bytes(frame_bytes) + if extracted_text: + text_parts.append(extracted_text) + has_qr = has_qr or frame_has_qr + nsfw_detections.extend(detect_nsfw_with_nudenet(frame_bytes)) + + capture.release() + except Exception: + return "", False, () + finally: + if temp_path: + try: + os.unlink(temp_path) + except OSError: + pass + + return "\n".join(text_parts), has_qr, tuple(nsfw_detections) + + +def extract_text_from_image_bytes(data: bytes) -> tuple[str, bool]: + text_parts: list[str] = [] + has_qr = False + rapid_text = extract_text_with_rapidocr(data) + if rapid_text: + text_parts.append(rapid_text) + + try: + from PIL import Image, ImageSequence + except ImportError: + return "", False + + try: + image = Image.open(BytesIO(data)) + except Exception: + return "", False + + frames = [] + try: + for index, frame in enumerate(ImageSequence.Iterator(image)): + if index >= MAX_OCR_FRAMES: + break + frames.append(frame.convert("RGB")) + except Exception: + frames = [image.convert("RGB")] + + try: + import pytesseract + except ImportError: + pytesseract = None + + try: + from pyzbar.pyzbar import decode as decode_qr + except ImportError: + decode_qr = None + + for frame in frames: + if pytesseract is not None: + try: + text = pytesseract.image_to_string(frame) + if text: + text_parts.append(text) + except Exception: + pass + + if decode_qr is not None: + try: + decoded = decode_qr(frame) + if decoded: + has_qr = True + for code in decoded: + data_text = getattr(code, "data", b"") + if isinstance(data_text, bytes): + data_text = data_text.decode("utf-8", errors="ignore") + if data_text: + text_parts.append(str(data_text)) + except Exception: + pass + + return "\n".join(text_parts), has_qr + + +class ScamImageDetector(commands.Cog): + def __init__(self, bot: APBot) -> None: + self.bot = bot + + def enabled_bot_ids(self) -> set[int]: + configured_ids = set(DEFAULT_ENABLED_BOT_IDS) + config = getattr(self.bot, "config", None) + + if config is not None: + application_id = config.get("application_id") + if application_id is not None: + try: + configured_ids.add(int(application_id)) + except (TypeError, ValueError): + pass + + for bot_id in config.get("scam_detector_enabled_bot_ids", []) or []: + try: + configured_ids.add(int(bot_id)) + except (TypeError, ValueError): + pass + + return configured_ids + + def is_enabled_for_current_bot(self) -> bool: + bot_user_id = getattr(getattr(self.bot, "user", None), "id", None) + return bot_user_id is None or bot_user_id in self.enabled_bot_ids() + + async def scan_attachment(self, attachment) -> tuple[str, bool, tuple[str, ...]]: + if not is_visual_attachment(attachment): + return "", False, () + + if not optional_ocr_available() and not optional_qr_available() and not optional_nudity_available(): + return "", False, () + + size = getattr(attachment, "size", 0) or 0 + if size > MAX_SCAN_BYTES: + return "", False, () + + try: + data = await attachment.read() + except (nextcord.HTTPException, AttributeError): + return "", False, () + + extracted_text, has_qr = await asyncio.to_thread(extract_text_from_image_bytes, data) + nsfw_detections = await asyncio.to_thread(detect_nsfw_with_nudenet, data) + return extracted_text, has_qr, nsfw_detections + + async def scan_video_attachment(self, attachment) -> tuple[str, bool, tuple[str, ...]]: + if not is_video_attachment(attachment): + return "", False, () + + size = getattr(attachment, "size", 0) or 0 + if size > MAX_VIDEO_SCAN_BYTES: + return "", False, () + + try: + data = await attachment.read() + except (nextcord.HTTPException, AttributeError): + return "", False, () + + suffix = Path(normalize_text(getattr(attachment, "filename", ""))).suffix or ".mp4" + return await asyncio.to_thread(scan_video_bytes, data, suffix) + + async def assess_message(self, message: nextcord.Message) -> Optional[ScamAssessment]: + visual_attachments = [ + attachment for attachment in getattr(message, "attachments", []) if is_visual_attachment(attachment) + ] + video_attachments = [ + attachment for attachment in getattr(message, "attachments", []) if is_video_attachment(attachment) + ] + visual_urls = visual_urls_from_message(message) + has_media = bool(visual_attachments or video_attachments or visual_urls) + + text_parts = [getattr(message, "content", "")] + has_qr = False + nsfw_detections = [] + content_kind = "video" if video_attachments and not visual_attachments and not visual_urls else "image" + + for attachment in visual_attachments: + text_parts.append(getattr(attachment, "filename", "")) + extracted_text, attachment_has_qr, attachment_nsfw_detections = await self.scan_attachment(attachment) + if extracted_text: + text_parts.append(extracted_text) + has_qr = has_qr or attachment_has_qr + nsfw_detections.extend(attachment_nsfw_detections) + + for attachment in video_attachments: + text_parts.append(getattr(attachment, "filename", "")) + extracted_text, attachment_has_qr, attachment_nsfw_detections = await self.scan_video_attachment(attachment) + if extracted_text: + text_parts.append(extracted_text) + has_qr = has_qr or attachment_has_qr + nsfw_detections.extend(attachment_nsfw_detections) + + assessment = assess_scam_text( + "\n".join(text_parts), + has_qr=has_qr, + has_visual_attachment=has_media, + nsfw_detections=nsfw_detections, + content_kind=content_kind if has_media else "text", + ) + if not has_media and not assessment.should_alert: + return None + return assessment + + def get_alert_channel(self, guild: nextcord.Guild, fallback) -> object: + text_channels = getattr(guild, "text_channels", []) + for channel_name in LOG_CHANNEL_NAMES: + channel = nextcord.utils.get(text_channels, name=channel_name) + if channel is not None: + return channel + return fallback + + def get_forward_channel(self, guild: nextcord.Guild, fallback) -> object: + config = getattr(self.bot, "config", None) + if config is not None: + for key in FORWARD_CHANNEL_ID_KEYS: + channel_id = config.get(key) + if channel_id is None: + continue + + try: + channel_id = int(channel_id) + except (TypeError, ValueError): + continue + + channel = None + if hasattr(self.bot, "get_channel"): + channel = self.bot.get_channel(channel_id) + if channel is None and hasattr(guild, "get_channel"): + channel = guild.get_channel(channel_id) + if channel is not None: + return channel + + return self.get_alert_channel(guild, fallback) + + def build_alert_embed(self, message: nextcord.Message, assessment: ScamAssessment) -> nextcord.Embed: + attachments = attachment_lines(message) + original_content = getattr(message, "content", "") or "" + author = getattr(message, "author", None) + author_id = getattr(author, "id", "unknown") + author_mention = getattr(author, "mention", str(author_id)) + channel_mention = getattr(getattr(message, "channel", None), "mention", "unknown channel") + + embed = nextcord.Embed( + title="Possible Unsafe Message Detected", + description=( + "This was forwarded for human review. The bot did not delete the original message or re-upload media." + ), + color=nextcord.Color.orange(), + ) + embed.add_field(name="User", value=f"{author_mention} (`{author_id}`)", inline=False) + embed.add_field(name="Channel", value=channel_mention, inline=True) + embed.add_field(name="Risk", value=f"{assessment.level.title()} ({assessment.score}/100)", inline=True) + embed.add_field(name="Reasons", value=clip("\n".join(f"- {reason}" for reason in assessment.reasons), 1000), inline=False) + if original_content: + embed.add_field(name="Original Text", value=clip(original_content, 1000), inline=False) + if assessment.scanned_text: + embed.add_field(name="Scanned Text", value=clip(f"`{assessment.scanned_text}`", 1000), inline=False) + + jump_url = getattr(message, "jump_url", None) + if jump_url: + embed.add_field(name="Message", value=f"[Jump to message]({jump_url})", inline=False) + + if attachments: + embed.add_field(name="Attachments", value=clip("\n".join(attachments), 1000), inline=False) + + return embed + + async def send_safety_marker(self, message: nextcord.Message, assessment: ScamAssessment) -> None: + if assessment.content_kind == "text" and not assessment.should_alert: + return + + if assessment.content_kind == "text": + marker = TEXT_DANGEROUS_MARKER + elif assessment.content_kind == "video": + marker = VIDEO_DANGEROUS_MARKER if assessment.should_alert else VIDEO_SAFE_MARKER + else: + marker = DANGEROUS_MARKER if assessment.should_alert else SAFE_MARKER + + try: + await message.channel.send(marker) + except (nextcord.Forbidden, nextcord.HTTPException): + pass + + @commands.Cog.listener() + async def on_message(self, message: nextcord.Message) -> None: + if not self.is_enabled_for_current_bot(): + return + + if getattr(message, "guild", None) is None: + return + + if getattr(getattr(message, "author", None), "bot", False): + return + + assessment = await self.assess_message(message) + if assessment is None: + return + + await self.send_safety_marker(message, assessment) + + if not assessment.should_alert: + return + + alert_channel = self.get_forward_channel(message.guild, message.channel) + embed = self.build_alert_embed(message, assessment) + try: + await alert_channel.send( + content=f"<@{REVIEW_USER_ID}>", + embed=embed, + allowed_mentions=nextcord.AllowedMentions(users=True, roles=False, everyone=False), + ) + except (nextcord.Forbidden, nextcord.HTTPException): + pass + + +def setup(bot: APBot) -> None: + bot.add_cog(ScamImageDetector(bot)) diff --git a/src/config.json.example b/src/config.json.example index 46fcac0..fe67513 100644 --- a/src/config.json.example +++ b/src/config.json.example @@ -16,6 +16,11 @@ 1, 1 ], + "application_id": 1, + "scam_detector_enabled_bot_ids": [ + 1 + ], + "scam_detector_forward_channel_id": 1, "command_prefix": "!!", "project_root": "APBot", "colors": { diff --git a/src/database_handler.py b/src/database_handler.py index 40ca3c5..f4b5bb5 100644 --- a/src/database_handler.py +++ b/src/database_handler.py @@ -299,7 +299,7 @@ def __call__(cls, *args, **kwargs): class BaseDatabase(metaclass=SingletonMeta): def __init__(self, conf=None): self.bot_user_id: int - self.database = database_client["ap-students"] + self.database = database_client["ap-test"] self.user_config = self.database["user_config"] self.bot_config = self.database["bot_config"] self.ban_appeals = self.database["ban_appeals"] diff --git a/src/startup.py b/src/startup.py index 03c71d0..4c633af 100644 --- a/src/startup.py +++ b/src/startup.py @@ -9,6 +9,7 @@ "cogs.recurrent", "cogs.tags", "cogs.wordle", + "cogs.scam_image_detector", "cogs.study", "cogs.helper_dm", "cogs.events", diff --git a/tests/test_scam_image_detector.py b/tests/test_scam_image_detector.py new file mode 100644 index 0000000..3a4d622 --- /dev/null +++ b/tests/test_scam_image_detector.py @@ -0,0 +1,373 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from cogs.scam_image_detector import ( + ALERT_THRESHOLD, + DANGEROUS_MARKER, + DEFAULT_ENABLED_BOT_IDS, + REVIEW_USER_ID, + SAFE_MARKER, + TEXT_DANGEROUS_MARKER, + VIDEO_DANGEROUS_MARKER, + VIDEO_SAFE_MARKER, + ScamImageDetector, + assess_scam_text, + is_visual_attachment, + is_visual_url, + is_video_attachment, +) +from startup import DEFAULT_COGS + + +def attachment(filename="scam.png", content_type="image/png", url="https://cdn.discordapp.com/scam.png"): + return SimpleNamespace( + filename=filename, + content_type=content_type, + url=url, + proxy_url=None, + size=1024, + read=AsyncMock(return_value=b""), + ) + + +def test_video_attachment_detects_video_files(): + assert is_video_attachment(attachment(filename="clip.mp4", content_type="video/mp4")) is True + assert is_video_attachment(attachment(filename="clip.webm", content_type=None)) is True + assert is_video_attachment(attachment(filename="photo.png", content_type="image/png")) is False + + +def test_scam_detector_is_loaded_by_default(): + assert "cogs.scam_image_detector" in DEFAULT_COGS + + +def test_visual_attachment_detects_images_and_gifs(): + assert is_visual_attachment(attachment(filename="giveaway.gif", content_type="image/gif")) is True + assert is_visual_attachment(attachment(filename="photo.webp", content_type=None)) is True + assert is_visual_attachment(attachment(filename="notes.txt", content_type="text/plain")) is False + + +def test_visual_url_detects_direct_images_and_gifs(): + assert is_visual_url("https://example.com/giveaway.gif") is True + assert is_visual_url("https://media.discordapp.net/attachments/1/2/image.png?width=800") is True + assert is_visual_url("https://example.com/page") is False + + +def test_joke_mrbeast_image_stays_below_alert_threshold(): + assessment = assess_scam_text( + "rough hand drawn meme of Mr Beast with +200 and continue button", + has_visual_attachment=True, + ) + + assert assessment.score < ALERT_THRESHOLD + assert assessment.should_alert is False + + +def test_realistic_mrbeast_claim_scam_alerts(): + assessment = assess_scam_text( + "MrBeast giveaway! Claim your free $500 prize now and verify at https://bit.ly/fake", + has_visual_attachment=True, + ) + + assert assessment.should_alert is True + assert assessment.level == "high" + assert "contains a link or suspicious domain" in assessment.reasons + + +def test_ocr_mrbeast_claim_scam_alerts_without_message_text(): + assessment = assess_scam_text( + "MrBeast everyone that visits our page gets $500 register claim reward", + has_visual_attachment=True, + ) + + assert assessment.should_alert is True + + +def test_mrbeast_free_money_page_ad_alerts_without_url(): + assessment = assess_scam_text( + "Ad MRBEAST everyone that visits our page gets $500 You're eligible for free $500", + has_visual_attachment=True, + ) + + assert assessment.should_alert is True + + +def test_nsfw_visual_detection_alerts(): + assessment = assess_scam_text( + "", + has_visual_attachment=True, + nsfw_detections=("FEMALE_BREAST_EXPOSED (0.91)",), + ) + + assert assessment.should_alert is True + assert "detects explicit or adult visual content" in assessment.reasons + + +def test_qr_claim_scam_alerts_even_without_link_text(): + assessment = assess_scam_text( + "Scan QR to claim free Nitro reward", + has_qr=True, + has_visual_attachment=True, + ) + + assert assessment.should_alert is True + + +def test_detector_is_enabled_only_for_target_bot_id(): + target_bot_id = next(iter(DEFAULT_ENABLED_BOT_IDS)) + + assert ScamImageDetector(SimpleNamespace(user=SimpleNamespace(id=target_bot_id))).is_enabled_for_current_bot() + assert not ScamImageDetector(SimpleNamespace(user=SimpleNamespace(id=123))).is_enabled_for_current_bot() + + +def test_detector_can_be_enabled_for_configured_application_id(): + bot = SimpleNamespace( + user=SimpleNamespace(id=456), + config=SimpleNamespace(get=lambda key, default=None: 456 if key == "application_id" else default), + ) + + assert ScamImageDetector(bot).is_enabled_for_current_bot() + + +def test_on_message_sends_staff_alert_to_logs_channel(): + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + cog = ScamImageDetector(bot) + cog.scan_attachment = AsyncMock(return_value=("", False, ())) + + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + guild = SimpleNamespace(id=1, text_channels=[logs_channel]) + message = SimpleNamespace( + guild=guild, + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=message_channel, + content="MrBeast giveaway! Claim your free $500 prize now and verify at https://bit.ly/fake", + attachments=[attachment()], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + logs_channel.send.assert_awaited_once() + assert logs_channel.send.await_args.kwargs["content"] == f"<@{REVIEW_USER_ID}>" + assert logs_channel.send.await_args.kwargs["embed"].title == "Possible Unsafe Message Detected" + message_channel.send.assert_awaited_once_with(DANGEROUS_MARKER) + + +def test_on_message_ignores_low_risk_joke_image(): + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + cog = ScamImageDetector(bot) + cog.scan_attachment = AsyncMock(return_value=("", False, ())) + + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message = SimpleNamespace( + guild=SimpleNamespace(id=1, text_channels=[logs_channel]), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=message_channel, + content="rough hand drawn meme of Mr Beast with +200 and continue button", + attachments=[attachment(filename="joke.png")], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + logs_channel.send.assert_not_awaited() + message_channel.send.assert_awaited_once_with(SAFE_MARKER) + + +def test_on_message_ignores_messages_when_running_on_other_bot(): + bot = SimpleNamespace(user=SimpleNamespace(id=123)) + cog = ScamImageDetector(bot) + cog.scan_attachment = AsyncMock(return_value=("", False, ())) + + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + message = SimpleNamespace( + guild=SimpleNamespace(id=1, text_channels=[logs_channel]), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=SimpleNamespace(name="general", mention="#general", send=AsyncMock()), + content="MrBeast giveaway! Claim your free $500 prize now and verify at https://bit.ly/fake", + attachments=[attachment()], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + logs_channel.send.assert_not_awaited() + + +def test_attachment_cdn_url_does_not_make_giveaway_joke_high_risk(): + bot = SimpleNamespace() + cog = ScamImageDetector(bot) + cog.scan_attachment = AsyncMock(return_value=("", False, ())) + + message = SimpleNamespace( + guild=SimpleNamespace(id=1), + content="MrBeast free $200 continue meme", + attachments=[attachment(url="https://cdn.discordapp.com/attachments/1/2/image.png")], + ) + + assessment = asyncio.run(cog.assess_message(message)) + + assert assessment is not None + assert assessment.score < ALERT_THRESHOLD + + +def test_ocr_attachment_text_makes_visual_message_dangerous(): + bot = SimpleNamespace() + cog = ScamImageDetector(bot) + cog.scan_attachment = AsyncMock( + return_value=("MrBeast everyone that visits our page gets $500 register claim reward", False, ()) + ) + + message = SimpleNamespace( + guild=SimpleNamespace(id=1), + content="", + attachments=[attachment(filename="screenshot.png")], + ) + + assessment = asyncio.run(cog.assess_message(message)) + + assert assessment is not None + assert assessment.should_alert is True + + +def test_direct_image_link_counts_as_visual_message(): + bot = SimpleNamespace() + cog = ScamImageDetector(bot) + message = SimpleNamespace( + guild=SimpleNamespace(id=1), + content="MrBeast giveaway claim $500 verify now https://example.xyz/free.png", + attachments=[], + embeds=[], + ) + + assessment = asyncio.run(cog.assess_message(message)) + + assert assessment is not None + assert assessment.should_alert is True + + +def test_plain_scam_text_is_marked_dangerous_without_media(): + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + cog = ScamImageDetector(bot) + + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message = SimpleNamespace( + guild=SimpleNamespace(id=1, text_channels=[logs_channel]), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=message_channel, + content="MrBeast giveaway! Claim your free $500 prize now and verify at https://bit.ly/fake", + attachments=[], + embeds=[], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + message_channel.send.assert_awaited_once_with(TEXT_DANGEROUS_MARKER) + logs_channel.send.assert_awaited_once() + sent_embed = logs_channel.send.await_args.kwargs["embed"] + assert sent_embed.title == "Possible Unsafe Message Detected" + assert any(field.name == "Original Text" for field in sent_embed.fields) + + +def test_normal_plain_text_is_ignored_without_safe_spam(): + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + cog = ScamImageDetector(bot) + + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message = SimpleNamespace( + guild=SimpleNamespace(id=1, text_channels=[logs_channel]), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=message_channel, + content="Does anyone know when the next assignment is due?", + attachments=[], + embeds=[], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + message_channel.send.assert_not_awaited() + logs_channel.send.assert_not_awaited() + + +def test_safe_video_gets_video_safe_marker(): + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + cog = ScamImageDetector(bot) + cog.scan_video_attachment = AsyncMock(return_value=("", False, ())) + + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message = SimpleNamespace( + guild=SimpleNamespace(id=1, text_channels=[logs_channel]), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=message_channel, + content="", + attachments=[attachment(filename="clip.mp4", content_type="video/mp4")], + embeds=[], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + message_channel.send.assert_awaited_once_with(VIDEO_SAFE_MARKER) + logs_channel.send.assert_not_awaited() + + +def test_explicit_video_gets_video_dangerous_marker_and_alert(): + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + cog = ScamImageDetector(bot) + cog.scan_video_attachment = AsyncMock(return_value=("", False, ("FEMALE_GENITALIA_EXPOSED (0.93)",))) + + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message = SimpleNamespace( + guild=SimpleNamespace(id=1, text_channels=[logs_channel]), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=message_channel, + content="", + attachments=[attachment(filename="clip.mp4", content_type="video/mp4")], + embeds=[], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + message_channel.send.assert_awaited_once_with(VIDEO_DANGEROUS_MARKER) + logs_channel.send.assert_awaited_once() + + +def test_dangerous_message_forwards_to_configured_review_channel(): + review_channel = SimpleNamespace(id=999, name="scam-review", send=AsyncMock()) + logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) + bot = SimpleNamespace( + user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS))), + config=SimpleNamespace( + get=lambda key, default=None: 999 if key == "scam_detector_forward_channel_id" else default + ), + get_channel=lambda channel_id: review_channel if channel_id == 999 else None, + ) + cog = ScamImageDetector(bot) + cog.scan_video_attachment = AsyncMock(return_value=("", False, ("MALE_GENITALIA_EXPOSED (0.93)",))) + + message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message = SimpleNamespace( + guild=SimpleNamespace(id=1, text_channels=[logs_channel]), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=message_channel, + content="", + attachments=[attachment(filename="clip.mp4", content_type="video/mp4", url="https://cdn.discordapp.com/clip.mp4")], + embeds=[], + jump_url="https://discord.com/channels/1/2/3", + ) + + asyncio.run(cog.on_message(message)) + + review_channel.send.assert_awaited_once() + logs_channel.send.assert_not_awaited() + sent_embed = review_channel.send.await_args.kwargs["embed"] + assert any("clip.mp4" in field.value for field in sent_embed.fields if field.name == "Attachments") From 46678fe59b25ac7ac8ac81fda20cdbf49322ea5a Mon Sep 17 00:00:00 2001 From: pushiscool Date: Fri, 19 Jun 2026 00:47:41 -0400 Subject: [PATCH 2/4] Fix scam detector: only send alerts to review channel, no pings - Remove send_safety_marker that was sending messages in the source channel - Send alerts only to hardcoded review channel (1517350483646484480) - Replace user/channel mentions with plain text to avoid pings - Use AllowedMentions.none() and no content ping - Skip scanning messages posted in the review channel itself - Improve scam detection with DM lure, off-platform, and bare domain checks Co-Authored-By: Claude Opus 4.6 --- src/cogs/scam_image_detector.py | 178 +++++++++++++++++++------------- 1 file changed, 107 insertions(+), 71 deletions(-) diff --git a/src/cogs/scam_image_detector.py b/src/cogs/scam_image_detector.py index 4dd6428..c492a98 100644 --- a/src/cogs/scam_image_detector.py +++ b/src/cogs/scam_image_detector.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import logging import os import re import tempfile @@ -15,15 +16,11 @@ from bot_base import APBot -REVIEW_USER_ID = 920819377627099166 +log = logging.getLogger(__name__) + + DEFAULT_ENABLED_BOT_IDS = {1508281890820460604} -LOG_CHANNEL_NAMES = ("logs", "server-log", "bot-logs") -FORWARD_CHANNEL_ID_KEYS = ("scam_detector_forward_channel_id", "scam_detector_review_channel_id") -SAFE_MARKER = "(image marked safe)" -DANGEROUS_MARKER = "(image marked dangerous)" -VIDEO_SAFE_MARKER = "(video marked safe)" -VIDEO_DANGEROUS_MARKER = "(video marked dangerous)" -TEXT_DANGEROUS_MARKER = "(message marked dangerous)" +REVIEW_CHANNEL_ID = 1517350483646484480 IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"} VIDEO_EXTENSIONS = {".mp4", ".mov", ".webm", ".m4v", ".mkv"} MAX_SCAN_BYTES = 8 * 1024 * 1024 @@ -49,8 +46,39 @@ r"\b[\w.-]+\.(?:ru|cn|top|xyz|click|link|zip|mov|work|live|site|online|shop|gift|claim|win)\b", re.IGNORECASE, ) -MONEY_RE = re.compile(r"(?:\$|usd|cash|money|gift\s*card|nitro|robux|v-?bucks|crypto)", re.IGNORECASE) +MONEY_RE = re.compile( + r"(?:\$\s?\d" + r"|\d[\d,]*\s*(?:k\b|grand|dollars?|usd|euros?|pounds?|bucks|thousand|million|billion)" + r"|(?:hundred|thousand|million|billion)\s+(?:dollars?|usd|euros?|pounds?|bucks)" + r"|\bdollars?\b|\beuros?\b" + r"|usd|cash(?:\s*app)?|money|gift\s*card|nitro|robux|v-?bucks|crypto|bitcoin|btc|eth|paypal|venmo|zelle)", + re.IGNORECASE, +) HANDLE_RE = re.compile(r"@(?:everyone|here|[\w.-]{2,32})", re.IGNORECASE) +DM_LURE_RE = re.compile( + r"\b(?:dm|dms|pm|pms|msg|message|inbox|hmu|contact|text|add)\s+me\b" + r"|\bdm\s+(?:me|us)\b" + r"|\bfirst\s+(?:\d+|person|people|few|one|to)\b" + r"|\bwho(?:ever)?\s+(?:dms|messages|pms|contacts)\b", + re.IGNORECASE, +) +HOOK_RE = re.compile( + r"\b(?:get|getting|become|becoming)\s+rich\b" + r"|\brich\s+quick\b" + r"|\bmake\s+money\s+fast\b" + r"|\bquick\s+(?:money|cash)\b" + r"|\bchance\s+to\s+win\b" + r"|\bwin\s+(?:\$|\d|big|one\s+(?:hundred|thousand|million|billion))", + re.IGNORECASE, +) +OFF_PLATFORM_RE = re.compile( + r"\b(?:telegram|whatsapp|whats\s*app|signal|wechat|kik|snapchat|t\.me)\b", + re.IGNORECASE, +) +BARE_DOMAIN_RE = re.compile( + r"\b[\w-]{2,}\.(?:com|net|org|io|me|gg|co|app|info|biz|live|online|site|shop|store|vip|win|top|fun|cc|tk|ml|xyz|click|link)\b", + re.IGNORECASE, +) BRAND_TERMS = ( "mrbeast", @@ -78,6 +106,11 @@ "congratulations", "selected", "airdrop", + "giving away", + "give away", + "handing out", + "free money", + "free cash", ) STRONG_ACTION_TERMS = ( "claim", @@ -188,9 +221,14 @@ def assess_scam_text( reasons.append("detects explicit or adult visual content") has_brand = contains_any(normalized, BRAND_TERMS) - has_giveaway = contains_any(normalized, GIVEAWAY_TERMS) or bool(MONEY_RE.search(normalized)) + has_money = bool(MONEY_RE.search(normalized)) + has_hook = bool(HOOK_RE.search(normalized)) + has_giveaway = contains_any(normalized, GIVEAWAY_TERMS) or has_money or has_hook has_strong_action = contains_any(normalized, STRONG_ACTION_TERMS) has_weak_action = contains_any(normalized, WEAK_ACTION_TERMS) + has_dm_lure = bool(DM_LURE_RE.search(normalized)) + has_off_platform = bool(OFF_PLATFORM_RE.search(normalized)) + has_bare_domain = bool(BARE_DOMAIN_RE.search(normalized)) has_url = bool(URL_RE.search(normalized) or SHORTENER_RE.search(normalized) or SUSPICIOUS_DOMAIN_RE.search(normalized)) has_parody = contains_any(normalized, PARODY_TERMS) @@ -209,6 +247,25 @@ def assess_scam_text( score += 5 reasons.append("uses weak call-to-action language") + if has_dm_lure: + score += 25 + reasons.append("asks people to DM, contact privately, or be the 'first' to respond") + + if has_money and has_dm_lure: + score += 30 + reasons.append("offers money or prizes in exchange for DMing or contacting privately") + elif has_giveaway and has_dm_lure: + score += 10 + reasons.append("combines giveaway or free-stuff language with a request to contact privately") + + if has_bare_domain and not has_url: + score += 20 + reasons.append("points users to an external website") + + if has_off_platform and (has_dm_lure or has_url or has_bare_domain): + score += 20 + reasons.append("tries to move users to an off-platform app like Telegram or WhatsApp") + if has_url: score += 30 reasons.append("contains a link or suspicious domain") @@ -312,8 +369,8 @@ def optional_ocr_available() -> bool: return True try: - import PIL.Image # noqa: F401 - import pytesseract # noqa: F401 + import PIL.Image + import pytesseract except ImportError: return False return True @@ -321,8 +378,8 @@ def optional_ocr_available() -> bool: def optional_qr_available() -> bool: try: - import pyzbar.pyzbar # noqa: F401 - import PIL.Image # noqa: F401 + import pyzbar.pyzbar + import PIL.Image except ImportError: return False return True @@ -555,6 +612,12 @@ class ScamImageDetector(commands.Cog): def __init__(self, bot: APBot) -> None: self.bot = bot + def _config_get(self, key, default=None): + config = getattr(self.bot, "config", None) + if config is None: + return default + return config.get(key, default) + def enabled_bot_ids(self) -> set[int]: configured_ids = set(DEFAULT_ENABLED_BOT_IDS) config = getattr(self.bot, "config", None) @@ -576,6 +639,8 @@ def enabled_bot_ids(self) -> set[int]: return configured_ids def is_enabled_for_current_bot(self) -> bool: + if getattr(self.bot, "scam_detector_always_enabled", False): + return True bot_user_id = getattr(getattr(self.bot, "user", None), "id", None) return bot_user_id is None or bot_user_id in self.enabled_bot_ids() @@ -653,48 +718,30 @@ async def assess_message(self, message: nextcord.Message) -> Optional[ScamAssess nsfw_detections=nsfw_detections, content_kind=content_kind if has_media else "text", ) + if not has_media and not assessment.should_alert: return None return assessment - def get_alert_channel(self, guild: nextcord.Guild, fallback) -> object: - text_channels = getattr(guild, "text_channels", []) - for channel_name in LOG_CHANNEL_NAMES: - channel = nextcord.utils.get(text_channels, name=channel_name) - if channel is not None: - return channel - return fallback - - def get_forward_channel(self, guild: nextcord.Guild, fallback) -> object: - config = getattr(self.bot, "config", None) - if config is not None: - for key in FORWARD_CHANNEL_ID_KEYS: - channel_id = config.get(key) - if channel_id is None: - continue - - try: - channel_id = int(channel_id) - except (TypeError, ValueError): - continue - - channel = None - if hasattr(self.bot, "get_channel"): - channel = self.bot.get_channel(channel_id) - if channel is None and hasattr(guild, "get_channel"): - channel = guild.get_channel(channel_id) - if channel is not None: - return channel - - return self.get_alert_channel(guild, fallback) + async def get_forward_channel(self): + channel = self.bot.get_channel(REVIEW_CHANNEL_ID) + if channel is not None: + return channel + try: + return await self.bot.fetch_channel(REVIEW_CHANNEL_ID) + except (AttributeError, nextcord.Forbidden, nextcord.NotFound, nextcord.HTTPException) as exc: + log.warning("Unable to access review channel %s: %s", REVIEW_CHANNEL_ID, exc) + return None def build_alert_embed(self, message: nextcord.Message, assessment: ScamAssessment) -> nextcord.Embed: attachments = attachment_lines(message) original_content = getattr(message, "content", "") or "" author = getattr(message, "author", None) author_id = getattr(author, "id", "unknown") - author_mention = getattr(author, "mention", str(author_id)) - channel_mention = getattr(getattr(message, "channel", None), "mention", "unknown channel") + author_name = str(author) if author else str(author_id) + channel = getattr(message, "channel", None) + channel_name = getattr(channel, "name", None) or "unknown channel" + channel_id = getattr(channel, "id", "?") embed = nextcord.Embed( title="Possible Unsafe Message Detected", @@ -703,8 +750,8 @@ def build_alert_embed(self, message: nextcord.Message, assessment: ScamAssessmen ), color=nextcord.Color.orange(), ) - embed.add_field(name="User", value=f"{author_mention} (`{author_id}`)", inline=False) - embed.add_field(name="Channel", value=channel_mention, inline=True) + embed.add_field(name="User", value=f"{author_name} (`{author_id}`)", inline=False) + embed.add_field(name="Channel", value=f"#{channel_name} (`{channel_id}`)", inline=True) embed.add_field(name="Risk", value=f"{assessment.level.title()} ({assessment.score}/100)", inline=True) embed.add_field(name="Reasons", value=clip("\n".join(f"- {reason}" for reason in assessment.reasons), 1000), inline=False) if original_content: @@ -721,22 +768,6 @@ def build_alert_embed(self, message: nextcord.Message, assessment: ScamAssessmen return embed - async def send_safety_marker(self, message: nextcord.Message, assessment: ScamAssessment) -> None: - if assessment.content_kind == "text" and not assessment.should_alert: - return - - if assessment.content_kind == "text": - marker = TEXT_DANGEROUS_MARKER - elif assessment.content_kind == "video": - marker = VIDEO_DANGEROUS_MARKER if assessment.should_alert else VIDEO_SAFE_MARKER - else: - marker = DANGEROUS_MARKER if assessment.should_alert else SAFE_MARKER - - try: - await message.channel.send(marker) - except (nextcord.Forbidden, nextcord.HTTPException): - pass - @commands.Cog.listener() async def on_message(self, message: nextcord.Message) -> None: if not self.is_enabled_for_current_bot(): @@ -752,21 +783,26 @@ async def on_message(self, message: nextcord.Message) -> None: if assessment is None: return - await self.send_safety_marker(message, assessment) - if not assessment.should_alert: return - alert_channel = self.get_forward_channel(message.guild, message.channel) + alert_channel = await self.get_forward_channel() + if alert_channel is None: + return + + source_channel_id = getattr(getattr(message, "channel", None), "id", None) + if source_channel_id == REVIEW_CHANNEL_ID: + return + embed = self.build_alert_embed(message, assessment) try: await alert_channel.send( - content=f"<@{REVIEW_USER_ID}>", + content=None, embed=embed, - allowed_mentions=nextcord.AllowedMentions(users=True, roles=False, everyone=False), + allowed_mentions=nextcord.AllowedMentions.none(), ) - except (nextcord.Forbidden, nextcord.HTTPException): - pass + except (nextcord.Forbidden, nextcord.HTTPException) as exc: + log.warning("Unable to forward message %s: %s", getattr(message, "id", "unknown"), exc) def setup(bot: APBot) -> None: From fa99c692641bffb62b182a30a2d6d8b1ce491100 Mon Sep 17 00:00:00 2001 From: pushiscool Date: Fri, 19 Jun 2026 22:20:04 -0400 Subject: [PATCH 3/4] Update scam detector tests for review-channel-only alerts Remove references to deleted markers and REVIEW_USER_ID, update tests to verify alerts go to the review channel without pings. Co-Authored-By: Claude Opus 4.6 --- tests/test_scam_image_detector.py | 150 +++++++++++++++++++++++------- 1 file changed, 114 insertions(+), 36 deletions(-) diff --git a/tests/test_scam_image_detector.py b/tests/test_scam_image_detector.py index 3a4d622..a534bf3 100644 --- a/tests/test_scam_image_detector.py +++ b/tests/test_scam_image_detector.py @@ -4,13 +4,9 @@ from cogs.scam_image_detector import ( ALERT_THRESHOLD, - DANGEROUS_MARKER, DEFAULT_ENABLED_BOT_IDS, - REVIEW_USER_ID, - SAFE_MARKER, - TEXT_DANGEROUS_MARKER, - VIDEO_DANGEROUS_MARKER, - VIDEO_SAFE_MARKER, + REVIEW_CHANNEL_ID, + REVIEW_THRESHOLD, ScamImageDetector, assess_scam_text, is_visual_attachment, @@ -129,13 +125,23 @@ def test_detector_can_be_enabled_for_configured_application_id(): assert ScamImageDetector(bot).is_enabled_for_current_bot() -def test_on_message_sends_staff_alert_to_logs_channel(): - bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) +def test_detector_can_be_forced_on_for_standalone_safety_bot(): + bot = SimpleNamespace(user=SimpleNamespace(id=123), scam_detector_always_enabled=True) + + assert ScamImageDetector(bot).is_enabled_for_current_bot() + + +def test_on_message_forwards_alert_without_ping_or_deletion(): + review_channel = SimpleNamespace(id=REVIEW_CHANNEL_ID, send=AsyncMock()) + bot = SimpleNamespace( + user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS))), + get_channel=lambda channel_id: review_channel if channel_id == REVIEW_CHANNEL_ID else None, + ) cog = ScamImageDetector(bot) cog.scan_attachment = AsyncMock(return_value=("", False, ())) logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) - message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message_channel = SimpleNamespace(id=123, name="general", mention="#general", send=AsyncMock()) guild = SimpleNamespace(id=1, text_channels=[logs_channel]) message = SimpleNamespace( guild=guild, @@ -144,14 +150,18 @@ def test_on_message_sends_staff_alert_to_logs_channel(): content="MrBeast giveaway! Claim your free $500 prize now and verify at https://bit.ly/fake", attachments=[attachment()], jump_url="https://discord.com/channels/1/2/3", + delete=AsyncMock(), ) asyncio.run(cog.on_message(message)) - logs_channel.send.assert_awaited_once() - assert logs_channel.send.await_args.kwargs["content"] == f"<@{REVIEW_USER_ID}>" - assert logs_channel.send.await_args.kwargs["embed"].title == "Possible Unsafe Message Detected" - message_channel.send.assert_awaited_once_with(DANGEROUS_MARKER) + review_channel.send.assert_awaited_once() + assert review_channel.send.await_args.kwargs["content"] is None + assert review_channel.send.await_args.kwargs["allowed_mentions"].users is False + assert review_channel.send.await_args.kwargs["embed"].title == "Possible Unsafe Message Detected" + logs_channel.send.assert_not_awaited() + message_channel.send.assert_not_awaited() + message.delete.assert_not_awaited() def test_on_message_ignores_low_risk_joke_image(): @@ -160,7 +170,7 @@ def test_on_message_ignores_low_risk_joke_image(): cog.scan_attachment = AsyncMock(return_value=("", False, ())) logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) - message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message_channel = SimpleNamespace(id=123, name="general", mention="#general", send=AsyncMock()) message = SimpleNamespace( guild=SimpleNamespace(id=1, text_channels=[logs_channel]), author=SimpleNamespace(id=111, mention="<@111>", bot=False), @@ -173,7 +183,7 @@ def test_on_message_ignores_low_risk_joke_image(): asyncio.run(cog.on_message(message)) logs_channel.send.assert_not_awaited() - message_channel.send.assert_awaited_once_with(SAFE_MARKER) + message_channel.send.assert_not_awaited() def test_on_message_ignores_messages_when_running_on_other_bot(): @@ -196,6 +206,21 @@ def test_on_message_ignores_messages_when_running_on_other_bot(): logs_channel.send.assert_not_awaited() +def test_on_message_ignores_the_review_channel(): + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + cog = ScamImageDetector(bot) + cog.assess_message = AsyncMock() + message = SimpleNamespace( + guild=SimpleNamespace(id=1), + author=SimpleNamespace(id=111, mention="<@111>", bot=False), + channel=SimpleNamespace(id=REVIEW_CHANNEL_ID), + ) + + asyncio.run(cog.on_message(message)) + + cog.assess_message.assert_not_awaited() + + def test_attachment_cdn_url_does_not_make_giveaway_joke_high_risk(): bot = SimpleNamespace() cog = ScamImageDetector(bot) @@ -248,12 +273,16 @@ def test_direct_image_link_counts_as_visual_message(): assert assessment.should_alert is True -def test_plain_scam_text_is_marked_dangerous_without_media(): - bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) +def test_plain_scam_text_is_forwarded_without_source_channel_marker(): + review_channel = SimpleNamespace(id=REVIEW_CHANNEL_ID, send=AsyncMock()) + bot = SimpleNamespace( + user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS))), + get_channel=lambda channel_id: review_channel if channel_id == REVIEW_CHANNEL_ID else None, + ) cog = ScamImageDetector(bot) logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) - message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message_channel = SimpleNamespace(id=123, name="general", mention="#general", send=AsyncMock()) message = SimpleNamespace( guild=SimpleNamespace(id=1, text_channels=[logs_channel]), author=SimpleNamespace(id=111, mention="<@111>", bot=False), @@ -266,9 +295,10 @@ def test_plain_scam_text_is_marked_dangerous_without_media(): asyncio.run(cog.on_message(message)) - message_channel.send.assert_awaited_once_with(TEXT_DANGEROUS_MARKER) - logs_channel.send.assert_awaited_once() - sent_embed = logs_channel.send.await_args.kwargs["embed"] + message_channel.send.assert_not_awaited() + logs_channel.send.assert_not_awaited() + review_channel.send.assert_awaited_once() + sent_embed = review_channel.send.await_args.kwargs["embed"] assert sent_embed.title == "Possible Unsafe Message Detected" assert any(field.name == "Original Text" for field in sent_embed.fields) @@ -295,13 +325,13 @@ def test_normal_plain_text_is_ignored_without_safe_spam(): logs_channel.send.assert_not_awaited() -def test_safe_video_gets_video_safe_marker(): +def test_safe_video_does_not_post_a_marker_or_alert(): bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) cog = ScamImageDetector(bot) cog.scan_video_attachment = AsyncMock(return_value=("", False, ())) logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) - message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message_channel = SimpleNamespace(id=123, name="general", mention="#general", send=AsyncMock()) message = SimpleNamespace( guild=SimpleNamespace(id=1, text_channels=[logs_channel]), author=SimpleNamespace(id=111, mention="<@111>", bot=False), @@ -314,17 +344,21 @@ def test_safe_video_gets_video_safe_marker(): asyncio.run(cog.on_message(message)) - message_channel.send.assert_awaited_once_with(VIDEO_SAFE_MARKER) + message_channel.send.assert_not_awaited() logs_channel.send.assert_not_awaited() -def test_explicit_video_gets_video_dangerous_marker_and_alert(): - bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) +def test_explicit_video_is_forwarded_without_source_channel_marker(): + review_channel = SimpleNamespace(id=REVIEW_CHANNEL_ID, send=AsyncMock()) + bot = SimpleNamespace( + user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS))), + get_channel=lambda channel_id: review_channel if channel_id == REVIEW_CHANNEL_ID else None, + ) cog = ScamImageDetector(bot) cog.scan_video_attachment = AsyncMock(return_value=("", False, ("FEMALE_GENITALIA_EXPOSED (0.93)",))) logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) - message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message_channel = SimpleNamespace(id=123, name="general", mention="#general", send=AsyncMock()) message = SimpleNamespace( guild=SimpleNamespace(id=1, text_channels=[logs_channel]), author=SimpleNamespace(id=111, mention="<@111>", bot=False), @@ -337,24 +371,22 @@ def test_explicit_video_gets_video_dangerous_marker_and_alert(): asyncio.run(cog.on_message(message)) - message_channel.send.assert_awaited_once_with(VIDEO_DANGEROUS_MARKER) - logs_channel.send.assert_awaited_once() + message_channel.send.assert_not_awaited() + logs_channel.send.assert_not_awaited() + review_channel.send.assert_awaited_once() -def test_dangerous_message_forwards_to_configured_review_channel(): - review_channel = SimpleNamespace(id=999, name="scam-review", send=AsyncMock()) +def test_dangerous_message_forwards_to_fixed_review_channel(): + review_channel = SimpleNamespace(id=REVIEW_CHANNEL_ID, name="scam-review", send=AsyncMock()) logs_channel = SimpleNamespace(name="logs", send=AsyncMock()) bot = SimpleNamespace( user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS))), - config=SimpleNamespace( - get=lambda key, default=None: 999 if key == "scam_detector_forward_channel_id" else default - ), - get_channel=lambda channel_id: review_channel if channel_id == 999 else None, + get_channel=lambda channel_id: review_channel if channel_id == REVIEW_CHANNEL_ID else None, ) cog = ScamImageDetector(bot) cog.scan_video_attachment = AsyncMock(return_value=("", False, ("MALE_GENITALIA_EXPOSED (0.93)",))) - message_channel = SimpleNamespace(name="general", mention="#general", send=AsyncMock()) + message_channel = SimpleNamespace(id=123, name="general", mention="#general", send=AsyncMock()) message = SimpleNamespace( guild=SimpleNamespace(id=1, text_channels=[logs_channel]), author=SimpleNamespace(id=111, mention="<@111>", bot=False), @@ -371,3 +403,49 @@ def test_dangerous_message_forwards_to_configured_review_channel(): logs_channel.send.assert_not_awaited() sent_embed = review_channel.send.await_args.kwargs["embed"] assert any("clip.mp4" in field.value for field in sent_embed.fields if field.name == "Attachments") + + +def test_dm_for_money_scam_text_alerts(): + assessment = assess_scam_text( + "Hello! My name is Push and I'm giving away 200 THOUSAND dollars to the first person who DMs me!", + has_visual_attachment=False, + content_kind="text", + ) + + assert assessment.should_alert is True + assert assessment.content_kind == "text" + + +def test_free_nitro_dm_scam_alerts(): + assessment = assess_scam_text("Free Nitro! DM me to claim", has_visual_attachment=False) + + assert assessment.should_alert is True + + +def test_dm_me_homework_is_ignored(): + assessment = assess_scam_text("hey can you dm me your homework notes", has_visual_attachment=False) + + assert assessment.score < REVIEW_THRESHOLD + assert assessment.should_alert is False + + +def test_first_to_finish_is_not_an_alert(): + assessment = assess_scam_text("first person to finish the quiz gets a gold star", has_visual_attachment=False) + + assert assessment.should_alert is False + + +def test_get_rich_offsite_telegram_scam_alerts(): + assessment = assess_scam_text( + "Hello, are you looking to get rich! Just go to push.com or DM me on telegram at @pushh for a chance to win one million dollars!", + has_visual_attachment=False, + content_kind="text", + ) + + assert assessment.should_alert is True + + +def test_million_dollars_spelled_out_counts_as_money(): + assessment = assess_scam_text("win one million dollars, dm me now", has_visual_attachment=False) + + assert assessment.should_alert is True From cb87ebacb15ebbe783a37a99257f75cd82db04db Mon Sep 17 00:00:00 2001 From: pushiscool Date: Fri, 19 Jun 2026 22:34:13 -0400 Subject: [PATCH 4/4] Skip review channel messages before scanning to fix test Move the REVIEW_CHANNEL_ID early-out check before assess_message so messages in the review channel are ignored without unnecessary work. Co-Authored-By: Claude Opus 4.6 --- src/cogs/scam_image_detector.py | 8 ++++---- tests/test_scam_image_detector.py | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/cogs/scam_image_detector.py b/src/cogs/scam_image_detector.py index c492a98..5e5cabc 100644 --- a/src/cogs/scam_image_detector.py +++ b/src/cogs/scam_image_detector.py @@ -779,6 +779,10 @@ async def on_message(self, message: nextcord.Message) -> None: if getattr(getattr(message, "author", None), "bot", False): return + source_channel_id = getattr(getattr(message, "channel", None), "id", None) + if source_channel_id == REVIEW_CHANNEL_ID: + return + assessment = await self.assess_message(message) if assessment is None: return @@ -790,10 +794,6 @@ async def on_message(self, message: nextcord.Message) -> None: if alert_channel is None: return - source_channel_id = getattr(getattr(message, "channel", None), "id", None) - if source_channel_id == REVIEW_CHANNEL_ID: - return - embed = self.build_alert_embed(message, assessment) try: await alert_channel.send( diff --git a/tests/test_scam_image_detector.py b/tests/test_scam_image_detector.py index a534bf3..a1de2f6 100644 --- a/tests/test_scam_image_detector.py +++ b/tests/test_scam_image_detector.py @@ -207,7 +207,7 @@ def test_on_message_ignores_messages_when_running_on_other_bot(): def test_on_message_ignores_the_review_channel(): - bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS)))) + bot = SimpleNamespace(user=SimpleNamespace(id=next(iter(DEFAULT_ENABLED_BOT_IDS))), get_channel=lambda _: None) cog = ScamImageDetector(bot) cog.assess_message = AsyncMock() message = SimpleNamespace(