feat: block non-whitelisted guest bot messages with progressive restriction

Telegram Bot API 10.0 (May 2026) introduced Guest Mode, allowing any
user to @mention a bot in any group chat without the bot being a member.
The bot posts a reply directly in the chat. This feature adds moderation
for those guest bot messages.

New handler (src/bot/handlers/guest_bot.py):
- GuestBotFilter: custom MessageFilter matching only guest bot messages
  (guest_bot_caller_user or guest_bot_caller_chat set)
- is_guest_bot_whitelisted: case-insensitive username matching against
  per-group whitelist (strips @, lowercases)
- handle_guest_bot_message: deletes non-whitelisted guest bot messages
  and progressively restricts the invoking user (not the bot) using the
  existing UserWarning state machine with a new warning_kind=guest_bot
  discriminator — 1st violation: warning, 2nd to (N-1): silent increment,
  Nth: restrict + notification. Admins/trusted users are exempt. Channel
  callers are delete-only (no human to restrict). Already-restricted
  callers skip new warning cycles.

Plugin wiring:
- New plugin guest_bot_block registered at group=1, before
  inline_keyboard_spam, with GuestBotFilter for precise matching
- Gated by guard_plugin(guest_bot_block) for per-group toggle control
- Blocking handler (no block=False) so ApplicationHandlerStop works

Config:
- guest_bot_whitelist field in Settings (NoDecode annotation for
  comma-separated env parsing) and GroupConfig (normalization validator)
- GUEST_BOT_WHITELIST env var and groups.json support
- Plugin toggle via plugins map in groups.json

DB schema:
- Added warning_kind column to UserWarning (default profile)
- SQLite migration for existing databases
- All DB service methods accept and filter by warning_kind
- Scheduler only processes profile warnings (not guest bot)
- DM/verify unrestriction only lifts profile restrictions

Tests: 1048 passed, ruff clean, mypy clean

Author: rezhajulio

.env.example

@@ -86,6 +86,13 @@ BIO_BAIT_MONITOR_ONLY=false
 # Example: PLUGINS_DEFAULT={"captcha":true,"dm":false}
 # PLUGINS_DEFAULT={"captcha":true,"dm":false}
 
+# Guest Bot Whitelist (Telegram Guest Mode - Bot API 10.0)
+# Comma-separated list of bot usernames allowed to post guest messages
+# Messages from non-whitelisted guest bots are deleted and the invoking user is warned
+# Usernames are case-insensitive, @ prefix is optional
+# Example: GUEST_BOT_WHITELIST=@somebot,anotherbot
+GUEST_BOT_WHITELIST=
+
 # Logfire Configuration (optional - for production logging)
 # Get your token from https://logfire.pydantic.dev
 LOGFIRE_TOKEN=your_logfire_token_here

groups.json.example

@@ -20,6 +20,7 @@
     "bio_bait_monitor_only": false,
     "bio_bait_alert_chat_id": null,
     "moderation_topic_id": null,
