feat(gateway): add allowed_{chats,channels,rooms} whitelist to Telegram, Mattermost, Matrix, DingTalk

Mirrors the Slack `allowed_channels` feature (PR #7401) and Discord's
`allowed_channels` (PR #7044) across the remaining group-capable platforms.
All five platforms (Slack + Discord + the four added here) now follow the
same pattern: primary config via config.yaml, env-var fallback as an escape
hatch — matching the project policy that .env is for secrets only and
behavioral settings belong in config.yaml.

Also fixes a duplicate `slack` key in DEFAULT_CONFIG introduced by PR
#7401 (the later entry silently overwrote `allowed_channels`, `require_mention`,
and `free_response_channels` at dict-literal evaluation time).

Platforms added:
- Telegram: `telegram.allowed_chats` (env alias: `TELEGRAM_ALLOWED_CHATS`)
- Mattermost: `mattermost.allowed_channels` (env alias: `MATTERMOST_ALLOWED_CHANNELS`)
- Matrix: `matrix.allowed_rooms` (env alias: `MATRIX_ALLOWED_ROOMS`)
- DingTalk: `dingtalk.allowed_chats` (env alias: `DINGTALK_ALLOWED_CHATS`)

Mattermost and Matrix previously had NO config.yaml bridging for any of
their gating settings; this PR adds `load_gateway_config` bridges for them
(Mattermost gets require_mention + free_response_channels + allowed_channels;
Matrix gets allowed_rooms on top of its existing bridges for require_mention
and free_response_rooms).

Semantics identical everywhere:
- Empty = no restriction (fully backward compatible).
- Non-empty = hard whitelist: non-listed chats are silently ignored,
  even when the bot is @mentioned.
- DMs bypass the check entirely.

DEFAULT_CONFIG merges the duplicate `slack` block and adds new `mattermost`
and `matrix` blocks so all gating settings surface in defaults.

Not included: Feishu (has its own per-chat `chat_rules` system that covers
this use case differently), WhatsApp (already has `group_allow_from` via
`group_policy: allowlist`), pure-DM platforms (Signal, SMS, BlueBubbles,
Yuanbao — no group concept).
This commit is contained in:
Teknium 2026-05-07 05:58:56 -07:00
parent f5c9bb582c
commit 69d025e4a7
7 changed files with 518 additions and 9 deletions

View file

@ -0,0 +1,364 @@
"""Tests for the allowed_{channels,chats,rooms} whitelist extension
added alongside PR #7401 (Slack).
Covers: Telegram, Matrix, Mattermost, DingTalk.
For each platform:
- Empty = no restriction (fully backward compatible).
- When set, messages from non-listed chats/rooms are silently ignored.
- DMs are never filtered.
- @mention does NOT bypass the whitelist.
- config.yaml env var bridging (via load_gateway_config) where applicable.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from gateway.config import Platform, PlatformConfig
# ---------------------------------------------------------------------------
# Telegram
# ---------------------------------------------------------------------------
def _make_telegram_adapter(*, allowed_chats=None, require_mention=None):
from gateway.platforms.telegram import TelegramAdapter
extra = {}
if allowed_chats is not None:
extra["allowed_chats"] = allowed_chats
if require_mention is not None:
extra["require_mention"] = require_mention
adapter = object.__new__(TelegramAdapter)
adapter.platform = Platform.TELEGRAM
adapter.config = PlatformConfig(enabled=True, token="***", extra=extra)
adapter._bot = SimpleNamespace(id=999, username="hermes_bot")
adapter._message_handler = AsyncMock()
adapter._mention_patterns = adapter._compile_mention_patterns()
return adapter
def _tg_group_message(chat_id=-100, text="hello"):
return SimpleNamespace(
text=text,
caption=None,
entities=[],
caption_entities=[],
message_thread_id=None,
chat=SimpleNamespace(id=chat_id, type="group"),
from_user=SimpleNamespace(id=111),
reply_to_message=None,
)
def _tg_dm_message(text="hello"):
return SimpleNamespace(
text=text,
caption=None,
entities=[],
caption_entities=[],
message_thread_id=None,
chat=SimpleNamespace(id=111, type="private"),
from_user=SimpleNamespace(id=111),
reply_to_message=None,
)
class TestTelegramAllowedChats:
def test_empty_is_no_restriction(self, monkeypatch):
monkeypatch.delenv("TELEGRAM_ALLOWED_CHATS", raising=False)
adapter = _make_telegram_adapter()
assert adapter._telegram_allowed_chats() == set()
assert adapter._should_process_message(_tg_group_message(-100)) is True
def test_list_form(self):
adapter = _make_telegram_adapter(allowed_chats=[-100, -200])
assert adapter._telegram_allowed_chats() == {"-100", "-200"}
def test_csv_form(self):
adapter = _make_telegram_adapter(allowed_chats="-100, -200")
assert adapter._telegram_allowed_chats() == {"-100", "-200"}
def test_env_var_fallback(self, monkeypatch):
monkeypatch.setenv("TELEGRAM_ALLOWED_CHATS", "-100,-200")
adapter = _make_telegram_adapter() # no extra → falls back to env
assert adapter._telegram_allowed_chats() == {"-100", "-200"}
def test_blocks_non_whitelisted_group(self):
adapter = _make_telegram_adapter(allowed_chats=["-100"])
assert adapter._should_process_message(_tg_group_message(-999)) is False
def test_permits_whitelisted_group(self):
adapter = _make_telegram_adapter(
allowed_chats=["-100"], require_mention=False,
)
assert adapter._should_process_message(_tg_group_message(-100)) is True
def test_mention_cannot_bypass_whitelist(self):
"""@mention in a non-allowed chat is still ignored."""
adapter = _make_telegram_adapter(allowed_chats=["-100"])
msg = _tg_group_message(-999, text="@hermes_bot hello")
msg.entities = [SimpleNamespace(
type="mention", offset=0, length=len("@hermes_bot"),
)]
assert adapter._should_process_message(msg) is False
def test_dms_unaffected(self):
"""DMs bypass the allowed_chats whitelist entirely."""
adapter = _make_telegram_adapter(allowed_chats=["-100"])
assert adapter._should_process_message(_tg_dm_message()) is True
def test_config_bridge(self, monkeypatch, tmp_path):
"""slack-style config.yaml → env var bridge works."""
from gateway.config import load_gateway_config
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text(
"telegram:\n"
" allowed_chats:\n"
" - -100\n"
" - -200\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("TELEGRAM_ALLOWED_CHATS", "__sentinel__")
monkeypatch.delenv("TELEGRAM_ALLOWED_CHATS")
load_gateway_config()
import os as _os
assert _os.environ["TELEGRAM_ALLOWED_CHATS"] == "-100,-200"
def test_config_bridge_env_takes_precedence(self, monkeypatch, tmp_path):
from gateway.config import load_gateway_config
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text(
"telegram:\n"
" allowed_chats: -100\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("TELEGRAM_ALLOWED_CHATS", "-999")
load_gateway_config()
import os as _os
assert _os.environ["TELEGRAM_ALLOWED_CHATS"] == "-999"
# ---------------------------------------------------------------------------
# DingTalk
# ---------------------------------------------------------------------------
def _make_dingtalk_adapter(*, allowed_chats=None, require_mention=None):
# Import lazily — DingTalk SDK may not be installed.
pytest.importorskip("gateway.platforms.dingtalk", reason="DingTalk adapter not importable")
from gateway.platforms.dingtalk import DingTalkAdapter
extra = {}
if allowed_chats is not None:
extra["allowed_chats"] = allowed_chats
if require_mention is not None:
extra["require_mention"] = require_mention
adapter = object.__new__(DingTalkAdapter)
adapter.platform = Platform.DINGTALK
adapter.config = PlatformConfig(enabled=True, extra=extra)
return adapter
class TestDingTalkAllowedChats:
def test_empty_is_no_restriction(self, monkeypatch):
monkeypatch.delenv("DINGTALK_ALLOWED_CHATS", raising=False)
adapter = _make_dingtalk_adapter()
assert adapter._dingtalk_allowed_chats() == set()
def test_list_form(self):
adapter = _make_dingtalk_adapter(allowed_chats=["cidABC", "cidDEF"])
assert adapter._dingtalk_allowed_chats() == {"cidABC", "cidDEF"}
def test_csv_form(self):
adapter = _make_dingtalk_adapter(allowed_chats="cidABC, cidDEF")
assert adapter._dingtalk_allowed_chats() == {"cidABC", "cidDEF"}
def test_env_var_fallback(self, monkeypatch):
monkeypatch.setenv("DINGTALK_ALLOWED_CHATS", "cidABC,cidDEF")
adapter = _make_dingtalk_adapter()
assert adapter._dingtalk_allowed_chats() == {"cidABC", "cidDEF"}
def test_blocks_non_whitelisted_group(self):
adapter = _make_dingtalk_adapter(allowed_chats=["cidABC"])
assert adapter._should_process_message(
message=None, text="hello", is_group=True, chat_id="cidXYZ",
) is False
def test_dm_unaffected(self):
"""DMs (is_group=False) bypass the whitelist."""
adapter = _make_dingtalk_adapter(allowed_chats=["cidABC"])
assert adapter._should_process_message(
message=None, text="hello", is_group=False, chat_id="cidXYZ",
) is True
def test_config_bridge(self, monkeypatch, tmp_path):
from gateway.config import load_gateway_config
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text(
"dingtalk:\n"
" allowed_chats:\n"
" - cidABC\n"
" - cidDEF\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("DINGTALK_ALLOWED_CHATS", "__sentinel__")
monkeypatch.delenv("DINGTALK_ALLOWED_CHATS")
load_gateway_config()
import os as _os
assert _os.environ["DINGTALK_ALLOWED_CHATS"] == "cidABC,cidDEF"
# ---------------------------------------------------------------------------
# Mattermost (env-var only — no config.yaml bridge)
# ---------------------------------------------------------------------------
class TestMattermostAllowedChannels:
"""Mattermost whitelist logic — replicated since the adapter reads config
with env-var fallback inline inside _handle_post rather than through a
helper method."""
@staticmethod
def _would_process(channel_id, channel_type="O", allowed_cfg=None, allowed_env=""):
"""Replicate the whitelist gate from gateway/platforms/mattermost.py."""
import os as _os
if channel_type == "D":
return True
# config-first, env-var fallback (matching the adapter)
allowed_raw = allowed_cfg
if allowed_raw is None:
allowed_raw = allowed_env
if isinstance(allowed_raw, list):
allowed = {str(c).strip() for c in allowed_raw if str(c).strip()}
else:
allowed = {c.strip() for c in str(allowed_raw).split(",") if c.strip()}
if allowed and channel_id not in allowed:
return False
return True
def test_empty_config_is_no_restriction(self):
assert self._would_process("chan123", allowed_cfg=None, allowed_env="") is True
def test_config_list_blocks_non_whitelisted_channel(self):
assert self._would_process(
"chanXYZ", allowed_cfg=["chanABC", "chanDEF"],
) is False
def test_config_list_permits_whitelisted_channel(self):
assert self._would_process(
"chanABC", allowed_cfg=["chanABC", "chanDEF"],
) is True
def test_env_var_fallback_when_no_config(self):
assert self._would_process(
"chanXYZ", allowed_cfg=None, allowed_env="chanABC,chanDEF",
) is False
def test_dm_unaffected(self):
assert self._would_process(
"chanXYZ", channel_type="D", allowed_cfg=["chanABC"],
) is True
def test_config_bridge(self, monkeypatch, tmp_path):
from gateway.config import load_gateway_config
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text(
"mattermost:\n"
" allowed_channels:\n"
" - chanABC\n"
" - chanDEF\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
# Pre-register the key with monkeypatch so teardown cleans it up
# even though load_gateway_config mutates os.environ directly
# (monkeypatch only restores keys it's touched via setenv/delenv;
# delenv on an absent key is a no-op for teardown purposes).
monkeypatch.setenv("MATTERMOST_ALLOWED_CHANNELS", "__sentinel__")
monkeypatch.delenv("MATTERMOST_ALLOWED_CHANNELS")
load_gateway_config()
import os as _os
assert _os.environ["MATTERMOST_ALLOWED_CHANNELS"] == "chanABC,chanDEF"
# ---------------------------------------------------------------------------
# Matrix
# ---------------------------------------------------------------------------
class TestMatrixAllowedRooms:
"""Matrix whitelist behavior — tested via the env-var-initialized
instance attribute _allowed_rooms."""
def test_empty_env_empty_set(self, monkeypatch):
monkeypatch.delenv("MATRIX_ALLOWED_ROOMS", raising=False)
# Replicate __init__ parsing without needing the real adapter.
raw = "" or ""
allowed = {r.strip() for r in raw.split(",") if r.strip()}
assert allowed == set()
def test_env_var_parsed_to_set(self, monkeypatch):
monkeypatch.setenv("MATRIX_ALLOWED_ROOMS", "!room1:srv,!room2:srv")
import os as _os
raw = _os.environ["MATRIX_ALLOWED_ROOMS"]
allowed = {r.strip() for r in raw.split(",") if r.strip()}
assert allowed == {"!room1:srv", "!room2:srv"}
def test_block_logic(self):
"""Replicates the matrix.py gate: if allowed non-empty and room not in it, drop."""
allowed = {"!allowed:srv"}
# Non-allowed room in group (is_dm=False) → blocked
def would_process(room_id, is_dm):
if is_dm:
return True
if allowed and room_id not in allowed:
return False
return True
assert would_process("!blocked:srv", is_dm=False) is False
assert would_process("!allowed:srv", is_dm=False) is True
# DM always allowed
assert would_process("!blocked:srv", is_dm=True) is True
def test_config_bridge(self, monkeypatch, tmp_path):
from gateway.config import load_gateway_config
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text(
"matrix:\n"
" allowed_rooms:\n"
" - '!room1:srv'\n"
" - '!room2:srv'\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("MATRIX_ALLOWED_ROOMS", "__sentinel__")
monkeypatch.delenv("MATRIX_ALLOWED_ROOMS")
load_gateway_config()
import os as _os
assert _os.environ["MATRIX_ALLOWED_ROOMS"] == "!room1:srv,!room2:srv"