From 1e6cfe6bfae9859c0e61cd4a7a30909ff0b1b17d Mon Sep 17 00:00:00 2001 From: fzf404 Date: Sat, 26 Sep 2026 21:02:03 +0800 Subject: [PATCH 1/3] fix: ignore Telegram commands for other bots --- .../platform/sources/telegram/tg_adapter.py | 13 +++--- tests/test_telegram_adapter.py | 46 +++++++++++++++++++ 2 files changed, 53 insertions(+), 6 deletions(-) diff --git a/astrbot/core/platform/sources/telegram/tg_adapter.py b/astrbot/core/platform/sources/telegram/tg_adapter.py index 9b2cfdd2abd..03d48556ee9 100644 --- a/astrbot/core/platform/sources/telegram/tg_adapter.py +++ b/astrbot/core/platform/sources/telegram/tg_adapter.py @@ -608,15 +608,16 @@ def _apply_caption() -> None: plain_text2 = f"/@{context.bot.username} " + plain_text plain_text = plain_text2 - # 群聊场景命令特殊处理 + # Normalize commands explicitly addressed to a Telegram bot. if plain_text.startswith("/"): command_parts = plain_text.split(" ", 1) if "@" in command_parts[0]: - command, bot_name = command_parts[0].split("@") - if bot_name == self.client.username: - plain_text = command + ( - f" {command_parts[1]}" if len(command_parts) > 1 else "" - ) + command, bot_name = command_parts[0].split("@", 1) + if bot_name.casefold() != context.bot.username.casefold(): + return None + plain_text = command + ( + f" {command_parts[1]}" if len(command_parts) > 1 else "" + ) if update.message.entities: for entity in update.message.entities: diff --git a/tests/test_telegram_adapter.py b/tests/test_telegram_adapter.py index 3001d5a217c..c30bd7b3949 100644 --- a/tests/test_telegram_adapter.py +++ b/tests/test_telegram_adapter.py @@ -84,6 +84,52 @@ def _build_context() -> MagicMock: return context +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("text", "expected"), + [ + ("/help@test_bot", "/help"), + ("/help@TEST_BOT details", "/help details"), + ("/help", "/help"), + ], +) +async def test_telegram_command_for_current_bot_is_normalized(text, expected): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + update = create_mock_update(message_text=text, chat_type="group") + + result = await adapter.convert_message(update, _build_context()) + + assert result is not None + assert result.message_str == expected + assert any( + isinstance(component, Comp.Plain) and component.text == expected + for component in result.message + ) + + +@pytest.mark.asyncio +async def test_telegram_command_for_another_bot_is_ignored(): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + update = create_mock_update( + message_text="/help@another_bot details", + chat_type="group", + ) + + result = await adapter.convert_message(update, _build_context()) + + assert result is None + + @pytest.mark.asyncio async def test_telegram_topic_with_missing_name_falls_back_to_group_name(): TelegramPlatformAdapter = _load_telegram_adapter() From d01f2c1b63b5a6380c876c415ea0cc6513d01f8e Mon Sep 17 00:00:00 2001 From: fzf404 Date: Sat, 26 Sep 2026 22:34:44 +0800 Subject: [PATCH 2/3] fix: reject foreign Telegram commands before reply parsing --- .../platform/sources/telegram/tg_adapter.py | 11 +++++++--- tests/test_telegram_adapter.py | 22 +++++++++++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/astrbot/core/platform/sources/telegram/tg_adapter.py b/astrbot/core/platform/sources/telegram/tg_adapter.py index 03d48556ee9..6e6e8a50f4b 100644 --- a/astrbot/core/platform/sources/telegram/tg_adapter.py +++ b/astrbot/core/platform/sources/telegram/tg_adapter.py @@ -562,6 +562,13 @@ def _apply_caption() -> None: message.message_str = "" message.message = [] + if update.message.text and update.message.text.startswith("/"): + command_token = update.message.text.split(" ", 1)[0] + if "@" in command_token: + _, bot_name = command_token.split("@", 1) + if bot_name.casefold() != context.bot.username.casefold(): + return None + if update.message.reply_to_message and not ( update.message.is_topic_message and update.message.message_thread_id @@ -612,9 +619,7 @@ def _apply_caption() -> None: if plain_text.startswith("/"): command_parts = plain_text.split(" ", 1) if "@" in command_parts[0]: - command, bot_name = command_parts[0].split("@", 1) - if bot_name.casefold() != context.bot.username.casefold(): - return None + command, _ = command_parts[0].split("@", 1) plain_text = command + ( f" {command_parts[1]}" if len(command_parts) > 1 else "" ) diff --git a/tests/test_telegram_adapter.py b/tests/test_telegram_adapter.py index c30bd7b3949..002a1dd8aa0 100644 --- a/tests/test_telegram_adapter.py +++ b/tests/test_telegram_adapter.py @@ -130,6 +130,28 @@ async def test_telegram_command_for_another_bot_is_ignored(): assert result is None +@pytest.mark.asyncio +async def test_telegram_command_for_another_bot_skips_reply_side_effects(): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + adapter.start = AsyncMock() + reply_to_message = create_mock_update(message_text="/start").message + update = create_mock_update( + message_text="/help@another_bot", + chat_type="group", + reply_to_message=reply_to_message, + ) + + result = await adapter.convert_message(update, _build_context()) + + assert result is None + adapter.start.assert_not_awaited() + + @pytest.mark.asyncio async def test_telegram_topic_with_missing_name_falls_back_to_group_name(): TelegramPlatformAdapter = _load_telegram_adapter() From ca10ea39202f7e2ab07ea797c913b02da11d1829 Mon Sep 17 00:00:00 2001 From: fzf404 Date: Sun, 27 Sep 2026 16:46:38 +0800 Subject: [PATCH 3/3] test: preserve Telegram command argument mentions --- .../platform/sources/telegram/tg_adapter.py | 15 ++++--- tests/test_telegram_adapter.py | 40 +++++++++++++++++++ 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/astrbot/core/platform/sources/telegram/tg_adapter.py b/astrbot/core/platform/sources/telegram/tg_adapter.py index 6e6e8a50f4b..2ead1bd20d1 100644 --- a/astrbot/core/platform/sources/telegram/tg_adapter.py +++ b/astrbot/core/platform/sources/telegram/tg_adapter.py @@ -562,7 +562,11 @@ def _apply_caption() -> None: message.message_str = "" message.message = [] - if update.message.text and update.message.text.startswith("/"): + if ( + message.type == MessageType.GROUP_MESSAGE + and update.message.text + and update.message.text.startswith("/") + ): command_token = update.message.text.split(" ", 1)[0] if "@" in command_token: _, bot_name = command_token.split("@", 1) @@ -619,10 +623,11 @@ def _apply_caption() -> None: if plain_text.startswith("/"): command_parts = plain_text.split(" ", 1) if "@" in command_parts[0]: - command, _ = command_parts[0].split("@", 1) - plain_text = command + ( - f" {command_parts[1]}" if len(command_parts) > 1 else "" - ) + command, bot_name = command_parts[0].split("@", 1) + if bot_name.casefold() == context.bot.username.casefold(): + plain_text = command + ( + f" {command_parts[1]}" if len(command_parts) > 1 else "" + ) if update.message.entities: for entity in update.message.entities: diff --git a/tests/test_telegram_adapter.py b/tests/test_telegram_adapter.py index 002a1dd8aa0..44191c78009 100644 --- a/tests/test_telegram_adapter.py +++ b/tests/test_telegram_adapter.py @@ -130,6 +130,46 @@ async def test_telegram_command_for_another_bot_is_ignored(): assert result is None +@pytest.mark.asyncio +@pytest.mark.parametrize( + "text", + [ + "/help @some_user", + "hello @some_user", + ], +) +async def test_telegram_non_target_mentions_are_preserved(text): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + update = create_mock_update(message_text=text, chat_type="group") + + result = await adapter.convert_message(update, _build_context()) + + assert result is not None + assert result.message_str == text + + +@pytest.mark.asyncio +async def test_telegram_private_command_for_another_bot_is_preserved(): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + text = "/help@another_bot" + update = create_mock_update(message_text=text, chat_type="private") + + result = await adapter.convert_message(update, _build_context()) + + assert result is not None + assert result.message_str == text + + @pytest.mark.asyncio async def test_telegram_command_for_another_bot_skips_reply_side_effects(): TelegramPlatformAdapter = _load_telegram_adapter()