+    "guest_bot_whitelist": [],
     "plugins": {
       "captcha": false,
       "dm": true,
@@ -47,6 +48,7 @@
     "bio_bait_monitor_only": false,
     "bio_bait_alert_chat_id": null,
     "moderation_topic_id": null,
+    "guest_bot_whitelist": ["somebot"],
     "plugins": {
       "contact_spam": false,
       "duplicate_spam": false,

src/bot/config.py

@@ -11,9 +11,10 @@
 import os
 from functools import lru_cache
 from pathlib import Path
+from typing import Annotated
 
 from pydantic import field_validator
-from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 logger = logging.getLogger(__name__)
 
@@ -84,6 +85,7 @@ class Settings(BaseSettings):
     bio_bait_monitor_only: bool = False
     bio_bait_alert_chat_id: int | None = None
     moderation_topic_id: int | None = None
+    guest_bot_whitelist: Annotated[list[str], NoDecode] = []
     groups_config_path: str = "groups.json"
     logfire_token: str | None = None
     logfire_service_name: str = "pythonid-bot"
@@ -120,6 +122,18 @@ def parse_and_validate_plugins_default(cls, v: object) -> dict[str, bool]:
         from bot.plugins.config import validate_plugin_map
         return validate_plugin_map(parsed)
 
+    @field_validator("guest_bot_whitelist", mode="before")
+    @classmethod
+    def parse_guest_bot_whitelist(cls, v: object) -> list[str]:
+        """Parse GUEST_BOT_WHITELIST env var as comma-separated usernames."""
+        if isinstance(v, list):
+            return [str(entry).strip().removeprefix("@").lower() for entry in v if str(entry).strip()]
+        if isinstance(v, str):
+            if not v.strip():
+                return []
+            return [entry.strip().removeprefix("@").lower() for entry in v.split(",") if entry.strip()]
+        return []
+
     def model_post_init(self, __context):
         """Validate and log non-sensitive configuration values after initialization."""
         if self.group_id >= 0:

src/bot/constants.py

@@ -344,6 +344,19 @@ def format_hours_display(hours: int) -> str:
     "📌 [Peraturan Grup]({rules_link})"
 )
 
+GUEST_BOT_WARNING = (
+    "⚠️ {user_mention}, bot tamu tidak diizinkan di grup ini. "
+    "Pelanggaran berikutnya dapat menyebabkan pembatasan setelah "
+    "{warning_threshold} pesan.\n\n"
+    "Silakan baca [peraturan grup]({rules_link})."
+)
+
+GUEST_BOT_RESTRICTION = (
+    "🔇 {user_mention} dibatasi setelah memanggil bot tamu sebanyak "
+    "{message_count} kali.\n\n"
+    "Silakan baca [peraturan grup]({rules_link})."
+)
+
 # Duplicate message spam notification
 DUPLICATE_SPAM_RESTRICTION = (
     "🚫 *Spam Pesan Duplikat*\n\n"

src/bot/database/models.py

@@ -32,6 +32,9 @@ class UserWarning(SQLModel, table=True):
         restricted_by_bot: True if restriction was applied by this bot
             (vs manually by an admin). Only bot-created restrictions
             can be lifted via DM.
+        warning_kind: Discriminator for the warning source
+            (``"profile"`` or ``"guest_bot"``). Prevents cross-source
+            state interference.
     """
 
     __tablename__ = "user_warnings"
@@ -44,6 +47,7 @@ class UserWarning(SQLModel, table=True):
     last_message_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     is_restricted: bool = Field(default=False)
     restricted_by_bot: bool = Field(default=False)
+    warning_kind: str = Field(default="profile", index=True)
 
 
 class PhotoVerificationWhitelist(SQLModel, table=True):

src/bot/database/service.py

@@ -58,6 +58,7 @@ def __init__(self, database_path: str):
 
         # Migrate existing tables: add new columns if missing
         self._migrate_trusted_users()
+        self._migrate_user_warnings()
 
     def _migrate_trusted_users(self) -> None:
         """Add new columns to trusted_users if missing."""
@@ -80,7 +81,24 @@ def _migrate_trusted_users(self) -> None:
                     logger.info(f"Migrated trusted_users: added {col} column")
             conn.commit()
 
-    def get_or_create_user_warning(self, user_id: int, group_id: int) -> UserWarning:
+    def _migrate_user_warnings(self) -> None:
+        """Add warning_kind column to user_warnings if missing."""
+        with self._engine.connect() as conn:
+            columns = {
+                row[1] for row in conn.exec_driver_sql(
+                    "PRAGMA table_info(user_warnings)"
+                ).fetchall()
+            }
+            if "warning_kind" not in columns:
+                conn.exec_driver_sql(
+                    "ALTER TABLE user_warnings ADD COLUMN warning_kind TEXT DEFAULT 'profile'"
+                )
+                logger.info("Migrated user_warnings: added warning_kind column")
+            conn.commit()
+
+    def get_or_create_user_warning(
+        self, user_id: int, group_id: int, warning_kind: str = "profile"
+    ) -> UserWarning:
         """
         Get existing warning record or create a new one.
 
@@ -90,6 +108,7 @@ def get_or_create_user_warning(self, user_id: int, group_id: int) -> UserWarning
         Args:
             user_id: Telegram user ID.
             group_id: Telegram group ID.
+            warning_kind: Discriminator for the warning source.
 
         Returns:
             UserWarning: Active warning record for the user.
@@ -99,13 +118,14 @@ def get_or_create_user_warning(self, user_id: int, group_id: int) -> UserWarning
             statement = select(UserWarning).where(
                 UserWarning.user_id == user_id,
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
                 ~UserWarning.is_restricted,
             )
             record = session.exec(statement).first()
 
             if record:
                 logger.info(
-                    f"Returning existing warning for user_id={user_id}, group_id={group_id}"
+                    f"Returning existing warning for user_id={user_id}, group_id={group_id}, kind={warning_kind}"
                 )
                 return record
 
@@ -116,16 +136,19 @@ def get_or_create_user_warning(self, user_id: int, group_id: int) -> UserWarning
                 message_count=1,
                 first_warned_at=datetime.now(UTC),
                 last_message_at=datetime.now(UTC),
+                warning_kind=warning_kind,
             )
             session.add(new_record)
             session.commit()
             session.refresh(new_record)
             logger.info(
-                f"Created new warning for user_id={user_id}, group_id={group_id}"
+                f"Created new warning for user_id={user_id}, group_id={group_id}, kind={warning_kind}"
             )
             return new_record
 
-    def increment_message_count(self, user_id: int, group_id: int) -> UserWarning:
+    def increment_message_count(
+        self, user_id: int, group_id: int, warning_kind: str = "profile"
+    ) -> UserWarning:
         """
         Increment message count for an existing warning record.
 
@@ -135,6 +158,7 @@ def increment_message_count(self, user_id: int, group_id: int) -> UserWarning:
         Args:
             user_id: Telegram user ID.
             group_id: Telegram group ID.
+            warning_kind: Discriminator for the warning source.
 
         Returns:
             UserWarning: Updated warning record.
@@ -146,6 +170,7 @@ def increment_message_count(self, user_id: int, group_id: int) -> UserWarning:
             statement = select(UserWarning).where(
                 UserWarning.user_id == user_id,
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
                 ~UserWarning.is_restricted,
             )
             record = session.exec(statement).first()
@@ -157,15 +182,17 @@ def increment_message_count(self, user_id: int, group_id: int) -> UserWarning:
                 session.commit()
                 session.refresh(record)
                 logger.info(
-                    f"Incremented message count for user_id={user_id}, group_id={group_id}, new_count={record.message_count}"
+                    f"Incremented message count for user_id={user_id}, group_id={group_id}, kind={warning_kind}, new_count={record.message_count}"
                 )
                 return record
 
             raise ValueError(
-                f"No warning record found for user {user_id} in group {group_id}"
+                f"No warning record found for user {user_id} in group {group_id} (kind={warning_kind})"
             )
 
-    def mark_user_restricted(self, user_id: int, group_id: int) -> UserWarning:
+    def mark_user_restricted(
+        self, user_id: int, group_id: int, warning_kind: str = "profile"
+    ) -> UserWarning:
         """
         Mark user as restricted after reaching threshold.
 
@@ -175,6 +202,7 @@ def mark_user_restricted(self, user_id: int, group_id: int) -> UserWarning:
         Args:
             user_id: Telegram user ID.
             group_id: Telegram group ID.
+            warning_kind: Discriminator for the warning source.
 
         Returns:
             UserWarning: Updated warning record.
@@ -186,6 +214,7 @@ def mark_user_restricted(self, user_id: int, group_id: int) -> UserWarning:
             statement = select(UserWarning).where(
                 UserWarning.user_id == user_id,
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
                 ~UserWarning.is_restricted,
             )
             record = session.exec(statement).first()
@@ -198,15 +227,17 @@ def mark_user_restricted(self, user_id: int, group_id: int) -> UserWarning:
                 session.commit()
                 session.refresh(record)
                 logger.info(
-                    f"Marked user as restricted: user_id={user_id}, group_id={group_id}"
+                    f"Marked user as restricted: user_id={user_id}, group_id={group_id}, kind={warning_kind}"
                 )
                 return record
 
             raise ValueError(
-                f"No warning record found for user {user_id} in group {group_id}"
+                f"No warning record found for user {user_id} in group {group_id} (kind={warning_kind})"
             )
 
-    def is_user_restricted_by_bot(self, user_id: int, group_id: int) -> bool:
+    def is_user_restricted_by_bot(
+        self, user_id: int, group_id: int, warning_kind: str = "profile"
+    ) -> bool:
         """
         Check if user was restricted by this bot.
 
@@ -217,6 +248,7 @@ def is_user_restricted_by_bot(self, user_id: int, group_id: int) -> bool:
         Args:
             user_id: Telegram user ID.
             group_id: Telegram group ID.
+            warning_kind: Discriminator for the warning source.
 
         Returns:
             bool: True if user was restricted by this bot.
@@ -225,13 +257,16 @@ def is_user_restricted_by_bot(self, user_id: int, group_id: int) -> bool:
             statement = select(UserWarning).where(
                 UserWarning.user_id == user_id,
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
                 UserWarning.is_restricted,
                 UserWarning.restricted_by_bot,
             )
             record = session.exec(statement).first()
             return record is not None
 
-    def mark_user_unrestricted(self, user_id: int, group_id: int) -> None:
+    def mark_user_unrestricted(
+        self, user_id: int, group_id: int, warning_kind: str = "profile"
+    ) -> None:
         """
         Clear bot restriction flag after user is unrestricted via DM.
 
@@ -241,11 +276,13 @@ def mark_user_unrestricted(self, user_id: int, group_id: int) -> None:
         Args:
             user_id: Telegram user ID.
             group_id: Telegram group ID.
+            warning_kind: Discriminator for the warning source.
         """
         with Session(self._engine) as session:
             statement = select(UserWarning).where(
                 UserWarning.user_id == user_id,
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
                 UserWarning.is_restricted,
                 UserWarning.restricted_by_bot,
             )
@@ -256,10 +293,12 @@ def mark_user_unrestricted(self, user_id: int, group_id: int) -> None:
                 session.add(record)
                 session.commit()
                 logger.info(
-                    f"Cleared restriction flag: user_id={user_id}, group_id={group_id}"
+                    f"Cleared restriction flag: user_id={user_id}, group_id={group_id}, kind={warning_kind}"
                 )
 
-    def delete_user_warnings(self, user_id: int, group_id: int) -> int:
+    def delete_user_warnings(
+        self, user_id: int, group_id: int, warning_kind: str = "profile"
+    ) -> int:
         """
         Delete all warning records for a user in a specific group.
 
@@ -269,6 +308,7 @@ def delete_user_warnings(self, user_id: int, group_id: int) -> int:
         Args:
             user_id: Telegram user ID.
             group_id: Telegram group ID.
+            warning_kind: Discriminator for the warning source.
 
         Returns:
             int: Number of warning records deleted.
@@ -277,16 +317,19 @@ def delete_user_warnings(self, user_id: int, group_id: int) -> int:
             delete_statement = delete(UserWarning).where(
                 UserWarning.user_id == user_id,
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
             )
             result = session.exec(delete_statement)
             session.commit()
             count = result.rowcount
             logger.info(
-                f"Deleted warnings: user_id={user_id}, group_id={group_id}, count={count}"
+                f"Deleted warnings: user_id={user_id}, group_id={group_id}, kind={warning_kind}, count={count}"
             )
             return count
 
-    def get_active_user_warning(self, user_id: int, group_id: int) -> UserWarning | None:
+    def get_active_user_warning(
+        self, user_id: int, group_id: int, warning_kind: str = "profile"
+    ) -> UserWarning | None:
         """
         Get an existing active (non-restricted) warning record without creating one.
 
@@ -297,6 +340,7 @@ def get_active_user_warning(self, user_id: int, group_id: int) -> UserWarning |
         Args:
             user_id: Telegram user ID.
             group_id: Telegram group ID.
+            warning_kind: Discriminator for the warning source.
 
         Returns:
             UserWarning | None: Active warning record, or None if none exists.
@@ -305,6 +349,7 @@ def get_active_user_warning(self, user_id: int, group_id: int) -> UserWarning |
             statement = select(UserWarning).where(
                 UserWarning.user_id == user_id,
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
                 ~UserWarning.is_restricted,
             )
             return session.exec(statement).first()
@@ -540,14 +585,15 @@ def get_trusted_users(self) -> list[TrustedUser]:
             return list(session.exec(statement).all())
 
     def get_warnings_past_time_threshold_for_group(
-        self, group_id: int, threshold: timedelta
+        self, group_id: int, threshold: timedelta, warning_kind: str = "profile"
     ) -> list[UserWarning]:
         """
         Find active warnings for a specific group that exceeded the time threshold.
 
         Args:
             group_id: Telegram group ID to filter by.
             threshold: Time duration since first warning to trigger restriction.
+            warning_kind: Discriminator for the warning source.
 
         Returns:
             list[UserWarning]: Warning records that should be auto-restricted.
@@ -556,6 +602,7 @@ def get_warnings_past_time_threshold_for_group(
             cutoff_time = datetime.now(UTC) - threshold
             statement = select(UserWarning).where(
                 UserWarning.group_id == group_id,
+                UserWarning.warning_kind == warning_kind,
                 ~UserWarning.is_restricted,
                 UserWarning.first_warned_at <= cutoff_time,
             )