From ff46376614e0175c4c20a75a30604dd4a9ca79de Mon Sep 17 00:00:00 2001 From: yungchentang <46495124+yungchentang@users.noreply.github.com> Date: Thu, 16 Jul 2026 02:13:49 +0200 Subject: [PATCH] fix(gateway): preserve shared route transport adapter --- gateway/authz_mixin.py | 52 +++++++++++++++++-- gateway/platforms/base.py | 8 ++- tests/gateway/test_multiplex_profile_authz.py | 28 +++++++++- tests/gateway/test_profile_resolution.py | 20 ++++++- 4 files changed, 100 insertions(+), 8 deletions(-) diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 24910144d4ef..1e4b73fc4b64 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -92,6 +92,9 @@ class GatewayAuthorizationMixin: """Resolve the live adapter for an inbound ``SessionSource``.""" if source is None: return None + transport_adapter = self._registered_transport_adapter(source) + if transport_adapter is not None: + return transport_adapter # ``getattr`` guards test fixtures that build a bare source via # SimpleNamespace and omit ``profile`` (see AGENTS.md pitfall #17). return self._authorization_adapter( @@ -99,6 +102,43 @@ class GatewayAuthorizationMixin: getattr(source, "profile", None), ) + def _registered_transport_adapter(self, source: SessionSource): + """Return the registered adapter that created *source*, if retained. + + ``source.profile`` is the runtime/session namespace. A chat-based + profile route can therefore differ from the adapter profile when one + shared credential serves several routed runtimes. ``build_source`` + keeps the receiving adapter as in-process provenance so replies and + intake-policy checks stay on that transport without weakening the + fail-closed fallback for restored or hand-built sources. + """ + adapter_ref = getattr(source, "_transport_adapter_ref", None) + adapter = adapter_ref() if callable(adapter_ref) else None + platform = getattr(source, "platform", None) + if adapter is None or platform is None: + return None + if adapter is (getattr(self, "adapters", None) or {}).get(platform): + return adapter + profile_maps = getattr(self, "_profile_adapters", None) or {} + for profile_adapters in profile_maps.values(): + if adapter is profile_adapters.get(platform): + return adapter + return None + + def _adapter_profile_for_source(self, source: SessionSource) -> Optional[str]: + """Resolve the transport-owning profile for adapter policy lookups.""" + adapter = self._registered_transport_adapter(source) + platform = getattr(source, "platform", None) + if adapter is not None: + if adapter is (getattr(self, "adapters", None) or {}).get(platform): + return None + for profile, profile_adapters in ( + getattr(self, "_profile_adapters", None) or {} + ).items(): + if adapter is profile_adapters.get(platform): + return profile + return getattr(source, "profile", None) + def _adapter_authorization_is_upstream( self, platform: Optional[Platform], @@ -311,6 +351,8 @@ class GatewayAuthorizationMixin: if source.platform in {Platform.HOMEASSISTANT, Platform.WEBHOOK}: return True + adapter_profile = self._adapter_profile_for_source(source) + # Relay (and any adapter whose authorization is enforced by a trusted # authenticated upstream): the Team Gateway connector authenticates this # gateway's WS with a per-instance secret and resolves owner-only author @@ -340,7 +382,7 @@ class GatewayAuthorizationMixin: # tests) — defensive against accidental fail-open. if source.delivered_via_upstream_relay is True or self._adapter_authorization_is_upstream( source.platform, - profile=source.profile, + profile=adapter_profile, ): return True @@ -538,23 +580,23 @@ class GatewayAuthorizationMixin: # fail-open.) if self._adapter_enforces_own_access_policy( source.platform, - profile=source.profile, + profile=adapter_profile, ): if source.chat_type in {"group", "forum", "channel"}: effective_policy = self._adapter_group_policy( source.platform, - profile=source.profile, + profile=adapter_profile, ) if self._adapter_group_has_sender_allowlist( source.platform, source.chat_id, - profile=source.profile, + profile=adapter_profile, ): return True else: effective_policy = self._adapter_dm_policy( source.platform, - profile=source.profile, + profile=adapter_profile, ) if effective_policy == "allowlist": return True diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 7238c2c50952..c7ac4be0520d 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -17,6 +17,7 @@ import subprocess import sys import time import uuid +import weakref from abc import ABC, abstractmethod from urllib.parse import urlsplit @@ -5788,7 +5789,7 @@ class BasePlatformAdapter(ABC): self.platform, chat_id, exc_info=True, ) - return SessionSource( + source = SessionSource( platform=self.platform, chat_id=str(chat_id), chat_name=chat_name, @@ -5809,6 +5810,11 @@ class BasePlatformAdapter(ABC): auto_thread_created=auto_thread_created, auto_thread_initial_name=auto_thread_initial_name, ) + # In-process transport provenance is deliberately not serialized by + # SessionSource.to_dict(). The live receiving adapter is authoritative + # for this turn even when profile_routes selects a different runtime. + source._transport_adapter_ref = weakref.ref(self) + return source @abstractmethod async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: diff --git a/tests/gateway/test_multiplex_profile_authz.py b/tests/gateway/test_multiplex_profile_authz.py index fca5060b5b31..4289fcb1064d 100644 --- a/tests/gateway/test_multiplex_profile_authz.py +++ b/tests/gateway/test_multiplex_profile_authz.py @@ -127,6 +127,32 @@ def test_adapter_for_source_resolves_secondary_profile_adapter(monkeypatch): ) is default_adapter +def test_chat_routed_source_keeps_receiving_shared_adapter(monkeypatch): + """A runtime-only profile route must not discard the shared transport. + + ``source.profile`` selects the routed runtime/session namespace, but the + adapter that built the source still owns outbound delivery and intake + policy when that profile has no credential of its own. + """ + runner, default_adapter, _secondary_adapter = _make_multiplex_runner( + monkeypatch + ) + runner._profile_adapters["routed"] = {} + + source = SessionSource( + platform=Platform.WECOM, + user_id="allowed-user", + chat_id="dm-chat", + user_name="allowed-user", + chat_type="dm", + profile="routed", + ) + assert runner._adapter_for_source(source) is None + source._transport_adapter_ref = lambda: default_adapter + assert runner._adapter_for_source(source) is default_adapter + assert runner._is_user_authorized(source) is True + + def test_secondary_allowlist_dm_behavior_ignores_unauthorized(monkeypatch): """Unauthorized-DM behavior must read the secondary adapter's dm_policy.""" runner, _default_adapter, secondary_adapter = _make_multiplex_runner(monkeypatch) @@ -206,4 +232,4 @@ def test_secondary_open_policy_fails_startup_guard(monkeypatch): violation = _own_policy_open_startup_violation(secondary_cfg) assert violation is not None assert "wecom" in violation - assert "open policy" in violation \ No newline at end of file + assert "open policy" in violation diff --git a/tests/gateway/test_profile_resolution.py b/tests/gateway/test_profile_resolution.py index 799f20c2589c..3535ede2edad 100644 --- a/tests/gateway/test_profile_resolution.py +++ b/tests/gateway/test_profile_resolution.py @@ -9,7 +9,7 @@ import pytest from gateway.session import SessionSource, build_session_key from gateway.run import GatewayRunner from gateway.profile_routing import ProfileRoute -from gateway.config import Platform +from gateway.config import GatewayConfig, Platform from gateway.platforms.base import BasePlatformAdapter @@ -400,11 +400,29 @@ class TestAdapterToSessionKeyIntegration: chat_id="-1001234567890", chat_type="group", user_id="u1", ) assert source.profile == "ops" + assert source._transport_adapter_ref() is adapter key = build_session_key(source, profile=source.profile) assert key.startswith("agent:ops:"), key assert key != build_session_key(source, profile=None) + def test_chat_route_keeps_shared_adapter_for_delivery(self): + runner = object.__new__(GatewayRunner) + runner.config = GatewayConfig( + multiplex_profiles=True, + profile_routes=self._routes(), + ) + runner._profile_adapters = {"ops": {}} + adapter = _stub_adapter(Platform.TELEGRAM, runner) + runner.adapters = {Platform.TELEGRAM: adapter} + + source = adapter.build_source( + chat_id="-1001234567890", chat_type="group", user_id="u1", + ) + + assert source.profile == "ops" + assert runner._adapter_for_source(source) is adapter + def test_adapter_without_runner_falls_back_to_default_namespace(self, mock_runner): """Regression anchor: with no ``gateway_runner`` injected (the pre-fix state for non-Discord adapters), ``build_source`` leaves ``profile=None``