Telegram adapter missing is_bot flag for auth filtering

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #54052 fix(telegram): populate source.is_bot so bot messages hit the auth filter (#32188)

  merged to main

  files: gateway/authz_mixin.py, plugins/platforms/telegram/adapter.py, tests/gateway/test_config.py, tests/gateway/test_discord_bot_auth_bypass.py, tests/gateway/test_feishu_bot_auth_bypass.py, tests/gateway/test_telegram_bot_auth_bypass.py

  Show the diff

  • gateway/authz_mixin.py

    @@ -329,6 +329,7 @@ def _is_user_authorized(self, source: SessionSource) -> bool:
             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

  • plugins/platforms/telegram/adapter.py

    @@ -7253,6 +7253,7 @@ def _build_message_event(
                 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"):

  • tests/gateway/test_config.py

    @@ -840,6 +840,38 @@ def test_feishu_allow_bots_env_takes_precedence_over_config_yaml(self, tmp_path,
     
             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()

  • 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",
         ):

  • 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",
         ):

  • 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
    +
    +
    [redacted:email](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 plugins.platforms.telegram.adapter 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