Fix: detect rich-message spam bypassing text-based filters

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"

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,

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,13 +163,18 @@ 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.
 
     Returns:
         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,

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

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)