diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index a32200ad547..64fb05d3b09 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -329,6 +329,7 @@ class GatewayAuthorizationMixin: platform_allow_bots_map = { Platform.DISCORD: "DISCORD_ALLOW_BOTS", Platform.FEISHU: "FEISHU_ALLOW_BOTS", + Platform.TELEGRAM: "TELEGRAM_ALLOW_BOTS", } # Plugin platforms: check the registry for auth env var names diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index b379909026b..09a5777a4ce 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -7253,6 +7253,7 @@ class TelegramAdapter(BasePlatformAdapter): thread_id=thread_id_str, chat_topic=chat_topic, message_id=str(message.message_id), + is_bot=bool(getattr(user, "is_bot", False)) if user else False, ) # Extract reply context if this message is a reply. @@ -7528,6 +7529,8 @@ def _apply_yaml_config(yaml_cfg: dict, telegram_cfg: dict) -> dict | None: os.environ["TELEGRAM_MENTION_PATTERNS"] = _json.dumps(telegram_cfg["mention_patterns"]) if "exclusive_bot_mentions" in telegram_cfg and not os.getenv("TELEGRAM_EXCLUSIVE_BOT_MENTIONS"): os.environ["TELEGRAM_EXCLUSIVE_BOT_MENTIONS"] = str(telegram_cfg["exclusive_bot_mentions"]).lower() + if "allow_bots" in telegram_cfg and not os.getenv("TELEGRAM_ALLOW_BOTS"): + os.environ["TELEGRAM_ALLOW_BOTS"] = str(telegram_cfg["allow_bots"]).lower() if "guest_mode" in telegram_cfg and not os.getenv("TELEGRAM_GUEST_MODE"): os.environ["TELEGRAM_GUEST_MODE"] = str(telegram_cfg["guest_mode"]).lower() if "observe_unmentioned_group_messages" in telegram_cfg and not os.getenv("TELEGRAM_OBSERVE_UNMENTIONED_GROUP_MESSAGES"): diff --git a/tests/gateway/test_config.py b/tests/gateway/test_config.py index a4a4ffade06..7e29d75cc6c 100644 --- a/tests/gateway/test_config.py +++ b/tests/gateway/test_config.py @@ -840,6 +840,38 @@ class TestLoadGatewayConfig: assert os.environ.get("FEISHU_ALLOW_BOTS") == "none" + def test_bridges_telegram_allow_bots_from_config_yaml_to_env(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + config_path = hermes_home / "config.yaml" + config_path.write_text( + "telegram:\n allow_bots: mentions\n", + encoding="utf-8", + ) + + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.delenv("TELEGRAM_ALLOW_BOTS", raising=False) + + load_gateway_config() + + assert os.environ.get("TELEGRAM_ALLOW_BOTS") == "mentions" + + def test_telegram_allow_bots_env_takes_precedence_over_config_yaml(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + config_path = hermes_home / "config.yaml" + config_path.write_text( + "telegram:\n allow_bots: all\n", + encoding="utf-8", + ) + + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "none") + + load_gateway_config() + + assert os.environ.get("TELEGRAM_ALLOW_BOTS") == "none" + def test_invalid_quick_commands_in_config_yaml_are_ignored(self, tmp_path, monkeypatch): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() diff --git a/tests/gateway/test_discord_bot_auth_bypass.py b/tests/gateway/test_discord_bot_auth_bypass.py index 8e10dfbcb94..71be4edfb6c 100644 --- a/tests/gateway/test_discord_bot_auth_bypass.py +++ b/tests/gateway/test_discord_bot_auth_bypass.py @@ -31,6 +31,7 @@ def _isolate_discord_env(monkeypatch): "DISCORD_ALLOWED_USERS", "DISCORD_ALLOWED_ROLES", "DISCORD_ALLOW_ALL_USERS", + "TELEGRAM_ALLOW_BOTS", "GATEWAY_ALLOW_ALL_USERS", "GATEWAY_ALLOWED_USERS", ): diff --git a/tests/gateway/test_feishu_bot_auth_bypass.py b/tests/gateway/test_feishu_bot_auth_bypass.py index 4dd83a1bd37..3cd3a854a53 100644 --- a/tests/gateway/test_feishu_bot_auth_bypass.py +++ b/tests/gateway/test_feishu_bot_auth_bypass.py @@ -21,6 +21,7 @@ def _isolate_feishu_env(monkeypatch): "FEISHU_ALLOW_BOTS", "FEISHU_ALLOWED_USERS", "FEISHU_ALLOW_ALL_USERS", + "TELEGRAM_ALLOW_BOTS", "GATEWAY_ALLOW_ALL_USERS", "GATEWAY_ALLOWED_USERS", ): diff --git a/tests/gateway/test_telegram_bot_auth_bypass.py b/tests/gateway/test_telegram_bot_auth_bypass.py new file mode 100644 index 00000000000..d9a8d1e4f0f --- /dev/null +++ b/tests/gateway/test_telegram_bot_auth_bypass.py @@ -0,0 +1,158 @@ +"""Regression guard for Telegram bot-origin authorization (#32188).""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from gateway.session import Platform, SessionSource + + +@pytest.fixture(autouse=True) +def _isolate_telegram_env(monkeypatch): + for var in ( + "TELEGRAM_ALLOW_BOTS", + "TELEGRAM_ALLOWED_USERS", + "TELEGRAM_ALLOW_ALL_USERS", + "TELEGRAM_GROUP_ALLOWED_USERS", + "TELEGRAM_GROUP_ALLOWED_CHATS", + "GATEWAY_ALLOW_ALL_USERS", + "GATEWAY_ALLOWED_USERS", + ): + monkeypatch.delenv(var, raising=False) + + +def _make_bare_runner(): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner.pairing_store = SimpleNamespace(is_approved=lambda *_a, **_kw: False) + return runner + + +def _make_telegram_bot_source(bot_id: str = "999888777"): + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="123", + chat_type="dm", + user_id=bot_id, + user_name="OtherProfileBot", + is_bot=True, + ) + + +def _make_telegram_human_source(user_id: str = "100200300"): + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="123", + chat_type="dm", + user_id=user_id, + user_name="SomeHuman", + is_bot=False, + ) + + +def test_telegram_bot_authorized_when_allow_bots_mentions(monkeypatch): + runner = _make_bare_runner() + monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "mentions") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "100200300") + + assert runner._is_user_authorized(_make_telegram_bot_source("999888777")) is True + + +def test_telegram_bot_authorized_when_allow_bots_all(monkeypatch): + runner = _make_bare_runner() + monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "all") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "100200300") + + assert runner._is_user_authorized(_make_telegram_bot_source()) is True + + +def test_telegram_bot_not_authorized_when_allow_bots_unset(monkeypatch): + runner = _make_bare_runner() + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "100200300") + + assert runner._is_user_authorized(_make_telegram_bot_source("999888777")) is False + + +def test_telegram_bot_not_authorized_when_allow_bots_none(monkeypatch): + runner = _make_bare_runner() + monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "none") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "100200300") + + assert runner._is_user_authorized(_make_telegram_bot_source("999888777")) is False + + +def test_telegram_human_still_checked_against_allowlist_when_bot_policy_set(monkeypatch): + runner = _make_bare_runner() + monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "all") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "100200300") + + assert runner._is_user_authorized(_make_telegram_human_source("999999999")) is False + assert runner._is_user_authorized(_make_telegram_human_source("100200300")) is True + + +def _build_telegram_message(*, is_bot: bool): + user = SimpleNamespace( + id=999888777 if is_bot else 100200300, + full_name="OtherProfileBot" if is_bot else "Alice", + is_bot=is_bot, + ) + chat = SimpleNamespace( + id=123, + type="private", + title=None, + full_name="Alice", + is_forum=False, + ) + message = MagicMock() + message.from_user = user + message.chat = chat + message.message_id = 4242 + message.message_thread_id = None + message.is_topic_message = False + message.forum_topic_created = None + message.reply_to_message = None + message.quote = None + message.text = "hello" + message.caption = None + return message + + +def _capture_build_source_is_bot(is_bot: bool): + from gateway.platforms.base import MessageType + from gateway.platforms.telegram import TelegramAdapter + + adapter = object.__new__(TelegramAdapter) + adapter.platform = Platform.TELEGRAM + adapter.config = SimpleNamespace(extra={}) + message = _build_telegram_message(is_bot=is_bot) + captured: dict = {} + + def fake_build_source(**kwargs): + captured.update(kwargs) + return SessionSource( + platform=Platform.TELEGRAM, + chat_id=str(kwargs.get("chat_id") or ""), + chat_type=kwargs.get("chat_type") or "dm", + user_id=kwargs.get("user_id"), + is_bot=kwargs.get("is_bot", False), + ) + + with patch.object(adapter, "build_source", side_effect=fake_build_source): + try: + adapter._build_message_event(message, MessageType.TEXT, update_id=1) + except Exception: + # The method may continue into PTB-specific optional fields after + # source construction; this test only pins the source kwarg. + pass + + return captured.get("is_bot") + + +def test_telegram_adapter_propagates_is_bot_true(): + assert _capture_build_source_is_bot(True) is True + + +def test_telegram_adapter_propagates_is_bot_false(): + assert _capture_build_source_is_bot(False) is False