fix(telegram): fail-closed auth fallback when TELEGRAM_ALLOWED_USERS is empty

gateway/platforms/telegram.py

@@ -506,7 +506,11 @@ def _is_callback_user_authorized(
 
         allowed_csv = os.getenv("TELEGRAM_ALLOWED_USERS", "").strip()
         if not allowed_csv:
-            return True
+            # Fail-closed: no allowlist means deny by default.
+            # The runner auth path in _is_user_authorized() handles
+            # GATEWAY_ALLOW_ALL_USERS; this fallback must not silently
+            # allow everyone (fixes #24457).
+            return os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in {"true", "1", "yes"}
         allowed_ids = {uid.strip() for uid in allowed_csv.split(",") if uid.strip()}
         return "*" in allowed_ids or normalized_user_id in allowed_ids
 

tests/gateway/test_telegram_callback_auth_fail_closed.py

@@ -0,0 +1,108 @@
+"""Tests for Telegram adapter fail-closed auth fallback (#24457).
+
+The _is_callback_user_authorized fallback must deny users by default
+when TELEGRAM_ALLOWED_USERS is empty, instead of allowing everyone.
+"""
+
+import sys
+import types
+from types import SimpleNamespace
+
+import pytest
+
+from gateway.config import PlatformConfig, Platform
+
+
+# -- Fake telegram modules (minimal stubs) --------------------------------
+
+_fake_telegram_error = types.ModuleType("telegram.error")
+
+
+class _TelegramError(Exception):
+    pass
+
+
+_fake_telegram_error.TelegramError = _TelegramError
+_fake_telegram_error.BadRequest = type("BadRequest", (_TelegramError,), {})
+_fake_telegram_error.NetworkError = type("NetworkError", (_TelegramError,), {})
+
+_fake_telegram_constants = types.ModuleType("telegram.constants")
+_fake_telegram_constants.ParseMode = SimpleNamespace(HTML="HTML")
+
+_fake_telegram_request = types.ModuleType("telegram.request")
+_fake_telegram_request.HTTPXRequest = type("HTTPXRequest", (), {"__init__": lambda *a, **kw: None})
+
+_fake_telegram_ext = types.ModuleType("telegram.ext")
+_fake_telegram_ext.ApplicationBuilder = type("ApplicationBuilder", (), {
+    "token": lambda self, *a: self,
+    "build": lambda self: None,
+})
+
+_fake_telegram = types.ModuleType("telegram")
+_fake_telegram.error = _fake_telegram_error
+_fake_telegram.constants = _fake_telegram_constants
+_fake_telegram.ext = _fake_telegram_ext
+_fake_telegram.request = _fake_telegram_request
+
+
+@pytest.fixture(autouse=True)
+def _inject_fake_telegram(monkeypatch):
+    monkeypatch.setitem(sys.modules, "telegram", _fake_telegram)
+    monkeypatch.setitem(sys.modules, "telegram.error", _fake_telegram_error)
+    monkeypatch.setitem(sys.modules, "telegram.constants", _fake_telegram_constants)
+    monkeypatch.setitem(sys.modules, "telegram.ext", _fake_telegram_ext)
+    monkeypatch.setitem(sys.modules, "telegram.request", _fake_telegram_request)
+
+
+def _make_adapter():
+    from gateway.platforms.telegram import TelegramAdapter
+
+    config = PlatformConfig(enabled=True, token="fake-token")
+    adapter = object.__new__(TelegramAdapter)
+    adapter.config = config
+    adapter._config = config
+    adapter._platform = Platform.TELEGRAM
+    adapter._connected = True
+    return adapter
+
+
+class TestCallbackAuthFailClosed:
+    """_is_callback_user_authorized fallback must be fail-closed."""
+
+    def test_no_allowlist_no_allow_all_denies(self, monkeypatch):
+        """No TELEGRAM_ALLOWED_USERS and no GATEWAY_ALLOW_ALL_USERS → deny."""
+        monkeypatch.delenv("TELEGRAM_ALLOWED_USERS", raising=False)
+        monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False)
+        adapter = _make_adapter()
+        # Force the fallback path (no runner auth)
+        adapter._message_handler = None
+        assert adapter._is_callback_user_authorized("12345") is False
+
+    def test_no_allowlist_with_global_allow_all_permits(self, monkeypatch):
+        """No TELEGRAM_ALLOWED_USERS but GATEWAY_ALLOW_ALL_USERS=true → allow."""
+        monkeypatch.delenv("TELEGRAM_ALLOWED_USERS", raising=False)
+        monkeypatch.setenv("GATEWAY_ALLOW_ALL_USERS", "true")
+        adapter = _make_adapter()
+        adapter._message_handler = None
+        assert adapter._is_callback_user_authorized("12345") is True
+
+    def test_allowlist_with_matching_user_permits(self, monkeypatch):
+        """TELEGRAM_ALLOWED_USERS contains the user → allow."""
+        monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "12345,67890")
+        adapter = _make_adapter()
+        adapter._message_handler = None
+        assert adapter._is_callback_user_authorized("12345") is True
+
+    def test_allowlist_without_matching_user_denies(self, monkeypatch):
+        """TELEGRAM_ALLOWED_USERS does not contain the user → deny."""
+        monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "67890")
+        adapter = _make_adapter()
+        adapter._message_handler = None
+        assert adapter._is_callback_user_authorized("12345") is False
+
+    def test_allowlist_wildcard_permits(self, monkeypatch):
+        """TELEGRAM_ALLOWED_USERS=* → allow everyone."""
+        monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "*")
+        adapter = _make_adapter()
+        adapter._message_handler = None
+        assert adapter._is_callback_user_authorized("12345") is True

tests/gateway/test_telegram_group_gating.py

@@ -44,6 +44,11 @@ def _make_adapter(
     adapter._pending_text_batch_tasks = {}
     adapter._text_batch_delay_seconds = 0.01
     adapter._mention_patterns = adapter._compile_mention_patterns()
+    # Trigger-gating tests don't exercise the allowlist gate (added by
+    # #23795 + #24468).  Force-authorize all senders so the trigger logic
+    # under test runs.  Without this, every fake message hits the new
+    # fail-closed auth path and gets dropped before trigger evaluation.
+    adapter._is_callback_user_authorized = lambda user_id, **_kw: True
     return adapter