fix(gateway): don't treat dm_policy: pairing as open access on own-policy adapters

gateway/run.py

@@ -6987,6 +6987,38 @@ class GatewayRunner:
             return False
         return bool(getattr(adapter, "enforces_own_access_policy", False))
 
+    def _adapter_dm_policy(self, platform: Optional[Platform]) -> str:
+        """Best-effort read of an own-policy adapter's effective DM policy.
+
+        Returns the lowercased ``dm_policy`` (``"open"`` / ``"allowlist"`` /
+        ``"disabled"`` / ``"pairing"``) for *platform*, or ``""`` when unknown.
+        Prefers the live adapter's resolved ``_dm_policy`` — which already folds
+        in both ``config.extra`` and the ``<PLATFORM>_DM_POLICY`` env var (the
+        env var is not always bridged back into ``config.extra``) — and falls
+        back to ``config.extra`` for bare runners built without a live adapter.
+
+        Used by ``_is_user_authorized`` to carve ``dm_policy: pairing`` out of
+        the adapter-trust shortcut: in pairing mode the adapter forwards the DM
+        so the gateway can run its pairing handshake, so "reached the gateway"
+        must not be read as "authorized".
+        """
+        if not platform:
+            return ""
+        adapters = getattr(self, "adapters", None) or {}
+        adapter = adapters.get(platform)
+        policy = getattr(adapter, "_dm_policy", None) if adapter is not None else None
+        if policy 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):
+                policy = extra.get("dm_policy")
+        return str(policy or "").strip().lower()
+
     def _is_user_authorized(self, source: SessionSource) -> bool:
         """
         Check if a user is authorized to use the bot.
@@ -7134,7 +7166,20 @@ class GatewayRunner:
             # env-only default-deny below, which would silently break
             # `dm_policy: open` and config-only allowlists. (#34515)
             if self._adapter_enforces_own_access_policy(source.platform):
-                return True
+                # Exception: `dm_policy: pairing` does NOT authorize at intake.
+                # The adapter forwards the DM precisely so the gateway can run
+                # its pairing handshake (issue a code, consult the pairing
+                # store). The pairing-store approval check above already ran and
+                # returned False for this sender, so blanket-trusting the
+                # adapter here would silently turn pairing mode into open
+                # access. Fall through to default-deny so the unpaired sender is
+                # offered a pairing code instead. (Pairing is DM-only; group
+                # traffic keeps the adapter-trust path.)
+                if not (
+                    source.chat_type == "dm"
+                    and self._adapter_dm_policy(source.platform) == "pairing"
+                ):
+                    return True
             # No allowlists configured -- check global allow-all flag
             return os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in {"true", "1", "yes"}
 

tests/gateway/test_config_driven_access_policy.py

@@ -203,6 +203,76 @@ def test_unknown_adapter_does_not_crash_trust_check(monkeypatch):
     assert runner._is_user_authorized(_source(Platform.WECOM)) is False
 
 
+# ---------------------------------------------------------------------------
+# Layer 2b: `dm_policy: pairing` is NOT blanket-trusted
+# ---------------------------------------------------------------------------
+#
+# Regression: WeCom/Weixin document ``dm_policy: pairing`` and declare
+# ``enforces_own_access_policy=True``, but their intake helper only special-cases
+# ``disabled`` / ``allowlist`` — ``pairing`` falls through and forwards the DM so
+# the gateway can run its pairing handshake. With no env allowlist, the
+# adapter-trust shortcut above then authorized *every* unpaired sender, silently
+# degrading pairing mode to open access. The shortcut must skip pairing-mode DMs
+# so an unpaired sender falls through to default-deny (and gets a pairing code).
+
+
+@pytest.mark.parametrize("platform", [Platform.WECOM, Platform.WEIXIN])
+def test_pairing_dm_policy_not_blanket_authorized(monkeypatch, platform):
+    """An unpaired sender in ``dm_policy: pairing`` is NOT authorized."""
+    _clear_auth_env(monkeypatch)
+    config = GatewayConfig(
+        platforms={platform: PlatformConfig(enabled=True, extra={"dm_policy": "pairing"})}
+    )
+    runner, _adapter = _make_runner(platform, config, enforces=True)
+    # pairing_store.is_approved already returns False (set in _make_runner).
+
+    assert runner._is_user_authorized(_source(platform)) is False
+
+
+def test_pairing_dm_policy_authorizes_paired_user(monkeypatch):
+    """Once approved in the pairing store, the sender authorizes normally."""
+    _clear_auth_env(monkeypatch)
+    config = GatewayConfig(
+        platforms={Platform.WECOM: PlatformConfig(enabled=True, extra={"dm_policy": "pairing"})}
+    )
+    runner, _adapter = _make_runner(Platform.WECOM, config, enforces=True)
+    runner.pairing_store.is_approved.return_value = True
+
+    assert runner._is_user_authorized(_source(Platform.WECOM)) is True
+
+
+def test_pairing_carveout_reads_adapter_when_env_set(monkeypatch):
+    """Env-only ``WECOM_DM_POLICY=pairing`` (absent from config.extra) is honored.
+
+    The adapter resolves ``dm_policy`` from the env var, so its ``_dm_policy`` is
+    authoritative even when ``config.extra`` is empty. The carve-out must read
+    that, not just config.
+    """
+    _clear_auth_env(monkeypatch)
+    config = GatewayConfig(
+        platforms={Platform.WECOM: PlatformConfig(enabled=True, extra={})}
+    )
+    runner, adapter = _make_runner(Platform.WECOM, config, enforces=True)
+    adapter._dm_policy = "pairing"  # as the adapter would resolve from the env var
+
+    assert runner._is_user_authorized(_source(Platform.WECOM)) is False
+
+
+def test_pairing_dm_policy_group_chat_still_trusted(monkeypatch):
+    """Pairing is DM-only — group traffic keeps the adapter-trust path."""
+    _clear_auth_env(monkeypatch)
+    config = GatewayConfig(
+        platforms={
+            Platform.WECOM: PlatformConfig(
+                enabled=True, extra={"dm_policy": "pairing", "group_policy": "open"}
+            )
+        }
+    )
+    runner, _adapter = _make_runner(Platform.WECOM, config, enforces=True)
+
+    assert runner._is_user_authorized(_source(Platform.WECOM, chat_type="group")) is True
+
+
 # ---------------------------------------------------------------------------
 # Layer 3: unauthorized-DM behavior reads config dm_policy
 # ---------------------------------------------------------------------------