diff --git a/src/bot/constants.py b/src/bot/constants.py index fb0a585..5263555 100644 --- a/src/bot/constants.py +++ b/src/bot/constants.py @@ -361,6 +361,25 @@ def format_hours_display(hours: int) -> str: "📌 [Peraturan Grup]({rules_link})" ) +# Rich message spam notification +RICH_MESSAGE_SPAM_NOTIFICATION = ( + "🚫 *Spam Terdeteksi*\n\n" + "Pesan{forward_note} dari {user_mention} telah dihapus karena mengandung " + "tombol pada pesan rich text (tombol tidak dapat dibuat dari " + "aplikasi Telegram biasa).\n\n" + "Pengguna telah dibatasi. Merasa ini kesalahan? Hubungi admin grup.\n\n" + "📌 [Peraturan Grup]({rules_link})" +) + +RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT = ( + "🚫 *Spam Terdeteksi*\n\n" + "Pesan{forward_note} dari {user_mention} telah dihapus karena mengandung " + "tombol pada pesan rich text (tombol tidak dapat dibuat dari " + "aplikasi Telegram biasa).\n\n" + "Merasa ini kesalahan? Hubungi admin grup.\n\n" + "📌 [Peraturan Grup]({rules_link})" +) + # Contact spam notification CONTACT_SPAM_NOTIFICATION = ( "🚫 *Kontak Dihapus*\n\n" diff --git a/src/bot/handlers/ai_spam_monitor.py b/src/bot/handlers/ai_spam_monitor.py index e21c381..9eb8ba1 100644 --- a/src/bot/handlers/ai_spam_monitor.py +++ b/src/bot/handlers/ai_spam_monitor.py @@ -16,7 +16,9 @@ """ import logging +import re import time +import unicodedata from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, Message, Update, User @@ -56,6 +58,7 @@ daily_budget_exhausted, try_spend_budget, ) +from bot.services.rich_message import extract_rich_text, extract_rich_urls from bot.services.telegram_utils import ( is_user_admin_in_group, is_user_admin_or_trusted, @@ -65,9 +68,34 @@ logger = logging.getLogger(__name__) +_ZERO_WIDTH_RE = re.compile(r"[\u200b-\u200d\ufeff\u202a-\u202e\u2060-\u2064]") + + +def _normalize_for_classification(text: str) -> str: + """Light normalization for classifier input. + + NFKC + strip zero-width and bidi-override characters so obfuscated + (especially rich-block) text reaches the classifier clean. Unlike the + bio-bait normalizer this keeps case and punctuation intact. + """ + return _ZERO_WIDTH_RE.sub("", unicodedata.normalize("NFKC", text)) + + def _classifiable_text(message: Message) -> str: - """Return the text available for classification: body text or media caption.""" - return message.text or message.caption or "" + """Return the text available for classification. + + Covers body text, media captions, and rich message content: rich + messages arrive with empty ``text`` while their payload (headings, + table cells, button labels and URLs) lives in ``rich_message`` + blocks, so that content is extracted too. + """ + parts = [message.text or "", message.caption or ""] + rich_message = getattr(message, "rich_message", None) + rich_text = extract_rich_text(rich_message) + if rich_text: + parts.append(rich_text) + parts.extend(extract_rich_urls(rich_message)) + return _normalize_for_classification(" ".join(part for part in parts if part)) def ai_spam_filter(update: Update) -> bool: @@ -83,6 +111,7 @@ def ai_spam_filter(update: Update) -> bool: AI_SPAM_MIN_LENGTH = 20 AI_SPAM_ALERT_MESSAGE_MAX_LEN = 500 +AI_SPAM_CLASSIFY_MAX_LEN = 8000 ALERTS_KEY = "ai_spam_alerts" ALERTS_MAX_SIZE = 500 @@ -217,7 +246,7 @@ async def _classify_and_alert( return result = await classify_text( - message_text, + message_text[:AI_SPAM_CLASSIFY_MAX_LEN], labels=list(AI_SPAM_LABELS), instructions=AI_SPAM_INSTRUCTIONS, timeout=settings.classifier_timeout_seconds, diff --git a/src/bot/handlers/anti_spam.py b/src/bot/handlers/anti_spam.py index b29cb7b..dec6322 100644 --- a/src/bot/handlers/anti_spam.py +++ b/src/bot/handlers/anti_spam.py @@ -4,6 +4,7 @@ This module enforces anti-spam rules including: - Contact card spam detection (all members) - Inline keyboard URL spam detection (all members) +- Rich message button spam detection (all members) - Probation enforcement for new users (forwards, links, external replies, stories, media) """ @@ -23,10 +24,13 @@ NEW_USER_SPAM_RESTRICTION, NEW_USER_SPAM_WARNING, RESTRICTED_PERMISSIONS, + RICH_MESSAGE_SPAM_NOTIFICATION, + RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT, format_hours_display, ) from bot.database.service import get_database from bot.group_config import get_group_config_for_update +from bot.services.rich_message import extract_rich_urls, has_rich_buttons from bot.services.telegram_utils import ( get_user_mention, is_url_whitelisted, @@ -159,6 +163,10 @@ def has_non_whitelisted_link(message: Message) -> bool: """ Check if a message contains non-whitelisted URLs. + Covers URLs in text/caption entities as well as URLs hidden in rich + message blocks, so probation users cannot smuggle links past the + filter inside rich text. + Args: message: Telegram message to check. @@ -166,6 +174,7 @@ def has_non_whitelisted_link(message: Message) -> bool: bool: True if message contains non-whitelisted links. """ urls = extract_urls(message) + urls.extend(extract_rich_urls(getattr(message, "rich_message", None))) if not urls: return False @@ -221,6 +230,26 @@ def has_non_whitelisted_inline_keyboard_urls(message: Message) -> bool: return False +def has_rich_message_buttons(message: Message) -> bool: + """Check if a message's rich content contains any buttons. + + No stock Telegram client can compose rich messages or their buttons — + they are only sendable via the Bot API (bots) or crafted over MTProto + (spammers), or forwarded. The same "not client-craftable" rationale as + inline keyboard spam applies: buttons in a group rich message from a + human sender are treated as spam, regardless of where the buttons + point. URL-only rich content without buttons is left to the AI + monitor, mirroring how plain text links are handled. + + Args: + message: Telegram message to check. + + Returns: + bool: True if any rich block or inline text carries a button. + """ + return has_rich_buttons(getattr(message, "rich_message", None)) + + def has_contact(message: Message) -> bool: """ Check if a message contains a contact card. @@ -243,6 +272,7 @@ async def _handle_group_spam( should_restrict, template_restricted: str, template_no_restrict: str, + extra_format: dict[str, str] | None = None, ) -> None: """ Shared helper for handling group spam detection and enforcement. @@ -258,6 +288,7 @@ async def _handle_group_spam( should_restrict: Callable taking group_config and returning bool. template_restricted: Notification template when user is restricted. template_no_restrict: Notification template when user is not restricted. + extra_format: Additional template placeholders (e.g. forward_note). """ message = update.message or update.edited_message if not message or not message.from_user: @@ -319,6 +350,7 @@ async def _handle_group_spam( notification_text = template.format( user_mention=user_mention, rules_link=group_config.rules_link, + **(extra_format or {}), ) await context.bot.send_message( chat_id=group_config.group_id, @@ -364,16 +396,39 @@ async def handle_inline_keyboard_spam( update: Update, context: HandlerContext ) -> None: """ - Handle spam messages containing inline keyboard buttons with non-whitelisted URLs. + Handle spam messages with non-client-craftable buttons. - Regular users cannot create inline keyboards from the Telegram client. - Any group message with inline keyboard URL buttons pointing to - non-whitelisted domains is treated as spam. + Regular users cannot create inline keyboards — or rich message buttons — + from the Telegram client. Any group message with inline keyboard URL + buttons pointing to non-whitelisted domains, or with rich message + buttons at all, is treated as spam: the message is deleted and the + sender restricted. Args: update: Telegram update containing the message. context: Bot context with helper methods. """ + message = update.message or update.edited_message + if message is None: + return + if getattr(message, "is_automatic_forward", False): + # Linked-channel auto-forwards are admin-configured; never punish + # the channel (or the 777000 service sender) for its content. + return + if has_rich_message_buttons(message): + await _handle_group_spam( + update, + context, + detector=has_rich_message_buttons, + label="rich message spam", + should_restrict=lambda cfg: True, + template_restricted=RICH_MESSAGE_SPAM_NOTIFICATION, + template_no_restrict=RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT, + extra_format={ + "forward_note": " terusan" if is_forwarded(message) else "" + }, + ) + return await _handle_group_spam( update, context, diff --git a/src/bot/handlers/bio_bait.py b/src/bot/handlers/bio_bait.py index 19ae0d8..a14becd 100644 --- a/src/bot/handlers/bio_bait.py +++ b/src/bot/handlers/bio_bait.py @@ -43,6 +43,7 @@ WHITELISTED_TELEGRAM_PATHS, ) from bot.group_config import get_group_config_for_update +from bot.services.rich_message import extract_rich_text from bot.services.telegram_utils import ( get_user_mention, is_url_whitelisted, @@ -425,7 +426,11 @@ async def handle_bio_bait_spam( if is_user_admin_or_trusted(context, group_config.group_id, user.id): return - text = message.text or message.caption or "" + text = ( + message.text + or message.caption + or extract_rich_text(getattr(message, "rich_message", None)) + ) detection_reason: str | None = None user_bio: str | None = None diff --git a/src/bot/handlers/duplicate_spam.py b/src/bot/handlers/duplicate_spam.py index bc8eb24..475d9a7 100644 --- a/src/bot/handlers/duplicate_spam.py +++ b/src/bot/handlers/duplicate_spam.py @@ -33,6 +33,7 @@ ) from bot.group_config import GroupConfig, get_group_config_for_update from bot.handlers.anti_spam import has_non_whitelisted_link +from bot.services.rich_message import extract_rich_text from bot.services.telegram_utils import ( get_user_mention, is_user_admin_or_trusted, @@ -172,7 +173,11 @@ async def handle_duplicate_spam( if is_user_admin_or_trusted(context, group_config.group_id, user.id): return - text = message.text or message.caption + text = ( + message.text + or message.caption + or extract_rich_text(getattr(message, "rich_message", None)) + ) is_media = False if text: normalized = normalize_text(text) diff --git a/src/bot/services/rich_message.py b/src/bot/services/rich_message.py new file mode 100644 index 0000000..2fcaf64 --- /dev/null +++ b/src/bot/services/rich_message.py @@ -0,0 +1,230 @@ +"""Extract searchable text and URLs from Telegram rich messages. + +Rich messages (``Message.rich_message``, Bot API 10.1+/aiogram 3.29+) +carry their payload in page blocks instead of ``message.text``. Spam +filters that only read ``text``/``caption`` are blind to them — a rich +spam message arrives with empty text while its links and buttons hide +in the block tree. This module walks that tree and returns plain text +and URLs for the spam detectors. + +The walker is duck-typed on purpose: aiogram exposes ~25 block types +and a ~30-variant rich-text union, and a newly added variant must not +silently become invisible to the filters again. +""" + +from __future__ import annotations + +import logging +import re +from typing import Any +from urllib.parse import urlparse + +logger = logging.getLogger(__name__) + +_URL_SHAPE_RE = re.compile(r"https?://[^\s<>\"]+", re.IGNORECASE) + + +def _url_shaped_candidates(text: str) -> list[str]: + """Find URL-shaped strings inside plain text. + + Bare URLs in rich blocks have no entity metadata (unlike text + messages, where Telegram tags them), so shape-check them here. + Requires a dotted hostname to avoid flagging fragments like + ``https://`` on their own. + """ + out: list[str] = [] + for match in _URL_SHAPE_RE.finditer(text): + candidate = match.group(0).rstrip(".,;:!?)") + try: + hostname = urlparse(candidate).hostname + except Exception: + continue + if hostname and "." in hostname: + out.append(candidate) + return list(dict.fromkeys(out)) + +# Attributes that hold nested blocks, cells, buttons, or list items. +# ``button`` is singular (``RichTextButton.button``); ``credit`` carries +# rich text inside quotation/media captions (``RichBlockCaption.credit``). +_CONTAINER_ATTRS = ("blocks", "cells", "buttons", "items", "button") +# Text-bearing attributes besides the generic ``text``. ``name`` (anchor / +# reference ids) is deliberately skipped: internal link targets, not content. +_TEXT_ATTRS = ( + "caption", + "summary", + "alternative_text", + "credit", + "label", + "forward_text", +) +# Sub-objects whose own fields are walked for text/URLs: copy-text clipboard +# payloads, login URLs, web apps, inline-query chat pickers. +_PAYLOAD_ATTRS = ( + "login_url", + "web_app", + "copy_text", + "switch_inline_query_chosen_chat", +) + +# Nesting deeper than this cannot occur in genuine payloads (Telegram +# documents depth <= 16 for sending); deeper trees are treated as hostile. +_MAX_WALK_DEPTH = 64 +# Total nodes visited per walk. Depth alone cannot bound the walk: an +# object with every attribute truthy (e.g. unittest mocks, or hostile +# crafted objects) branches exponentially, so the visited-node count +# is capped as well. Genuine payloads stay far below this. +_MAX_WALK_NODES = 10_000 + + +def _is_rich_node(value: Any) -> bool: + """Check whether a value is a traversable rich object. + + Rich nodes always come from aiogram's parser. Anything else — + mocks, foreign objects, hostile lookalikes — is not content and + must not be descended into (mocks in particular generate infinite + attribute trees). + """ + return type(value).__module__.split(".")[0] == "aiogram" + + +def _walk( + node: Any, + texts: list[str], + urls: list[str], + depth: int = 0, + budget: list[int] | None = None, +) -> None: + """Recursively collect text fragments and URLs from a rich node.""" + if node is None or depth > _MAX_WALK_DEPTH: + return + if budget is not None: + if budget[0] <= 0: + return + budget[0] -= 1 + if isinstance(node, str): + stripped = node.strip() + if stripped: + texts.append(stripped) + return + if isinstance(node, (list, tuple)): + for item in node: + _walk(item, texts, urls, depth + 1, budget) + return + if not _is_rich_node(node): + return + + url = getattr(node, "url", None) + if isinstance(url, str) and url.strip(): + urls.append(url.strip()) + for attr in _PAYLOAD_ATTRS: + value = getattr(node, attr, None) + if value is not None: + _walk(value, texts, urls, depth + 1, budget) + + for attr in ("text",) + _TEXT_ATTRS: + value = getattr(node, attr, None) + if value is not None: + _walk(value, texts, urls, depth + 1, budget) + + for attr in _CONTAINER_ATTRS: + value = getattr(node, attr, None) + if value: + _walk(value, texts, urls, depth + 1, budget) + + +def extract_rich_text(rich_message: Any | None) -> str: + """Return searchable plain text from a rich message, or "" when none. + + Covers headings, paragraphs, table cells, button labels, and custom + emoji alternative text. URLs are intentionally excluded here — use + :func:`extract_rich_urls` for those. + """ + if rich_message is None: + return "" + texts: list[str] = [] + try: + _walk(rich_message, texts, [], 0, [_MAX_WALK_NODES]) + except Exception: + logger.info("extract_rich_text: failed to walk rich message") + return "" + return " ".join(texts) + + +def extract_rich_urls(rich_message: Any | None) -> list[str]: + """Return deduplicated URLs from a rich message. + + Covers explicit URL fields (buttons, text links, login/web-app URLs) + plus URL-shaped strings in plain text: rich blocks carry no entity + metadata, so a bare ``https://…`` in a paragraph or a copy-text + payload would otherwise evade link checks. + """ + if rich_message is None: + return [] + urls: list[str] = [] + texts: list[str] = [] + try: + _walk(rich_message, texts, urls, 0, [_MAX_WALK_NODES]) + except Exception: + logger.info("extract_rich_urls: failed to walk rich message") + return [] + for text in texts: + urls.extend(_url_shaped_candidates(text)) + return list(dict.fromkeys(urls)) + + +def has_rich_buttons(rich_message: Any | None) -> bool: + """Return True if a rich message contains any buttons. + + Buttons appear as ``RichBlockButtons`` blocks or as inline + ``RichTextButton`` nodes in rich text. No stock Telegram client can + compose buttons, so their presence marks the message as + automation-crafted — the same rationale as the inline keyboard spam + rule. + + The check is duck-typed (block containers plus the inline button + class name) so new button-carrying variants do not silently slip + through. + """ + if rich_message is None: + return False + found = False + budget = [_MAX_WALK_NODES] + + def _visit(node: Any, depth: int = 0) -> None: + nonlocal found + if found or node is None or isinstance(node, str): + return + if depth > _MAX_WALK_DEPTH or budget[0] <= 0: + # Absurd nesting or an exploding attribute tree is only + # craftable, never client-composed: treat as suspicious + # rather than failing open. + found = True + return + budget[0] -= 1 + if isinstance(node, (list, tuple)): + for item in node: + _visit(item, depth + 1) + return + if not _is_rich_node(node): + return + if getattr(node, "buttons", None): + found = True + return + if getattr(node, "button", None) is not None: + # Inline text button (``RichTextButton``); the only node type + # with a singular ``button`` attribute. + found = True + return + for attr in ("text",) + _TEXT_ATTRS + _CONTAINER_ATTRS: + value = getattr(node, attr, None) + if value: + _visit(value, depth + 1) + if found: + return + + try: + _visit(rich_message) + except Exception: + logger.info("has_rich_buttons: failed to walk rich message") + return False + return found diff --git a/tests/test_bio_bait.py b/tests/test_bio_bait.py index 86e6137..ef36284 100644 --- a/tests/test_bio_bait.py +++ b/tests/test_bio_bait.py @@ -3,7 +3,12 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from aiogram.types import Message, User +from aiogram.types import ( + Message, + RichBlockParagraph, + RichMessage, + User, +) from bot.dispatch import AppState, HandlerContext, StopPropagation @@ -375,6 +380,17 @@ async def test_uses_caption_when_no_text(self, mock_update, mock_context, group_ await handle_bio_bait_spam(mock_update, mock_context) mock_update.message.delete.assert_called_once() + async def test_detects_bait_in_rich_text(self, mock_update, mock_context, group_config): + mock_update.message.text = None + mock_update.message.caption = None + mock_update.message.rich_message = RichMessage( + blocks=[RichBlockParagraph(text="cek bio aku dong")] + ) + with patch("bot.handlers.bio_bait.get_group_config_for_update", return_value=group_config): + with pytest.raises(StopPropagation): + await handle_bio_bait_spam(mock_update, mock_context) + mock_update.message.delete.assert_called_once() + async def test_detects_via_bio_links_with_innocuous_message( self, mock_update, mock_context, group_config ): diff --git a/tests/test_duplicate_spam.py b/tests/test_duplicate_spam.py index 8a51c4e..b57af4f 100644 --- a/tests/test_duplicate_spam.py +++ b/tests/test_duplicate_spam.py @@ -11,11 +11,14 @@ Message, MessageEntity, PhotoSize, + RichBlockParagraph, + RichMessage, Sticker, Update, User, ) + from bot.dispatch import AppState, HandlerContext, StopPropagation from bot.group_config import GroupConfig @@ -32,6 +35,22 @@ DEFAULT_SIMILARITY = 0.95 +def _make_rich_update(message_id, rich_text): + """Build a real aiogram Update carrying a text-only rich message.""" + user = User(id=42, is_bot=False, first_name="Test", username="testuser") + chat = Chat(id=-100, type="group", title="Test") + msg = Message( + message_id=message_id, + date=datetime.now(UTC), + chat=chat, + from_user=user, + rich_message=RichMessage( + blocks=[RichBlockParagraph(text=rich_text)] + ), + ) + return Update(update_id=message_id, message=msg) + + def _make_sticker(file_unique_id): """Build a real aiogram Sticker with the given file_unique_id.""" return Sticker( @@ -326,6 +345,50 @@ async def test_identical_stickers_trigger_restriction(self, mock_context, group_ assert mock_context.bot.delete_message.call_count == 2 mock_context.bot.restrict_chat_member.assert_called_once() + async def test_identical_rich_text_triggers_restriction( + self, mock_context, group_config + ): + """Rich-only messages are tracked by their extracted rich text.""" + rich_text = "Barangkali di sini ada yang sedang mencari kerja bisa menghubungi saya" + with patch( + "bot.handlers.duplicate_spam.get_group_config_for_update", + return_value=group_config, + ): + with patch( + "bot.handlers.duplicate_spam.get_user_mention", return_value="@testuser" + ): + await handle_duplicate_spam( + _make_rich_update(100, rich_text), mock_context + ) + with pytest.raises(StopPropagation): + await handle_duplicate_spam( + _make_rich_update(101, rich_text), mock_context + ) + + assert mock_context.bot.delete_message.call_count == 2 + mock_context.bot.restrict_chat_member.assert_called_once() + + async def test_different_rich_text_not_counted( + self, mock_context, group_config + ): + with patch( + "bot.handlers.duplicate_spam.get_group_config_for_update", + return_value=group_config, + ): + for i, text in enumerate( + ( + "the quick brown fox jumps over the lazy dog today", + "python packaging with uv is fast and reproducible", + "jakarta traffic during rush hour is unbearable", + ) + ): + await handle_duplicate_spam( + _make_rich_update(100 + i, text), mock_context + ) + + mock_context.bot.delete_message.assert_not_called() + mock_context.bot.restrict_chat_member.assert_not_called() + async def test_different_stickers_not_counted(self, mock_context, group_config): with patch("bot.handlers.duplicate_spam.get_group_config_for_update", return_value=group_config): for i in range(3): diff --git a/tests/test_properties.py b/tests/test_properties.py index 0550b72..1aa4362 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -9,6 +9,8 @@ """ +from urllib.parse import urlparse + import pytest from hypothesis import HealthCheck, assume, given, settings from hypothesis import strategies as st @@ -17,6 +19,7 @@ from bot.handlers.ai_spam_monitor import flagged_aspects from bot.handlers.trust import _format_person, _format_person_with_username from bot.services.classifier_client import ClassificationResult +from bot.services.rich_message import _url_shaped_candidates from bot.services.telegram_utils import is_url_whitelisted # Strategies ----------------------------------------------------------------- @@ -300,3 +303,30 @@ def test_legacy_single_label_path_follows_whitelist(self, threshold: float) -> N @given(st.integers(min_value=0, max_value=1000)) def test_sanity_integer_strategy(x: int) -> None: assert 0 <= x <= 1000 + + +class TestUrlShapedCandidates: + """Invariants for rich-block bare-URL detection.""" + + @given(st.text(max_size=200)) + @settings(max_examples=200) + def test_no_http_means_no_candidates(self, text: str) -> None: + """Without an http(s) marker there is nothing URL-shaped.""" + assume("http" not in text.lower()) + assert _url_shaped_candidates(text) == [] + + @given(st.text(max_size=200)) + @settings(max_examples=200) + def test_candidates_are_well_formed_urls(self, text: str) -> None: + """Every candidate has an http(s) scheme and a dotted hostname.""" + for candidate in _url_shaped_candidates(text): + lowered = candidate.lower() + assert lowered.startswith(("http://", "https://")) + hostname = urlparse(candidate).hostname or "" + assert "." in hostname + + @given(st.text(max_size=200)) + @settings(max_examples=200) + def test_no_duplicates(self, text: str) -> None: + candidates = _url_shaped_candidates(text) + assert len(candidates) == len(set(candidates)) diff --git a/tests/test_rich_message.py b/tests/test_rich_message.py new file mode 100644 index 0000000..580c1d3 --- /dev/null +++ b/tests/test_rich_message.py @@ -0,0 +1,768 @@ +"""Tests for rich message text/URL extraction and rich spam detection.""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from aiogram.types import ( + Chat, + CopyTextButton, + InlineKeyboardButton, + InlineKeyboardMarkup, + LoginUrl, + Message, + RichBlockButtons, + RichBlockList, + RichBlockListItem, + RichBlockParagraph, + RichBlockPullQuotation, + RichBlockSectionHeading, + RichBlockTable, + RichBlockTableCell, + RichMessage, + RichMessageButton, + RichTextBold, + RichTextButton, + RichTextCustomEmoji, + RichTextUrl, + Update, + User, + WebAppInfo, +) + +from bot.dispatch import AppState, HandlerContext, StopPropagation +from bot.group_config import GroupConfig +from bot.handlers import ai_spam_monitor +from bot.handlers.ai_spam_monitor import ( + AI_SPAM_CLASSIFY_MAX_LEN, + _classifiable_text, +) +from bot.handlers.anti_spam import ( + handle_inline_keyboard_spam, + has_non_whitelisted_link, + has_rich_message_buttons, +) +from bot.services.rich_message import extract_rich_text, extract_rich_urls, has_rich_buttons + +SPAM_URL = "https://t.me/base6d1ef16_bot?startapp=138762" + + +def _spam_blocks(): + """Rich blocks mirroring the 2026-10-17 adult-spam incident.""" + return [ + RichBlockSectionHeading(text="HOT 2026 DROP", size=1), + RichBlockParagraph( + text=[ + RichTextBold(text="FRESH LEAKS"), + " ", + RichTextCustomEmoji( + custom_emoji_id="5375191900686927814", alternative_text="🍑" + ), + ] + ), + RichBlockTable( + cells=[ + [ + RichBlockTableCell( + text=RichTextUrl(text="BIKINI SHOWS", url=SPAM_URL), + align="center", + valign="middle", + ), + RichBlockTableCell( + text="EXCLUSIVE PICS", align="center", valign="middle" + ), + ] + ] + ), + RichBlockButtons( + buttons=[ + RichMessageButton(text="OPEN LINK", url=SPAM_URL), + RichMessageButton(text="ENTER NOW", url=SPAM_URL), + ] + ), + ] + + +def _make_message(blocks=None, text=None): + return Message( + message_id=1, + date=datetime.now(UTC), + chat=Chat(id=-100123456, type="supergroup"), + text=text, + rich_message=RichMessage(blocks=blocks) if blocks is not None else None, + ) + + +class TestExtractRichText: + def test_extracts_heading_paragraph_table_and_buttons(self): + text = extract_rich_text(RichMessage(blocks=_spam_blocks())) + for expected in ( + "HOT 2026 DROP", + "FRESH LEAKS", + "🍑", + "BIKINI SHOWS", + "EXCLUSIVE PICS", + "OPEN LINK", + "ENTER NOW", + ): + assert expected in text + + def test_none_returns_empty(self): + assert extract_rich_text(None) == "" + + def test_empty_blocks_returns_empty(self): + assert extract_rich_text(RichMessage(blocks=[])) == "" + + def test_plain_string_blocks(self): + text = extract_rich_text( + RichMessage(blocks=[RichBlockParagraph(text="hello world")]) + ) + assert text == "hello world" + + def test_quotation_credit_included(self): + blocks = [ + RichBlockPullQuotation(text="a wise quote", credit="— cited source") + ] + text = extract_rich_text(RichMessage(blocks=blocks)) + assert "a wise quote" in text + assert "cited source" in text + + def test_inline_text_button_content_included(self): + blocks = [ + RichBlockParagraph( + text=[ + "tap ", + RichTextButton( + button=RichMessageButton(text="here", url=SPAM_URL) + ), + ] + ) + ] + text = extract_rich_text(RichMessage(blocks=blocks)) + assert "tap" in text + assert "here" in text + + def test_walk_failure_returns_empty(self): + class _Boom: + @property + def url(self): + raise RuntimeError("boom") + + assert extract_rich_text(_Boom()) == "" + + +class TestExtractRichUrls: + def test_extracts_button_and_text_link_urls_deduped(self): + urls = extract_rich_urls(RichMessage(blocks=_spam_blocks())) + assert urls == [SPAM_URL] + + def test_inline_text_button_url_extracted(self): + blocks = [ + RichBlockParagraph( + text=RichTextButton( + button=RichMessageButton(text="here", url=SPAM_URL) + ) + ) + ] + assert extract_rich_urls(RichMessage(blocks=blocks)) == [SPAM_URL] + + def test_bare_url_in_paragraph_detected(self): + blocks = [RichBlockParagraph(text="visit https://spam.example/deal now")] + assert extract_rich_urls(RichMessage(blocks=blocks)) == [ + "https://spam.example/deal" + ] + + def test_url_fragment_without_host_ignored(self): + blocks = [RichBlockParagraph(text="learn about the https:// protocol")] + assert extract_rich_urls(RichMessage(blocks=blocks)) == [] + + def test_none_returns_empty(self): + assert extract_rich_urls(None) == [] + + def test_no_urls_returns_empty(self): + assert ( + extract_rich_urls( + RichMessage(blocks=[RichBlockParagraph(text="no links here")]) + ) + == [] + ) + + def test_walk_failure_returns_empty(self): + class _Boom: + @property + def url(self): + raise RuntimeError("boom") + + assert extract_rich_urls(_Boom()) == [] + + +class TestHasRichButtons: + def test_block_buttons_detected(self): + assert has_rich_buttons(RichMessage(blocks=_spam_blocks())) is True + + def test_inline_text_button_detected(self): + blocks = [ + RichBlockParagraph( + text=[ + "click ", + RichTextButton( + button=RichMessageButton(text="here", url="https://example.com") + ), + ] + ) + ] + assert has_rich_buttons(RichMessage(blocks=blocks)) is True + + def test_no_buttons_returns_false(self): + blocks = [ + RichBlockParagraph( + text=RichTextUrl(text="just a link", url="https://spam.example/x") + ) + ] + assert has_rich_buttons(RichMessage(blocks=blocks)) is False + + def test_button_inside_credit_detected(self): + blocks = [ + RichBlockPullQuotation( + text="a wise quote", + credit=RichTextButton( + button=RichMessageButton(text="source", url=SPAM_URL) + ), + ) + ] + assert has_rich_buttons(RichMessage(blocks=blocks)) is True + + def test_none_returns_false(self): + assert has_rich_buttons(None) is False + + def test_walk_failure_returns_false(self): + class _Boom: + @property + def buttons(self): + raise RuntimeError("boom") + + assert has_rich_buttons(_Boom()) is False + + +class TestHasRichMessageButtons: + def test_spam_rich_message_detected(self): + assert has_rich_message_buttons(_make_message(_spam_blocks())) is True + + def test_buttons_with_whitelisted_urls_still_detected(self): + # Buttons are the trigger, not the URL: no stock client can + # compose them, so even whitelisted targets mean automation. + blocks = [ + RichBlockButtons( + buttons=[ + RichMessageButton( + text="Docs", url="https://docs.python.org/3/library/" + ) + ] + ) + ] + assert has_rich_message_buttons(_make_message(blocks)) is True + + def test_no_rich_message_ignored(self): + assert has_rich_message_buttons(_make_message()) is False + + def test_rich_without_buttons_ignored(self): + # URL-only rich content goes to the AI monitor, mirroring how + # plain text links are handled. + blocks = [ + RichBlockParagraph( + text=RichTextUrl(text="shady link", url="https://spam.example/x") + ) + ] + assert has_rich_message_buttons(_make_message(blocks)) is False + + +class TestHasNonWhitelistedLinkRich: + """Probation link check also sees URLs hidden in rich blocks.""" + + def test_rich_text_link_detected(self): + blocks = [ + RichBlockParagraph( + text=RichTextUrl(text="shady link", url="https://spam.example/x") + ) + ] + assert has_non_whitelisted_link(_make_message(blocks)) is True + + def test_rich_button_url_detected(self): + blocks = [ + RichBlockButtons( + buttons=[RichMessageButton(text="OPEN", url=SPAM_URL)] + ) + ] + assert has_non_whitelisted_link(_make_message(blocks)) is True + + def test_rich_whitelisted_url_ignored(self): + blocks = [ + RichBlockParagraph( + text=RichTextUrl( + text="docs", url="https://docs.python.org/3/library/" + ) + ) + ] + assert has_non_whitelisted_link(_make_message(blocks)) is False + + def test_rich_without_urls_ignored(self): + blocks = [RichBlockParagraph(text="just text, no links")] + assert has_non_whitelisted_link(_make_message(blocks)) is False + + +class TestClassifiableTextRich: + def test_rich_content_included(self): + text = _classifiable_text(_make_message(_spam_blocks())) + assert "HOT 2026 DROP" in text + assert "BIKINI SHOWS" in text + assert SPAM_URL in text + + def test_plain_text_unchanged(self): + assert _classifiable_text(_make_message(text="hello")) == "hello" + + def test_caption_unchanged(self): + msg = Message( + message_id=1, + date=datetime.now(UTC), + chat=Chat(id=1, type="private"), + caption="a caption", + ) + assert _classifiable_text(msg) == "a caption" + + def test_empty_message_stays_empty(self): + assert _classifiable_text(_make_message()) == "" + + def test_text_and_rich_combined(self): + text = _classifiable_text(_make_message(_spam_blocks(), text="intro")) + assert text.startswith("intro") + assert "HOT 2026 DROP" in text + + +class TestHandleInlineKeyboardSpamRich: + """Rich-message spam takes the auto-delete path at group=1.""" + + @pytest.fixture + def mock_update(self): + update = MagicMock() + update.message = MagicMock(spec=Message) + update.message.from_user = MagicMock(spec=User) + update.message.from_user.id = 12345 + update.message.from_user.is_bot = False + update.message.from_user.full_name = "Spam User" + update.message.from_user.username = "spamuser" + update.message.delete = AsyncMock() + update.message.reply_markup = None + update.message.forward_origin = None + update.edited_message = None + return update + + @pytest.fixture + def mock_context(self): + bot = MagicMock() + bot.send_message = AsyncMock() + bot.restrict_chat_member = AsyncMock() + state = AppState() + state.group_admin_ids = {} + state.trusted_user_ids = set() + return HandlerContext(bot=bot, state=state, args=[]) + + @pytest.fixture + def group_config(self): + return GroupConfig( + group_id=-100123456, + warning_topic_id=123, + rules_link="https://example.com/rules", + ) + + async def test_rich_spam_deleted_restricted_notified( + self, mock_update, mock_context, group_config + ): + mock_update.message.rich_message = RichMessage(blocks=_spam_blocks()) + + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + with pytest.raises(StopPropagation): + await handle_inline_keyboard_spam(mock_update, mock_context) + + mock_update.message.delete.assert_called_once() + mock_context.bot.restrict_chat_member.assert_called_once() + mock_context.bot.send_message.assert_called_once() + sent_text = mock_context.bot.send_message.call_args.kwargs["text"] + assert "tombol pada pesan rich text" in sent_text + + async def test_buttons_with_whitelisted_urls_still_deleted( + self, mock_update, mock_context, group_config + ): + # Buttons are the trigger: no stock client can compose them, so the + # URL whitelist does not apply to rich message buttons. + mock_update.message.rich_message = RichMessage( + blocks=[ + RichBlockButtons( + buttons=[ + RichMessageButton( + text="Docs", url="https://docs.python.org/3/library/" + ) + ] + ) + ] + ) + + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + with pytest.raises(StopPropagation): + await handle_inline_keyboard_spam(mock_update, mock_context) + + mock_update.message.delete.assert_called_once() + mock_context.bot.restrict_chat_member.assert_called_once() + + async def test_rich_without_buttons_ignored( + self, mock_update, mock_context, group_config + ): + # URL-only rich content is left to the AI monitor, like plain text. + mock_update.message.rich_message = RichMessage( + blocks=[ + RichBlockParagraph( + text=RichTextUrl(text="shady link", url="https://spam.example/x") + ) + ] + ) + + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + await handle_inline_keyboard_spam(mock_update, mock_context) + + mock_update.message.delete.assert_not_called() + mock_context.bot.restrict_chat_member.assert_not_called() + + async def test_no_rich_message_ignored(self, mock_update, mock_context, group_config): + mock_update.message.rich_message = None + + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + await handle_inline_keyboard_spam(mock_update, mock_context) + + mock_update.message.delete.assert_not_called() + + async def test_automatic_forward_skipped( + self, mock_update, mock_context, group_config + ): + # Linked-channel auto-forwards are admin-configured: never punish + # the channel for rich buttons in its posts. + mock_update.message.rich_message = RichMessage(blocks=_spam_blocks()) + mock_update.message.is_automatic_forward = True + + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + await handle_inline_keyboard_spam(mock_update, mock_context) + + mock_update.message.delete.assert_not_called() + mock_context.bot.restrict_chat_member.assert_not_called() + + +def _make_update(blocks=None, text=None): + return Update( + update_id=1, + message=Message( + message_id=1, + date=datetime.now(UTC), + chat=Chat(id=-100, type="supergroup"), + from_user=User(id=7, is_bot=False, first_name="U"), + text=text, + rich_message=RichMessage(blocks=blocks) if blocks is not None else None, + ), + ) + + +class TestRichWalkerGaps: + """Walker coverage for list labels, payload sub-objects, depth.""" + + def test_list_item_label_extracted(self): + blocks = [ + RichBlockList( + items=[ + RichBlockListItem( + label="PROMO SPESIAL HARI INI", + blocks=[RichBlockParagraph(text="detail")], + ) + ] + ) + ] + text = extract_rich_text(RichMessage(blocks=blocks)) + assert "PROMO SPESIAL HARI INI" in text + + def test_copy_text_payload_extracted(self): + blocks = [ + RichBlockButtons( + buttons=[ + RichMessageButton( + text="copy", + copy_text=CopyTextButton(text="https://spam.example/steal"), + ) + ] + ) + ] + assert extract_rich_urls(RichMessage(blocks=blocks)) == [ + "https://spam.example/steal" + ] + assert has_rich_buttons(RichMessage(blocks=blocks)) is True + + def test_login_url_and_web_app_extracted(self): + blocks = [ + RichBlockButtons( + buttons=[ + RichMessageButton( + text="login", + login_url=LoginUrl( + url="https://spam.example/login", + forward_text="open login", + ), + ), + RichMessageButton( + text="app", + web_app=WebAppInfo(url="https://spam.example/app"), + ), + ] + ) + ] + urls = extract_rich_urls(RichMessage(blocks=blocks)) + assert urls == ["https://spam.example/login", "https://spam.example/app"] + assert "open login" in extract_rich_text(RichMessage(blocks=blocks)) + + def test_credit_nested_button_url_extracted(self): + blocks = [ + RichBlockPullQuotation( + text="a wise quote", + credit=RichTextButton( + button=RichMessageButton(text="source", url=SPAM_URL) + ), + ) + ] + assert extract_rich_urls(RichMessage(blocks=blocks)) == [SPAM_URL] + + def test_deep_nesting_buttons_fail_closed(self): + node = ["deep"] + for _ in range(3000): + node = [node] + assert has_rich_buttons(node) is True + + def test_deep_nesting_extract_terminates(self): + node = ["deep"] + for _ in range(3000): + node = [node] + assert extract_rich_text(node) == "" + assert extract_rich_urls(node) == [] + + +class TestAiSpamFilterRich: + def test_rich_only_message_matches(self): + update = _make_update( + [RichBlockParagraph(text="this is a rich-only message with buttons")] + ) + assert ai_spam_monitor.AI_SPAM_FILTER(update) is True + + def test_rich_empty_no_match(self): + update = _make_update([]) + assert ai_spam_monitor.AI_SPAM_FILTER(update) is False + + def test_no_rich_no_match(self): + assert ai_spam_monitor.AI_SPAM_FILTER(_make_update()) is False + + +class TestClassifyTruncation: + async def test_long_input_truncated_before_classify(self): + captured = {} + + async def fake_classify(text, **kwargs): + captured["text"] = text + return None + + settings = MagicMock() + settings.classifier_cooldown_seconds = 600 + settings.classifier_timeout_seconds = 10 + settings.ai_spam_daily_budget = 15000 + context = HandlerContext(bot=MagicMock(), state=AppState(), args=[]) + user = User(id=1, is_bot=False, first_name="T") + with ( + patch("bot.handlers.ai_spam_monitor.classify_text", new=fake_classify), + patch( + "bot.handlers.ai_spam_monitor.get_settings", return_value=settings + ), + patch( + "bot.handlers.ai_spam_monitor.breaker_is_open", return_value=False + ), + patch( + "bot.handlers.ai_spam_monitor.try_spend_budget", return_value=True + ), + ): + await ai_spam_monitor._classify_and_alert( + context, + group_id=-100, + user=user, + message_id=1, + message_text="x" * 20000, + ) + assert len(captured["text"]) == AI_SPAM_CLASSIFY_MAX_LEN + + +class TestInlineKeyboardAutoForward: + """The is_automatic_forward guard covers the pre-existing keyboard path too.""" + + @pytest.fixture + def mock_update(self): + update = MagicMock() + update.message = MagicMock(spec=Message) + update.message.from_user = MagicMock(spec=User) + update.message.from_user.id = 12345 + update.message.from_user.is_bot = False + update.message.from_user.full_name = "Spam User" + update.message.from_user.username = "spamuser" + update.message.delete = AsyncMock() + update.message.reply_markup = InlineKeyboardMarkup( + inline_keyboard=[ + [InlineKeyboardButton(text="x", url="https://evil.example/")] + ] + ) + update.message.rich_message = None + update.message.is_automatic_forward = True + update.message.forward_origin = None + update.edited_message = None + return update + + @pytest.fixture + def mock_context(self): + bot = MagicMock() + bot.send_message = AsyncMock() + bot.restrict_chat_member = AsyncMock() + state = AppState() + state.group_admin_ids = {} + state.trusted_user_ids = set() + return HandlerContext(bot=bot, state=state, args=[]) + + @pytest.fixture + def group_config(self): + return GroupConfig( + group_id=-100123456, + warning_topic_id=123, + rules_link="https://example.com/rules", + ) + + async def test_keyboard_auto_forward_skipped( + self, mock_update, mock_context, group_config + ): + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + await handle_inline_keyboard_spam(mock_update, mock_context) + + mock_update.message.delete.assert_not_called() + mock_context.bot.restrict_chat_member.assert_not_called() + + +class TestRichNoRestrictAndExemptions: + @pytest.fixture + def mock_update(self): + update = MagicMock() + update.message = MagicMock(spec=Message) + update.message.from_user = MagicMock(spec=User) + update.message.from_user.id = 12345 + update.message.from_user.is_bot = False + update.message.from_user.full_name = "Spam User" + update.message.from_user.username = "spamuser" + update.message.delete = AsyncMock() + update.message.reply_markup = None + update.message.rich_message = RichMessage(blocks=_spam_blocks()) + update.message.is_automatic_forward = False + update.message.forward_origin = None + update.edited_message = None + return update + + @pytest.fixture + def mock_context(self): + bot = MagicMock() + bot.send_message = AsyncMock() + bot.restrict_chat_member = AsyncMock() + state = AppState() + state.group_admin_ids = {} + state.trusted_user_ids = set() + return HandlerContext(bot=bot, state=state, args=[]) + + @pytest.fixture + def group_config(self): + return GroupConfig( + group_id=-100123456, + warning_topic_id=123, + rules_link="https://example.com/rules", + ) + + async def test_no_restrict_template_on_failure( + self, mock_update, mock_context, group_config + ): + mock_context.bot.restrict_chat_member = AsyncMock( + side_effect=Exception("Restrict failed") + ) + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + with pytest.raises(StopPropagation): + await handle_inline_keyboard_spam(mock_update, mock_context) + + text = mock_context.bot.send_message.call_args.kwargs["text"] + assert "dibatasi" not in text + assert "tombol pada pesan rich text" in text + + async def test_admin_exempt_from_rich_delete( + self, mock_update, mock_context, group_config + ): + mock_context.state.group_admin_ids = { + group_config.group_id: [mock_update.message.from_user.id] + } + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + await handle_inline_keyboard_spam(mock_update, mock_context) + + mock_update.message.delete.assert_not_called() + + async def _notification_text( + self, mock_update, mock_context, group_config, forwarded: bool + ) -> str: + mock_update.message.forward_origin = MagicMock() if forwarded else None + with patch( + "bot.handlers.anti_spam.get_group_config_for_update", + return_value=group_config, + ): + with pytest.raises(StopPropagation): + await handle_inline_keyboard_spam(mock_update, mock_context) + return mock_context.bot.send_message.call_args.kwargs["text"] + + async def test_forwarded_wording_and_appeal_line( + self, mock_update, mock_context, group_config + ): + text = await self._notification_text( + mock_update, mock_context, group_config, forwarded=True + ) + assert "Pesan terusan dari" in text + assert "Hubungi admin grup" in text + + async def test_non_forwarded_wording_and_appeal_line( + self, mock_update, mock_context, group_config + ): + text = await self._notification_text( + mock_update, mock_context, group_config, forwarded=False + ) + assert "Pesan dari" in text + assert "Pesan terusan" not in text + assert "Hubungi admin grup" in text