diff --git a/backend/app/channels/wechat.py b/backend/app/channels/wechat.py index d15a542ad5..21f72400d7 100644 --- a/backend/app/channels/wechat.py +++ b/backend/app/channels/wechat.py @@ -1691,9 +1691,40 @@ class WechatChannel(Channel): @staticmethod def _coerce_str_set(value: Any, default: frozenset[str]) -> set[str]: - if not isinstance(value, (list, tuple, set, frozenset)): + """Normalize a configured extension allowlist, falling back to *default*. + + A bare string is the single entry it names — the same shape + ``_coerce_host_suffixes`` accepts one line below — so + ``allowed_file_extensions: .pdf`` restricts the channel instead of + silently widening back to the default allowlist. Non-string entries are + reported and skipped rather than stringified into bogus members such as + ``.123``, every fallback to *default* is logged, and a scalar string + that looks like several entries warns the way ``_parse_allowed_users`` + does: this set gates which files the Gateway may read, so a configured + restriction that changes shape without a trace is a posture change the + operator cannot see. + """ + if value is None: return set(default) - normalized = {str(item).strip().lower() if str(item).strip().startswith(".") else f".{str(item).strip().lower()}" for item in value if str(item).strip()} + if isinstance(value, str): + if "," in value or any(char.isspace() for char in value.strip()): + logger.warning("[WeChat] allowed_file_extensions is a scalar string containing separators; treating it as one literal extension. Use a YAML list for multiple extensions") + values: list[Any] = [value] + elif isinstance(value, (list, tuple, set, frozenset)): + values = list(value) + else: + logger.warning("[WeChat] ignoring allowed_file_extensions %r: expected a string or a list of strings, using the %d built-in extensions instead", value, len(default)) + return set(default) + normalized: set[str] = set() + for entry in values: + if not isinstance(entry, str): + logger.warning("[WeChat] ignoring allowed_file_extensions entry %r: expected a file extension string", entry) + continue + text = entry.strip().lower() + if text: + normalized.add(text if text.startswith(".") else f".{text}") + if not normalized: + logger.warning("[WeChat] allowed_file_extensions %r lists no usable extension, using the %d built-in extensions instead", value, len(default)) return normalized or set(default) def _coerce_host_suffixes(self, value: Any) -> frozenset[str]: diff --git a/backend/tests/test_wechat_allowed_file_extensions.py b/backend/tests/test_wechat_allowed_file_extensions.py new file mode 100644 index 0000000000..6540c540d7 --- /dev/null +++ b/backend/tests/test_wechat_allowed_file_extensions.py @@ -0,0 +1,107 @@ +"""The WeChat file-extension allowlist is only ever narrowed by entries it can read.""" + +from __future__ import annotations + +import logging + +from app.channels.message_bus import MessageBus +from app.channels.wechat import WechatChannel + +_CONFIG = {"bot_token": "test-token"} +_DEFAULTS = WechatChannel.DEFAULT_ALLOWED_FILE_EXTENSIONS + + +def _channel(extensions=None) -> WechatChannel: + config = dict(_CONFIG) + if extensions is not None: + config["allowed_file_extensions"] = extensions + return WechatChannel(bus=MessageBus(), config=config) + + +def _warning_messages(caplog) -> list[str]: + return [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] + + +def test_absent_allowed_file_extensions_uses_the_built_in_defaults(): + channel = _channel() + + assert len(_DEFAULTS) > 1 + assert channel._allowed_file_extensions == set(_DEFAULTS) + + +def test_documented_list_narrows_the_allowlist(): + channel = _channel([".pdf", ".zip"]) + + assert channel._allowed_file_extensions == {".pdf", ".zip"} + assert channel._is_allowed_file_type("report.pdf", "application/pdf") + assert not channel._is_allowed_file_type("notes.txt", "text/plain") + + +def test_dotless_and_padded_extension_names_are_normalized(): + channel = _channel(["pdf", " .DOC "]) + + assert channel._allowed_file_extensions == {".pdf", ".doc"} + + +def test_scalar_extension_config_narrows_the_allowlist(caplog): + """A single extension may be configured as a bare string, like its sibling host allowlist. + + ``allowed_file_extensions: .pdf`` is a restriction, so it must never widen + back to the built-in defaults; only the named extension is admitted. + """ + caplog.set_level(logging.WARNING, logger="app.channels.wechat") + channel = _channel(".pdf") + + assert channel._allowed_file_extensions == {".pdf"} + assert not channel._is_allowed_file_type("archive.zip", "application/zip") + assert _warning_messages(caplog) == [] + + +def test_scalar_extension_config_with_separators_warns(caplog): + """A comma-separated scalar is one literal entry, and that must be visible at startup. + + ``allowed_file_extensions: ".pdf, .zip"`` reads like a list but yields the + single bogus member ``{".pdf, .zip"}``, which matches no suffix and so blocks + every file. The value stays fail-closed; ``_parse_allowed_users`` logs the + same shape instead of leaving the operator to infer it from per-file warnings. + """ + caplog.set_level(logging.WARNING, logger="app.channels.wechat") + channel = _channel(".pdf, .zip") + + assert channel._allowed_file_extensions == {".pdf, .zip"} + assert not channel._is_allowed_file_type("report.pdf", "application/pdf") + assert any("scalar string containing separators" in message for message in _warning_messages(caplog)) + + +def test_space_separated_scalar_extension_config_warns(caplog): + caplog.set_level(logging.WARNING, logger="app.channels.wechat") + channel = _channel(".pdf .zip") + + assert channel._allowed_file_extensions == {".pdf .zip"} + assert any("scalar string containing separators" in message for message in _warning_messages(caplog)) + + +def test_unreadable_extension_entry_is_skipped_and_logged(caplog): + """A non-string entry must not become a bogus member such as ``.123``.""" + caplog.set_level(logging.WARNING, logger="app.channels.wechat") + channel = _channel(["pdf", 123]) + + assert channel._allowed_file_extensions == {".pdf"} + assert any("allowed_file_extensions" in message and "123" in message for message in _warning_messages(caplog)) + + +def test_extension_config_of_the_wrong_type_is_logged_before_falling_back(caplog): + caplog.set_level(logging.WARNING, logger="app.channels.wechat") + channel = _channel(123) + + assert channel._allowed_file_extensions == set(_DEFAULTS) + assert any("allowed_file_extensions" in message for message in _warning_messages(caplog)) + + +def test_empty_extension_config_stays_restricted(caplog): + """An empty config value must not read as "no filter" at the gate.""" + caplog.set_level(logging.WARNING, logger="app.channels.wechat") + channel = _channel([]) + + assert channel._allowed_file_extensions == set(_DEFAULTS) + assert not channel._is_allowed_file_type("setup.exe", "application/x-msdownload")