diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index e91a38ac6b10..4c1b04522900 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -496,7 +496,9 @@ def _is_callback_user_authorized( allowed_csv = os.getenv("TELEGRAM_ALLOWED_USERS", "").strip() if not allowed_csv: - return True + _global_open = os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes") + _tg_open = os.getenv("TELEGRAM_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes") + return _global_open or _tg_open allowed_ids = {uid.strip() for uid in allowed_csv.split(",") if uid.strip()} return "*" in allowed_ids or normalized_user_id in allowed_ids diff --git a/gateway/run.py b/gateway/run.py index 559adae89bf0..4f28f9c5459d 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3363,6 +3363,13 @@ async def start(self) -> bool: "Set GATEWAY_ALLOW_ALL_USERS=true in ~/.hermes/.env to allow open access, " "or configure platform allowlists (e.g., TELEGRAM_ALLOWED_USERS=your_id)." ) + elif _allow_all: + logger.warning( + "SECURITY WARNING: Gateway is open to ALL users (*_ALLOW_ALL_USERS=true). " + "Any user who finds this bot can interact with your local agent " + "(filesystem, terminal, credentials). " + "Set TELEGRAM_ALLOWED_USERS / DISCORD_ALLOWED_USERS etc. to restrict access." + ) # Discover Python plugins before shell hooks so plugin block # decisions take precedence in tie cases. The CLI startup path diff --git a/tests/gateway/test_telegram_auth_fail_closed.py b/tests/gateway/test_telegram_auth_fail_closed.py new file mode 100644 index 000000000000..12836a72f1a9 --- /dev/null +++ b/tests/gateway/test_telegram_auth_fail_closed.py @@ -0,0 +1,42 @@ +"""Tests for Telegram _is_authorized_user failing closed when no allowlist is set.""" + + +def _simulate_is_authorized(user_id: str, env: dict) -> bool: + """Replicate the env-based auth logic from TelegramAdapter._is_authorized_user.""" + allowed_csv = env.get("TELEGRAM_ALLOWED_USERS", "").strip() + if not allowed_csv: + _global_open = env.get("GATEWAY_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes") + _tg_open = env.get("TELEGRAM_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes") + return _global_open or _tg_open + allowed_ids = {uid.strip() for uid in allowed_csv.split(",") if uid.strip()} + return "*" in allowed_ids or user_id in allowed_ids + + +class TestTelegramAuthFailClosed: + def test_no_allowlist_no_allow_all_denies(self): + result = _simulate_is_authorized("999", {}) + assert result is False + + def test_no_allowlist_gateway_allow_all_true_permits(self): + result = _simulate_is_authorized("999", {"GATEWAY_ALLOW_ALL_USERS": "true"}) + assert result is True + + def test_no_allowlist_telegram_allow_all_true_permits(self): + result = _simulate_is_authorized("999", {"TELEGRAM_ALLOW_ALL_USERS": "1"}) + assert result is True + + def test_no_allowlist_allow_all_false_string_denies(self): + result = _simulate_is_authorized("999", {"GATEWAY_ALLOW_ALL_USERS": "false"}) + assert result is False + + def test_known_user_in_allowlist_permits(self): + result = _simulate_is_authorized("123456", {"TELEGRAM_ALLOWED_USERS": "123456,789012"}) + assert result is True + + def test_unknown_user_in_allowlist_denies(self): + result = _simulate_is_authorized("000000", {"TELEGRAM_ALLOWED_USERS": "123456,789012"}) + assert result is False + + def test_wildcard_in_allowlist_permits_anyone(self): + result = _simulate_is_authorized("999999", {"TELEGRAM_ALLOWED_USERS": "*"}) + assert result is True