fix(gateway): fail-close Telegram auth when TELEGRAM_ALLOWED_USERS is empty; warn on ALLOW_ALL - #24602
Conversation
… empty; warn on ALLOW_ALL When TELEGRAM_ALLOWED_USERS is unset, _is_authorized_user previously returned True unconditionally, allowing any Telegram user to access the local agent. Now returns False unless GATEWAY_ALLOW_ALL_USERS or TELEGRAM_ALLOW_ALL_USERS is explicitly set to true/1/yes. Also adds a startup security warning when the gateway is in open-to-all mode. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
This PR hardens gateway authorization defaults for Telegram by preventing “fail-open” behavior when no allowlist is configured, and improves operator visibility when the gateway is explicitly configured for open access.
Changes:
- Update Telegram callback authorization fallback to deny by default when
TELEGRAM_ALLOWED_USERSis empty, only permitting access when an explicit allow-all flag is set. - Add a loud startup
SECURITY WARNINGlog when any*_ALLOW_ALL_USERSmode is enabled. - Add a new test file covering allowlist/allow-all combinations for the intended fail-closed behavior.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
gateway/platforms/telegram.py |
Changes Telegram callback auth fallback to fail closed unless allow-all is explicitly enabled. |
gateway/run.py |
Adds a startup warning when open-access allow-all mode is active. |
tests/gateway/test_telegram_auth_fail_closed.py |
Introduces tests for the allowlist/allow-all decision logic (currently via a simulated helper). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| """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.""" |
|
|
||
| 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": "*"}) |
| 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 |
| 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." | ||
| ) |
| 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 " |
|
Minor issue in test documentation: The test file tests/gateway/test_telegram_auth_fail_closed.py contains a docstring that references a non-existent function: The actual function being modified is , not . The function does not exist in the codebase. Suggestion: Update the docstring to reference the correct function name for accuracy. Note: This doesn't affect the test logic (which correctly replicates the auth check), just the documentation within the test. |
|
Closing as duplicate/superseded by #24667 to keep one canonical PR per fix topic and avoid review split. |
Problem
When
TELEGRAM_ALLOWED_USERSis unset,TelegramAdapter._is_authorized_user()returnsTrueunconditionally — any Telegram account that discovers the bot gets full local agent access (filesystem, terminal, GitHub). Thegateway/run.pystartup log also emits no warning whenGATEWAY_ALLOW_ALL_USERS=trueis set, the most dangerous open state.Root cause
gateway/platforms/telegram.pyline 498–499:gateway/run.pyline 3360–3365 only warns when neither allowlist nor allow-all is set; it is silent when_allow_all=True.Fix
gateway/platforms/telegram.py— fail closed when no allowlist is configured; permit only when an explicit allow-all flag is set:gateway/run.py— add a loudSECURITY WARNINGlog when*_ALLOW_ALL_USERS=trueis active, so operators are never silently in the open state.flowchart TD A[Telegram update received] --> B{auth_fn available?} B -- yes --> C[auth_fn result] B -- no / raises --> D{TELEGRAM_ALLOWED_USERS set?} D -- yes --> E{user_id in allowlist or wildcard?} E -- yes --> PERMIT E -- no --> DENY D -- no --> F{GATEWAY_ALLOW_ALL_USERS or TELEGRAM_ALLOW_ALL_USERS = true?} F -- yes --> PERMIT F -- no --> DENYTests
New
tests/gateway/test_telegram_auth_fail_closed.py— 7 cases:False(deny)GATEWAY_ALLOW_ALL_USERS=trueTrue(permit)TELEGRAM_ALLOW_ALL_USERS=1True(permit)GATEWAY_ALLOW_ALL_USERS=falseFalse(deny)TELEGRAM_ALLOWED_USERSTrue(permit)TELEGRAM_ALLOWED_USERSFalse(deny)*in allowlistTrue(permit)All 7 pass.
Closes #24457