fix(wechat): a scalar allowed_file_extensions config silently widens back to all 37 defaults (#6737)

* test(wechat): cover allowed_file_extensions coercion boundaries

* fix(wechat): never widen a configured allowed_file_extensions back to defaults

* fix(wechat): warn when a scalar allowed_file_extensions looks like a list

A comma- or space-separated scalar string is still one literal extension, so
`allowed_file_extensions: ".pdf, .zip"` blocks every file with nothing in the
startup log pointing at the config. Mirror the `_parse_allowed_users` guard.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Grapette.L 2026-10-13 07:26:59 +08:00 committed by GitHub
parent 5ae95e6a34
commit c53621d769
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 140 additions and 2 deletions

View File

@ -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]:

View File

@ -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")