common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(telegram): support group user allowlist

Browse Source
This commit is contained in:
Anders Bell
2026-04-30 03:19:34 +03:00
committed by Teknium
parent dd2d1ba5e6
commit 1f712173b2
4 changed files with 222 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
tests/gateway/test_telegram_group_gating.py
View File

@@ -5,7 +5,14 @@ from unittest.mock import AsyncMock
from gateway.config import Platform, PlatformConfig, load_gateway_config
def _make_adapter(require_mention=None, free_response_chats=None, mention_patterns=None, ignored_threads=None):
def _make_adapter(
require_mention=None,
free_response_chats=None,
mention_patterns=None,
ignored_threads=None,
allow_from=None,
group_allow_from=None,
):
from gateway.platforms.telegram import TelegramAdapter
extra = {}
@@ -17,6 +24,10 @@ def _make_adapter(require_mention=None, free_response_chats=None, mention_patter
extra["mention_patterns"] = mention_patterns
if ignored_threads is not None:
extra["ignored_threads"] = ignored_threads
if allow_from is not None:
extra["allow_from"] = allow_from
if group_allow_from is not None:
extra["group_allow_from"] = group_allow_from
adapter = object.__new__(TelegramAdapter)
adapter.platform = Platform.TELEGRAM
@@ -34,6 +45,7 @@ def _group_message(
text="hello",
*,
chat_id=-100,
from_user_id=111,
thread_id=None,
reply_to_bot=False,
entities=None,
@@ -50,10 +62,24 @@ def _group_message(
caption_entities=caption_entities or [],
message_thread_id=thread_id,
chat=SimpleNamespace(id=chat_id, type="group"),
from_user=SimpleNamespace(id=from_user_id),
reply_to_message=reply_to_message,
)
def _dm_message(text="hello", *, from_user_id=111):
return SimpleNamespace(
text=text,
caption=None,
entities=[],
caption_entities=[],
message_thread_id=None,
chat=SimpleNamespace(id=from_user_id, type="private"),
from_user=SimpleNamespace(id=from_user_id),
reply_to_message=None,
)
def _mention_entity(text, mention="@hermes_bot"):
offset = text.index(mention)
return SimpleNamespace(type="mention", offset=offset, length=len(mention))
@@ -173,6 +199,68 @@ def test_config_bridges_telegram_group_settings(monkeypatch, tmp_path):
assert __import__("os").environ["TELEGRAM_FREE_RESPONSE_CHATS"] == "-123"
def test_config_bridges_telegram_user_allowlists(monkeypatch, tmp_path):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text(
"telegram:\n"
" allow_from:\n"
" - \"111\"\n"
" - \"222\"\n"
" group_allow_from:\n"
" - \"333\"\n"
" group_allowed_chats:\n"
" - \"-100\"\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.delenv("TELEGRAM_ALLOWED_USERS", raising=False)
monkeypatch.delenv("TELEGRAM_GROUP_ALLOWED_USERS", raising=False)
monkeypatch.delenv("TELEGRAM_GROUP_ALLOWED_CHATS", raising=False)
config = load_gateway_config()
assert config is not None
assert __import__("os").environ["TELEGRAM_ALLOWED_USERS"] == "111,222"
assert __import__("os").environ["TELEGRAM_GROUP_ALLOWED_USERS"] == "333"
assert __import__("os").environ["TELEGRAM_GROUP_ALLOWED_CHATS"] == "-100"
def test_config_env_overrides_telegram_user_allowlists(monkeypatch, tmp_path):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text(
"telegram:\n"
" allow_from: \"111\"\n"
" group_allow_from: \"222\"\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "999")
monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_USERS", "888")
config = load_gateway_config()
assert config is not None
assert __import__("os").environ["TELEGRAM_ALLOWED_USERS"] == "999"
assert __import__("os").environ["TELEGRAM_GROUP_ALLOWED_USERS"] == "888"
def test_dm_allow_from_is_enforced_by_gateway_authorization_not_trigger_gate():
adapter = _make_adapter(allow_from=["111", "222"])
assert adapter._should_process_message(_dm_message("hello", from_user_id=111)) is True
assert adapter._should_process_message(_dm_message("hello", from_user_id=333)) is True
def test_group_allow_from_is_enforced_by_gateway_authorization_not_trigger_gate():
adapter = _make_adapter(group_allow_from=["111"])
assert adapter._should_process_message(_group_message("hello", from_user_id=333)) is True
def test_config_bridges_telegram_ignored_threads(monkeypatch, tmp_path):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()

82
tests/gateway/test_unauthorized_dm_behavior.py
View File

@@ -16,6 +16,8 @@ def _clear_auth_env(monkeypatch) -> None:
"WHATSAPP_ALLOWED_USERS",
"SLACK_ALLOWED_USERS",
"SIGNAL_ALLOWED_USERS",
"SIGNAL_GROUP_ALLOWED_USERS",
"TELEGRAM_GROUP_ALLOWED_CHATS",
"EMAIL_ALLOWED_USERS",
"SMS_ALLOWED_USERS",
"MATTERMOST_ALLOWED_USERS",
@@ -178,9 +180,85 @@ def test_qq_group_allowlist_does_not_authorize_other_groups(monkeypatch):
assert runner._is_user_authorized(source) is False
def test_telegram_group_allowlist_authorizes_forum_chat_without_user_allowlist(monkeypatch):
def test_telegram_group_user_allowlist_authorizes_forum_sender_without_dm_allowlist(monkeypatch):
_clear_auth_env(monkeypatch)
monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_USERS", "-1001878443972")
monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_USERS", "999")
runner, _adapter = _make_runner(
Platform.TELEGRAM,
GatewayConfig(platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="t")}),
)
source = SessionSource(
platform=Platform.TELEGRAM,
user_id="999",
chat_id="-1001878443972",
user_name="tester",
chat_type="forum",
)
assert runner._is_user_authorized(source) is True
def test_telegram_group_user_allowlist_rejects_other_senders(monkeypatch):
_clear_auth_env(monkeypatch)
monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_USERS", "999")
runner, _adapter = _make_runner(
Platform.TELEGRAM,
GatewayConfig(platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="t")}),
)
source = SessionSource(
platform=Platform.TELEGRAM,
user_id="123",
chat_id="-1001878443972",
user_name="tester",
chat_type="group",
)
assert runner._is_user_authorized(source) is False
def test_telegram_group_user_allowlist_wildcard_authorizes_any_sender(monkeypatch):
_clear_auth_env(monkeypatch)
monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_USERS", "*")
runner, _adapter = _make_runner(
Platform.TELEGRAM,
GatewayConfig(platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="t")}),
)
source = SessionSource(
platform=Platform.TELEGRAM,
user_id="123",
chat_id="-1001878443972",
user_name="tester",
chat_type="group",
)
assert runner._is_user_authorized(source) is True
def test_telegram_group_user_allowlist_does_not_authorize_dms(monkeypatch):
_clear_auth_env(monkeypatch)
monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_USERS", "999")
runner, _adapter = _make_runner(
Platform.TELEGRAM,
GatewayConfig(platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="t")}),
)
source = SessionSource(
platform=Platform.TELEGRAM,
user_id="999",
chat_id="999",
user_name="tester",
chat_type="dm",
)
assert runner._is_user_authorized(source) is False
def test_telegram_group_chat_allowlist_authorizes_group_chat_without_user_allowlist(monkeypatch):
_clear_auth_env(monkeypatch)
monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_CHATS", "-1001878443972")
runner, _adapter = _make_runner(
Platform.TELEGRAM,