From 0966131b31da4ded8cbfb21b9c90ec1851e6dee4 Mon Sep 17 00:00:00 2001 From: Nan Gao Date: Tue, 16 Jun 2026 17:04:39 +0200 Subject: [PATCH] fix(channels): require bound identity for user-owned IM messages (#3578) * fix(channels): require bound identity for user-owned IM messages * make format * docs: document bound identity channel config * refactor: reuse channel connection config * refactor _requires_bound_identity() * refactor from_app_config() * make format * fix: reject unbound channel chats before semaphore * security enhancement * make format * fix: enforce bound-identity admission at command entry point The bound-identity gate only ran for non-command messages in _handle_message() and as a fallback inside _handle_chat(). Commands had no equivalent boundary, so an unbound platform user could send /new and reach _create_thread() directly, creating an unowned Gateway thread and empty checkpoint. Info commands (/status, /models, /memory) likewise leaked Gateway state to unbound users. Add the same _requires_bound_identity() check at the top of _handle_command(), rejecting via _reject_unbound_channel_message() before any thread creation or Gateway query. The gate is a no-op in legacy open-bot mode (require_bound_identity=False) and auth-disabled mode. Provider-level binding flows (/connect, /start) are consumed by the provider adapter before reaching the manager, so they are unaffected. Tests: - unbound auth-enabled /new is rejected before threads.create - bound auth-enabled /new still creates the thread Co-Authored-By: Claude Opus 4.8 * fix(channels): carry workspace fallback decision on inbound messages * fix(channels): recheck bound identity by normalized workspace * fix(channels): avoid duplicate bound identity checks * fix(channels): preserve verified routing for bound identity rejects * fix(channels): clarify bound identity upgrade failures --------- Co-authored-by: Claude Opus 4.8 Co-authored-by: Willem Jiang --- backend/app/channels/manager.py | 174 +++++-- backend/app/channels/service.py | 22 +- backend/docs/IM_CHANNEL_CONNECTIONS.md | 9 + .../config/channel_connections_config.py | 1 + .../tests/test_channel_connections_config.py | 8 + backend/tests/test_channels.py | 446 ++++++++++++++++++ config.example.yaml | 5 +- 7 files changed, 631 insertions(+), 34 deletions(-) diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index b129269f9..5d1705524 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -60,6 +60,8 @@ STREAM_UPDATE_MIN_INTERVAL_SECONDS = 0.35 STREAM_MODES = ["messages-tuple", "values"] MESSAGE_STREAM_EVENTS = ("messages-tuple", "messages") THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again." +BOUND_IDENTITY_REQUIRED_MESSAGE = "Connect this channel from DeerFlow Settings, complete the in-channel connect step, then send your message again." +BOUND_IDENTITY_UNAVAILABLE_MESSAGE = "Channel connection verification is temporarily unavailable. Please try again later or contact the DeerFlow operator." CHANNEL_CAPABILITIES = { "dingtalk": {"supports_streaming": False}, @@ -150,6 +152,19 @@ class _SlashSkillCommandResolution: failure_message: str | None = None +@dataclass(frozen=True, slots=True) +class _BoundIdentityRejection: + message: str = BOUND_IDENTITY_REQUIRED_MESSAGE + # Server-side connection id that may be used only as an outbound routing + # hint for the rejection message. This is never copied from the inbound + # message; it comes from the repository re-read when available. + outbound_connection_id: str | None = None + # Server-side owner for the outbound routing connection above. It lets + # channel senders preserve per-connection context without trusting the + # rejected inbound identity assertion. + outbound_owner_user_id: str | None = None + + def _is_thread_busy_error(exc: BaseException | None) -> bool: if exc is None: return False @@ -740,6 +755,7 @@ class ChannelManager: default_session: dict[str, Any] | None = None, channel_sessions: dict[str, Any] | None = None, connection_repo: Any | None = None, + require_bound_identity: bool = False, ) -> None: self.bus = bus self.store = store @@ -750,6 +766,7 @@ class ChannelManager: self._default_session = _as_dict(default_session) self._channel_sessions = dict(channel_sessions or {}) self._connection_repo = connection_repo + self._require_bound_identity = require_bound_identity self._client = None # lazy init — langgraph_sdk async client self._channel_metadata_synced: set[str] = set() self._skill_storage: SkillStorage | None = None @@ -923,38 +940,111 @@ class ChannelManager: async def _handle_message(self, msg: InboundMessage) -> None: msg = _apply_effective_owner(msg) - async with self._semaphore: - try: + try: + # Non-command chat can be rejected before it consumes a semaphore + # slot. Commands are handled below because provider adapters consume + # binding commands before manager dispatch, and _handle_command() + # applies its own admission gate for manager-level commands. + bound_identity_rejection = None + if msg.msg_type != InboundMessageType.COMMAND: + bound_identity_rejection = await self._get_bound_identity_rejection(msg) + if bound_identity_rejection is not None: + await self._reject_unbound_channel_message(msg, bound_identity_rejection=bound_identity_rejection) + return + + async with self._semaphore: if msg.msg_type == InboundMessageType.COMMAND: await self._handle_command(msg) else: - await self._handle_chat(msg) - except InvalidChannelSessionConfigError as exc: - logger.warning( - "Invalid channel session config for %s (chat=%s): %s", - msg.channel_name, - msg.chat_id, - exc, - ) - await self._send_error(msg, str(exc)) - except SlashSkillCommandResolutionError as exc: - logger.warning( - "Slash skill command resolution failed for %s (chat=%s): %s", - msg.channel_name, - msg.chat_id, - exc, - ) - await self._send_error(msg, str(exc)) - except Exception: - logger.exception( - "Error handling message from %s (chat=%s)", - msg.channel_name, - msg.chat_id, - ) - await self._send_error(msg, "An internal error occurred. Please try again.") + await self._handle_chat(msg, bound_identity_checked=True) + except InvalidChannelSessionConfigError as exc: + logger.warning( + "Invalid channel session config for %s (chat=%s): %s", + msg.channel_name, + msg.chat_id, + exc, + ) + await self._send_error(msg, str(exc)) + except SlashSkillCommandResolutionError as exc: + logger.warning( + "Slash skill command resolution failed for %s (chat=%s): %s", + msg.channel_name, + msg.chat_id, + exc, + ) + await self._send_error(msg, str(exc)) + except Exception: + logger.exception( + "Error handling message from %s (chat=%s)", + msg.channel_name, + msg.chat_id, + ) + await self._send_error(msg, "An internal error occurred. Please try again.") # -- chat handling ----------------------------------------------------- + async def _get_bound_identity_rejection(self, msg: InboundMessage) -> _BoundIdentityRejection | None: + """Return None when *msg* may proceed; otherwise return rejection routing hints. + + The returned object means the message lacks a verified bound identity. + Its fields are intentionally limited to server-side values re-read from + the connection repository, so rejection outbounds never trust a rejected + inbound message's asserted connection metadata. + """ + if not self._require_bound_identity: + return None + if _auth_disabled_owner_user_id(): + return None + + has_connection = bool(msg.connection_id) + has_owner = bool(msg.owner_user_id) + if not (has_connection and has_owner): + return _BoundIdentityRejection() + if self._connection_repo is None: + return _BoundIdentityRejection(message=BOUND_IDENTITY_UNAVAILABLE_MESSAGE) + + # The manager is the run-creation security boundary, so it does not + # trust mutable InboundMessage identity fields by themselves. Re-read + # the binding by provider identity before creating DeerFlow threads or + # runs. If the asserted identity does not match, keep only the + # server-side connection fields as outbound routing hints. + connection = await self._connection_repo.find_connection_by_external_identity( + provider=msg.channel_name, + external_account_id=msg.user_id, + workspace_id=msg.workspace_id or None, + ) + if connection is None: + return _BoundIdentityRejection() + + connection_id = connection.get("id") + owner_user_id = connection.get("owner_user_id") + if connection_id == msg.connection_id and owner_user_id == msg.owner_user_id: + return None + return _BoundIdentityRejection(outbound_connection_id=connection_id, outbound_owner_user_id=owner_user_id) + + async def _reject_unbound_channel_message( + self, + msg: InboundMessage, + *, + bound_identity_rejection: _BoundIdentityRejection, + ) -> None: + logger.info( + "[Manager] rejecting unbound channel message: channel=%s, chat_id=%s", + msg.channel_name, + msg.chat_id, + ) + outbound = OutboundMessage( + channel_name=msg.channel_name, + chat_id=msg.chat_id, + thread_id="", + text=bound_identity_rejection.message, + thread_ts=msg.thread_ts, + connection_id=bound_identity_rejection.outbound_connection_id, + owner_user_id=bound_identity_rejection.outbound_owner_user_id, + metadata=_slim_metadata(msg.metadata), + ) + await self.bus.publish_outbound(outbound) + async def _lookup_thread_id(self, msg: InboundMessage) -> str | None: if msg.connection_id and self._connection_repo is not None: return await self._connection_repo.get_thread_id( @@ -1016,7 +1106,21 @@ class ChannelManager: self._channel_metadata_synced.clear() self._channel_metadata_synced.add(thread_id) - async def _handle_chat(self, msg: InboundMessage, extra_context: dict[str, Any] | None = None) -> None: + async def _handle_chat( + self, + msg: InboundMessage, + extra_context: dict[str, Any] | None = None, + *, + bound_identity_checked: bool = False, + ) -> None: + # Normal entry paths already run the bound-identity check in + # _handle_message() or _handle_command(). Keep this default False so + # direct callers and future internal paths still fail closed. + bound_identity_rejection = None if bound_identity_checked else await self._get_bound_identity_rejection(msg) + if bound_identity_rejection is not None: + await self._reject_unbound_channel_message(msg, bound_identity_rejection=bound_identity_rejection) + return + client = self._get_client() # Look up existing DeerFlow thread. @@ -1242,6 +1346,18 @@ class ChannelManager: # -- command handling -------------------------------------------------- async def _handle_command(self, msg: InboundMessage) -> None: + # Commands are the other run-creation entry point besides chat: /new + # calls _create_thread() directly, and /bootstrap routes into + # _handle_chat(). Apply the same bound-identity admission boundary here + # so unbound platform users cannot create unowned threads/checkpoints or + # query Gateway state via commands. Provider-level binding flows + # (/connect , /start ) are consumed by the provider adapter + # before the message reaches the manager, so they are unaffected. + bound_identity_rejection = await self._get_bound_identity_rejection(msg) + if bound_identity_rejection is not None: + await self._reject_unbound_channel_message(msg, bound_identity_rejection=bound_identity_rejection) + return + raw_text = msg.text text = raw_text.strip() parts = text.split(maxsplit=1) @@ -1260,7 +1376,7 @@ class ChannelManager: chat_text = parts[1] if len(parts) > 1 else "Initialize workspace" chat_msg = _dc_replace(msg, text=chat_text, msg_type=InboundMessageType.CHAT) - await self._handle_chat(chat_msg, extra_context={"is_bootstrap": True}) + await self._handle_chat(chat_msg, extra_context={"is_bootstrap": True}, bound_identity_checked=True) return if reply is None and command == "new": @@ -1300,7 +1416,7 @@ class ChannelManager: from dataclasses import replace as _dc_replace chat_msg = _dc_replace(msg, msg_type=InboundMessageType.CHAT) - await self._handle_chat(chat_msg) + await self._handle_chat(chat_msg, bound_identity_checked=True) return else: reply = _unknown_command_reply(command) diff --git a/backend/app/channels/service.py b/backend/app/channels/service.py index a222c5a01..b689c9651 100644 --- a/backend/app/channels/service.py +++ b/backend/app/channels/service.py @@ -17,6 +17,7 @@ logger = logging.getLogger(__name__) if TYPE_CHECKING: from deerflow.config.app_config import AppConfig + from deerflow.config.channel_connections_config import ChannelConnectionsConfig # Channel name → import path for lazy loading _CHANNEL_REGISTRY: dict[str, str] = { @@ -64,8 +65,7 @@ def _merge_channel_connection_runtime_config(channels_config: dict[str, Any], ap merge_runtime_channel_configs(channels_config, connection_config) -def _make_connection_repo(app_config: AppConfig): - connection_config = getattr(app_config, "channel_connections", None) +def _make_connection_repo(connection_config: ChannelConnectionsConfig | None): if connection_config is None or not getattr(connection_config, "enabled", False): return None @@ -90,7 +90,13 @@ class ChannelService: instantiates enabled channels, and starts the ChannelManager dispatcher. """ - def __init__(self, channels_config: dict[str, Any] | None = None, *, connection_repo: Any | None = None) -> None: + def __init__( + self, + channels_config: dict[str, Any] | None = None, + *, + connection_repo: Any | None = None, + require_bound_identity: bool = False, + ) -> None: self.bus = MessageBus() self.store = ChannelStore() self._connection_repo = connection_repo @@ -107,6 +113,7 @@ class ChannelService: default_session=default_session if isinstance(default_session, dict) else None, channel_sessions=channel_sessions, connection_repo=connection_repo, + require_bound_identity=require_bound_identity, ) self._channels: dict[str, Any] = {} # name -> Channel instance self._config = config @@ -126,7 +133,14 @@ class ChannelService: if "channels" in extra: channels_config = dict(extra["channels"] or {}) _merge_channel_connection_runtime_config(channels_config, app_config) - return cls(channels_config=channels_config, connection_repo=_make_connection_repo(app_config)) + connection_config = getattr(app_config, "channel_connections", None) + connections_enabled = connection_config is not None and getattr(connection_config, "enabled", False) + require_bound_identity = bool(connections_enabled and getattr(connection_config, "require_bound_identity", True)) + return cls( + channels_config=channels_config, + connection_repo=_make_connection_repo(connection_config), + require_bound_identity=require_bound_identity, + ) async def start(self) -> None: """Start the manager and all enabled channels.""" diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 996c83568..8ed2e85a1 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -48,6 +48,11 @@ Then enable user bindings in `channel_connections`: ```yaml channel_connections: enabled: true + # Auth-enabled deployments require ordinary IM messages to come from a + # connected DeerFlow user by default. Set this to false only for legacy + # operator-owned/open-bot deployments that intentionally route unbound + # platform users to platform-ID user buckets. + require_bound_identity: true telegram: enabled: true @@ -74,6 +79,10 @@ channel_connections: `channel_connections` does not duplicate provider secrets. It only controls the browser-facing connect UI and stores per-user binding records. Telegram needs `bot_username` only so the frontend can open a deep link. +When `channel_connections.enabled` and `require_bound_identity` are true, auth-enabled deployments reject ordinary unbound IM messages before creating a DeerFlow thread or run. Users must connect the channel from DeerFlow Settings first. Auth-disabled local mode still routes channel messages to the auth-disabled default user, and legacy open-bot behavior can be restored explicitly with `require_bound_identity: false`. + +Upgrade note: existing auth-enabled deployments that already have `channel_connections.enabled: true` will start rejecting ordinary unbound IM messages after this field is introduced because `require_bound_identity` defaults to true. Legacy operator-owned/open-bot deployments that intentionally allow unbound platform users to create DeerFlow runs should set `require_bound_identity: false` before upgrading and restart the service. + ## Connect Flow Telegram: diff --git a/backend/packages/harness/deerflow/config/channel_connections_config.py b/backend/packages/harness/deerflow/config/channel_connections_config.py index 4092d5863..7e740b7af 100644 --- a/backend/packages/harness/deerflow/config/channel_connections_config.py +++ b/backend/packages/harness/deerflow/config/channel_connections_config.py @@ -42,6 +42,7 @@ class ChannelConnectionsConfig(BaseModel): """Top-level config for browser-connectable IM channels.""" enabled: bool = False + require_bound_identity: bool = True slack: SlackChannelConnectionConfig = Field(default_factory=SlackChannelConnectionConfig) telegram: TelegramChannelConnectionConfig = Field(default_factory=TelegramChannelConnectionConfig) discord: DiscordChannelConnectionConfig = Field(default_factory=DiscordChannelConnectionConfig) diff --git a/backend/tests/test_channel_connections_config.py b/backend/tests/test_channel_connections_config.py index 8a14878c0..48038abb1 100644 --- a/backend/tests/test_channel_connections_config.py +++ b/backend/tests/test_channel_connections_config.py @@ -7,6 +7,7 @@ def test_channel_connections_disabled_by_default(): config = ChannelConnectionsConfig() assert config.enabled is False + assert config.require_bound_identity is True assert config.slack.enabled is False assert config.telegram.enabled is False assert config.discord.enabled is False @@ -43,6 +44,13 @@ def test_enabled_channel_connections_do_not_require_public_url_or_encryption_key assert config.provider_status("wecom") == {"enabled": True, "configured": True} +def test_require_bound_identity_can_be_disabled_for_legacy_open_bot_mode(): + config = ChannelConnectionsConfig.model_validate({"enabled": True, "require_bound_identity": False}) + + assert config.enabled is True + assert config.require_bound_identity is False + + def test_provider_status_reports_disabled_and_unknown_providers(): config = ChannelConnectionsConfig.model_validate({"enabled": True}) diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index b4eea74ea..59f1539c1 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -2565,6 +2565,445 @@ class TestResolveRunParamsUserId: assert "channel_user_id" not in run_context +class _BoundIdentityRepo: + def __init__(self, connections: list[dict[str, str | None]] | None = None) -> None: + self.connections = list(connections or []) + self.lookups: list[dict[str, str | None]] = [] + self.thread_sets: list[dict[str, str | None]] = [] + + async def find_connection_by_external_identity(self, *, provider: str, external_account_id: str, workspace_id: str | None = None): + self.lookups.append( + { + "provider": provider, + "external_account_id": external_account_id, + "workspace_id": workspace_id, + } + ) + for connection in self.connections: + if connection.get("provider") == provider and connection.get("external_account_id") == external_account_id and connection.get("workspace_id") == workspace_id: + return connection + return None + + async def get_thread_id(self, connection_id: str, chat_id: str, topic_id: str | None = None): + return None + + async def set_thread_id( + self, + *, + connection_id: str, + owner_user_id: str, + provider: str, + external_conversation_id: str, + external_topic_id: str | None, + thread_id: str, + ) -> None: + self.thread_sets.append( + { + "connection_id": connection_id, + "owner_user_id": owner_user_id, + "provider": provider, + "external_conversation_id": external_conversation_id, + "external_topic_id": external_topic_id, + "thread_id": thread_id, + } + ) + + +class TestChannelManagerBoundIdentityPolicy: + def test_unbound_auth_enabled_chat_is_rejected_before_thread_or_run_creation(self, monkeypatch): + from app.channels.manager import BOUND_IDENTITY_REQUIRED_MESSAGE, ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store, require_bound_identity=True) + mock_client = _make_mock_langgraph_client() + manager._client = mock_client + outbound_received = [] + + async def capture(msg): + outbound_received.append(msg) + + bus.subscribe_outbound(capture) + await manager._handle_chat( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + text="hi", + thread_ts="1710000000.000100", + ) + ) + + assert len(outbound_received) == 1 + assert outbound_received[0].text == BOUND_IDENTITY_REQUIRED_MESSAGE + assert outbound_received[0].thread_id == "" + assert outbound_received[0].connection_id is None + assert outbound_received[0].owner_user_id is None + mock_client.threads.create.assert_not_called() + mock_client.runs.wait.assert_not_called() + + _run(go()) + + def test_bound_identity_repo_unavailable_uses_transient_failure_message(self, monkeypatch): + from app.channels.manager import BOUND_IDENTITY_UNAVAILABLE_MESSAGE, ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store, require_bound_identity=True) + mock_client = _make_mock_langgraph_client() + manager._client = mock_client + outbound_received = [] + + async def capture(msg): + outbound_received.append(msg) + + bus.subscribe_outbound(capture) + await manager._handle_chat( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + owner_user_id="deerflow-user-1", + connection_id="connection-1", + workspace_id="T123", + text="hi", + ) + ) + + assert len(outbound_received) == 1 + assert outbound_received[0].text == BOUND_IDENTITY_UNAVAILABLE_MESSAGE + assert outbound_received[0].connection_id is None + assert outbound_received[0].owner_user_id is None + mock_client.threads.create.assert_not_called() + mock_client.runs.wait.assert_not_called() + + _run(go()) + + def test_unbound_auth_enabled_chat_is_rejected_before_semaphore(self, monkeypatch): + from app.channels.manager import BOUND_IDENTITY_REQUIRED_MESSAGE, ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store, require_bound_identity=True) + outbound_received = [] + + async def capture(msg): + outbound_received.append(msg) + + bus.subscribe_outbound(capture) + await manager.start() + assert manager._semaphore is not None + await manager._semaphore.acquire() + try: + await asyncio.wait_for( + manager._handle_message( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + text="hi", + ) + ), + timeout=0.5, + ) + finally: + manager._semaphore.release() + await manager.stop() + + assert len(outbound_received) == 1 + assert outbound_received[0].text == BOUND_IDENTITY_REQUIRED_MESSAGE + assert outbound_received[0].connection_id is None + assert outbound_received[0].owner_user_id is None + + _run(go()) + + def test_bound_auth_enabled_chat_is_allowed_when_bound_identity_is_required(self, monkeypatch): + from app.channels.manager import ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + repo = _BoundIdentityRepo( + [ + { + "id": "connection-1", + "owner_user_id": "deerflow-user-1", + "provider": "slack", + "external_account_id": "U-platform", + "workspace_id": "T123", + } + ] + ) + manager = ChannelManager(bus=bus, store=store, connection_repo=repo, require_bound_identity=True) + mock_client = _make_mock_langgraph_client(thread_id="thread-bound") + manager._client = mock_client + + await manager._handle_chat( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + owner_user_id="deerflow-user-1", + connection_id="connection-1", + workspace_id="T123", + text="hi", + ) + ) + + mock_client.threads.create.assert_called_once() + mock_client.runs.wait.assert_called_once() + run_context = mock_client.runs.wait.call_args.kwargs["context"] + assert run_context["user_id"] == "deerflow-user-1" + assert run_context["channel_user_id"] == "U-platform" + + _run(go()) + + def test_bound_auth_enabled_message_checks_bound_identity_once_on_hot_path(self, monkeypatch): + from app.channels.manager import ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + repo = _BoundIdentityRepo( + [ + { + "id": "connection-1", + "owner_user_id": "deerflow-user-1", + "provider": "slack", + "external_account_id": "U-platform", + "workspace_id": "T123", + } + ] + ) + manager = ChannelManager(bus=bus, store=store, connection_repo=repo, require_bound_identity=True) + mock_client = _make_mock_langgraph_client(thread_id="thread-bound") + manager._client = mock_client + await manager.start() + try: + await manager._handle_message( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + owner_user_id="deerflow-user-1", + connection_id="connection-1", + workspace_id="T123", + text="hi", + ) + ) + finally: + await manager.stop() + + assert repo.lookups == [ + { + "provider": "slack", + "external_account_id": "U-platform", + "workspace_id": "T123", + } + ] + mock_client.threads.create.assert_called_once() + mock_client.runs.wait.assert_called_once() + + _run(go()) + + def test_auth_enabled_chat_rejects_unverified_bound_identity(self, monkeypatch): + from app.channels.manager import BOUND_IDENTITY_REQUIRED_MESSAGE, ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + repo = _BoundIdentityRepo( + [ + { + "id": "actual-connection", + "owner_user_id": "actual-owner", + "provider": "slack", + "external_account_id": "U-platform", + "workspace_id": None, + } + ] + ) + manager = ChannelManager(bus=bus, store=store, connection_repo=repo, require_bound_identity=True) + mock_client = _make_mock_langgraph_client() + manager._client = mock_client + outbound_received = [] + + async def capture(msg): + outbound_received.append(msg) + + bus.subscribe_outbound(capture) + await manager._handle_chat( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + owner_user_id="forged-owner", + connection_id="forged-connection", + text="hi", + ) + ) + + assert len(outbound_received) == 1 + assert outbound_received[0].text == BOUND_IDENTITY_REQUIRED_MESSAGE + assert outbound_received[0].connection_id == "actual-connection" + assert outbound_received[0].owner_user_id == "actual-owner" + mock_client.threads.create.assert_not_called() + mock_client.runs.wait.assert_not_called() + + _run(go()) + + def test_auth_disabled_chat_keeps_default_user_when_bound_identity_is_required(self, monkeypatch): + from app.channels.manager import ChannelManager + from app.gateway.auth_disabled import AUTH_DISABLED_USER_ID + + monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1") + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store, require_bound_identity=True) + mock_client = _make_mock_langgraph_client(thread_id="thread-local") + manager._client = mock_client + + await manager._handle_chat( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + text="hi", + ) + ) + + mock_client.threads.create.assert_called_once() + mock_client.runs.wait.assert_called_once() + run_context = mock_client.runs.wait.call_args.kwargs["context"] + assert run_context["user_id"] == AUTH_DISABLED_USER_ID + assert run_context["channel_user_id"] == "U-platform" + + _run(go()) + + def test_legacy_open_bot_mode_allows_unbound_auth_enabled_chat(self, monkeypatch): + from app.channels.manager import ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store, require_bound_identity=False) + mock_client = _make_mock_langgraph_client(thread_id="thread-legacy") + manager._client = mock_client + + await manager._handle_chat( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + text="hi", + ) + ) + + mock_client.threads.create.assert_called_once() + mock_client.runs.wait.assert_called_once() + run_context = mock_client.runs.wait.call_args.kwargs["context"] + assert run_context["user_id"] == "U-platform" + assert run_context["channel_user_id"] == "U-platform" + + _run(go()) + + def test_unbound_auth_enabled_new_command_is_rejected_before_thread_creation(self, monkeypatch): + from app.channels.manager import BOUND_IDENTITY_REQUIRED_MESSAGE, ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store, require_bound_identity=True) + mock_client = _make_mock_langgraph_client() + manager._client = mock_client + outbound_received = [] + + async def capture(msg): + outbound_received.append(msg) + + bus.subscribe_outbound(capture) + await manager._handle_command( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + text="/new", + msg_type=InboundMessageType.COMMAND, + thread_ts="1710000000.000100", + ) + ) + + assert len(outbound_received) == 1 + assert outbound_received[0].text == BOUND_IDENTITY_REQUIRED_MESSAGE + assert outbound_received[0].thread_id == "" + assert outbound_received[0].connection_id is None + assert outbound_received[0].owner_user_id is None + mock_client.threads.create.assert_not_called() + + _run(go()) + + def test_bound_auth_enabled_new_command_creates_thread(self, monkeypatch): + from app.channels.manager import ChannelManager + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + repo = _BoundIdentityRepo( + [ + { + "id": "connection-1", + "owner_user_id": "deerflow-user-1", + "provider": "slack", + "external_account_id": "U-platform", + "workspace_id": "T123", + } + ] + ) + manager = ChannelManager(bus=bus, store=store, connection_repo=repo, require_bound_identity=True) + mock_client = _make_mock_langgraph_client(thread_id="thread-bound") + manager._client = mock_client + + await manager._handle_command( + InboundMessage( + channel_name="slack", + chat_id="C123", + user_id="U-platform", + owner_user_id="deerflow-user-1", + connection_id="connection-1", + workspace_id="T123", + text="/new", + msg_type=InboundMessageType.COMMAND, + ) + ) + + mock_client.threads.create.assert_called_once() + + _run(go()) + + class TestChannelManagerConnectionRouting: def test_connection_scoped_conversations_do_not_share_threads(self, tmp_path, monkeypatch): from app.channels.manager import ChannelManager @@ -3811,6 +4250,13 @@ class TestChannelService: assert service.manager._connection_repo is repo + def test_require_bound_identity_is_forwarded_to_manager(self): + from app.channels.service import ChannelService + + service = ChannelService(channels_config={}, require_bound_identity=True) + + assert service.manager._require_bound_identity is True + def test_remove_channel_stops_running_channel_and_forgets_config(self): from app.channels.service import ChannelService diff --git a/config.example.yaml b/config.example.yaml index d2affc98a..9ac6e177d 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -15,7 +15,7 @@ # ============================================================================ # Bump this number when the config schema changes. # Run `make config-upgrade` to merge new fields into your local config.yaml. -config_version: 13 +config_version: 14 # ============================================================================ # Logging @@ -1208,6 +1208,9 @@ run_events: # # channel_connections: # enabled: false +# # Security: keep this enabled unless you intentionally want legacy open-bot behavior. +# # Disabling it lets unbound external IM users create DeerFlow threads/runs. +# require_bound_identity: true # # telegram: # enabled: false