fix(gateway): preserve WeCom per-group sender allowlists

Keep the own-policy fail-closed hardening from PR #45444, but still trust WeCom groups.<id>.allow_from because the adapter already checked that sender allowlist before dispatching to gateway auth.
This commit is contained in:
Teknium
2026-06-13 07:18:54 -07:00
parent fc46354580
commit ad7436a5d9
2 changed files with 100 additions and 0 deletions
+57
View File
@@ -121,6 +121,58 @@ class GatewayAuthorizationMixin:
policy = extra.get("group_policy")
return str(policy or "").strip().lower()
def _adapter_group_has_sender_allowlist(
self,
platform: Optional[Platform],
chat_id: Optional[str],
) -> bool:
"""Whether a per-group sender allowlist gated this group message.
WeCom supports ``groups.<group_id>.allow_from`` on top of the top-level
``group_policy``. A group may be open at the chat level while still
restricting which senders inside that group can invoke Hermes. If such a
message reached the gateway, the adapter already checked that sender
allowlist, so it is a trustworthy intake decision rather than the
fail-open ``group_policy: open`` case.
"""
if not platform or not chat_id:
return False
adapters = getattr(self, "adapters", None) or {}
adapter = adapters.get(platform)
groups = getattr(adapter, "_groups", None) if adapter is not None else None
if groups is None:
config = getattr(self, "config", None)
platform_cfg = (
config.platforms.get(platform)
if config is not None and hasattr(config, "platforms")
else None
)
extra = getattr(platform_cfg, "extra", None) if platform_cfg else None
if isinstance(extra, dict):
groups = extra.get("groups")
if not isinstance(groups, dict):
return False
chat_id_str = str(chat_id)
group_cfg = groups.get(chat_id_str)
if not isinstance(group_cfg, dict):
lowered = chat_id_str.lower()
for key, value in groups.items():
if isinstance(key, str) and key.lower() == lowered and isinstance(value, dict):
group_cfg = value
break
if not isinstance(group_cfg, dict):
group_cfg = groups.get("*")
if not isinstance(group_cfg, dict):
return False
sender_allow = group_cfg.get("allow_from") or group_cfg.get("allowFrom")
if isinstance(sender_allow, str):
return bool(sender_allow.strip())
if isinstance(sender_allow, (list, tuple, set)):
return any(str(item).strip() for item in sender_allow)
return False
def _is_user_authorized(self, source: SessionSource) -> bool:
"""
Check if a user is authorized to use the bot.
@@ -297,6 +349,11 @@ class GatewayAuthorizationMixin:
if self._adapter_enforces_own_access_policy(source.platform):
if source.chat_type in {"group", "forum", "channel"}:
effective_policy = self._adapter_group_policy(source.platform)
if self._adapter_group_has_sender_allowlist(
source.platform,
source.chat_id,
):
return True
else:
effective_policy = self._adapter_dm_policy(source.platform)
if effective_policy == "allowlist":