teknium1
7370c82c2d
WeComAdapter mixed in OwnAccessPolicyMixin without ALLOW_ALL_ENV_PREFIX, so _allow_all_env_names() read "_ALLOW_ALL_USERS" and every open-DM WeCom deployment (setup still writes WECOM_ALLOW_ALL_USERS) silently denied all DMs. - WeComAdapter.ALLOW_ALL_ENV_PREFIX = "WECOM" - OwnAccessPolicyMixin.__init_subclass__ raises TypeError on an empty prefix so the omission cannot ship again (every other host already sets one). - Parity test now runs a third matrix row that sets each host's own <PREFIX>_ALLOW_ALL_USERS; the GATEWAY-only row is why this slipped through. Sabotage: prefix removed -> [platform] row fails even with the guard reverted.
70 lines
3.1 KiB
Python
70 lines
3.1 KiB
Python
"""Config-driven DM / group intake policy shared by the adapters that gate access themselves
|
|
(Weixin, WeCom, QQBot, WhatsApp, Yuanbao's ``AccessPolicy``).
|
|
|
|
Host contract — the adapter sets these on ``self`` before any predicate runs: ``_dm_policy`` /
|
|
``_group_policy`` (``open`` | ``allowlist`` | ``disabled`` | ``pairing``), ``_allow_from`` /
|
|
``_group_allow_from`` (iterable of allowlist entries), and ``ALLOW_ALL_ENV_PREFIX`` (class
|
|
attribute, e.g. ``"WEIXIN"``). Platform-specific allowlist matching goes through
|
|
``_entry_matches``; env-seeded allowlists that must be re-read live go through
|
|
``_live_dm_allow_from``.
|
|
|
|
Every env read is profile-scoped and fails closed: under multiplexing ``os.environ`` holds the
|
|
DEFAULT profile's opt-in, which must never open a secondary bot's DMs (#93522).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Iterable
|
|
|
|
from gateway.platforms._shared import get_scoped_secret
|
|
|
|
OPTIN_TRUTHY = frozenset({"true", "1", "yes"})
|
|
|
|
|
|
class OwnAccessPolicyMixin:
|
|
ALLOW_ALL_ENV_PREFIX: str = ""
|
|
|
|
def __init_subclass__(cls, **kwargs):
|
|
super().__init_subclass__(**kwargs)
|
|
# A host that forgets its prefix would silently read ``_ALLOW_ALL_USERS`` and deny every
|
|
# open-DM deployment whose setup wrote ``<PLATFORM>_ALLOW_ALL_USERS`` — refuse at class creation.
|
|
if not str(cls.ALLOW_ALL_ENV_PREFIX or "").strip():
|
|
raise TypeError(f"{cls.__qualname__} mixes in OwnAccessPolicyMixin but sets no ALLOW_ALL_ENV_PREFIX")
|
|
|
|
@property
|
|
def enforces_own_access_policy(self) -> bool:
|
|
return True
|
|
|
|
def _allow_all_env_names(self) -> tuple[str, ...]:
|
|
return ("GATEWAY_ALLOW_ALL_USERS", f"{self.ALLOW_ALL_ENV_PREFIX}_ALLOW_ALL_USERS")
|
|
|
|
def _open_dm_opted_in(self) -> bool:
|
|
return any(str(get_scoped_secret(name, "") or "").strip().lower() in OPTIN_TRUTHY
|
|
for name in self._allow_all_env_names())
|
|
|
|
def _entry_matches(self, entries: Iterable[str], target: str) -> bool:
|
|
return target in entries
|
|
|
|
def _live_dm_allow_from(self) -> Iterable[str]:
|
|
return self._allow_from
|
|
|
|
def _is_dm_allowed(self, sender_id: str) -> bool:
|
|
"""Strict DM authorization — pairing does not imply access."""
|
|
if self._dm_policy == "allowlist":
|
|
return self._entry_matches(self._live_dm_allow_from(), sender_id)
|
|
return self._dm_policy == "open" and self._open_dm_opted_in()
|
|
|
|
def _is_dm_intake_allowed(self, sender_id: str) -> bool:
|
|
"""Whether a DM may reach gateway intake; ``pairing`` admits everyone with a principal so the
|
|
handshake can run (the pairing gate itself runs later). A blank principal is never admitted."""
|
|
principal = str(sender_id or "").strip()
|
|
if not principal:
|
|
return False
|
|
return self._dm_policy == "pairing" or self._is_dm_allowed(principal)
|
|
|
|
def _is_group_allowed(self, chat_id: str, sender_id: str = "") -> bool:
|
|
"""``disabled``/``pairing``/unknown policies never forward group traffic at intake."""
|
|
if self._group_policy == "allowlist":
|
|
return self._entry_matches(self._group_allow_from, chat_id)
|
|
return self._group_policy == "open"
|