diff --git a/backend/AGENTS.md b/backend/AGENTS.md index b5bddca79..876889104 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -446,7 +446,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk **Message Flow**: 1. External platform -> Channel impl -> `MessageBus.publish_inbound()` 2. `ChannelManager._dispatch_loop()` consumes from queue -3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id` +3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`. The Gateway forwards `channel_user_id` from `body.context` into the runtime context only (never `configurable`, which is checkpointed), and `bash_tool` exposes it to sandbox commands as the fixed env var `DEERFLOW_CHANNEL_USER_ID` — via a shell-quoted command-string prefix, NOT the `execute_command(env=...)` channel, which is reserved for request-scoped secrets and would switch `AioSandbox` onto the `bash.exec` path (image >= 1.9.3, fresh session per call). Per-call injection keeps group-chat identity correct (one thread/sandbox, many senders) **without depending on the AIO shell's session semantics**: every IM-channel command carries an explicit `export VAR=; ` (valid id) or `unset VAR; ` (empty / non-str / over the 256-char cap, since `body.context` is client-writable). The AIO no-env path reuses a persistent shell session (the reason for the class lock, #1433), so a bare command could otherwise resolve a stale id an earlier sender exported; the `unset` closes the window the length/type guard would open (a dropped id would inherit the previous sender's value). Non-IM runs (no `channel_user_id` in context) are left untouched. Not injected on the Windows local sandbox (its PowerShell/cmd.exe fallback has no `export`/`unset`). Propagates across `task` delegation: `task_tool` captures the dispatching turn's id and the subagent executor forwards it into the subagent's runtime context, same as the guardrail attribution fields. The var is informational, never authorization-grade: any bash command can overwrite it (and web clients can set `body.context.channel_user_id`), so skills must not treat it as authenticated identity. Tests: `tests/test_channel_user_id_env.py` 4. For chat: look up/create thread through Gateway's LangGraph-compatible API 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`) 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 1eb5e0860..277de1eb3 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -194,6 +194,11 @@ def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, An runtime_context.setdefault(key, context[key]) if "user_id" in context and isinstance(runtime_context, dict): runtime_context.setdefault("user_id", context["user_id"]) + # The raw platform user id from IM channels (Feishu open_id, Slack Uxxx, ...) + # follows the same runtime-context-only rule as user_id: tools may read it, + # but it never enters ``configurable`` (checkpointed with the thread). + if "channel_user_id" in context and isinstance(runtime_context, dict): + runtime_context.setdefault("channel_user_id", context["channel_user_id"]) def inject_authenticated_user_context(config: dict[str, Any], request: Request) -> None: diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py index dcff44a00..0dcdebbcc 100644 --- a/backend/packages/harness/deerflow/sandbox/tools.py +++ b/backend/packages/harness/deerflow/sandbox/tools.py @@ -1544,6 +1544,54 @@ def _truncate_ls_output(output: str, max_chars: int) -> str: return f"{output[:kept]}{marker}" +# Fixed env var exposing the IM-channel platform user id (Feishu open_id, +# Slack Uxxx, ...) to sandbox commands, so skills can act on the current end +# user's channel identity (#3914). An identifier, not a secret. +CHANNEL_USER_ID_ENV = "DEERFLOW_CHANNEL_USER_ID" + +_CHANNEL_USER_ID_CONTEXT_KEY = "channel_user_id" + +# body.context is client-writable on web requests, so bound the value: real +# platform ids are tens of chars; anything past this is hostile or corrupt and +# must not bloat every command string sent to the sandbox. +_CHANNEL_USER_ID_MAX_LEN = 256 + + +def _is_windows() -> bool: + return os.name == "nt" + + +def _channel_identity_prefix(runtime: Runtime) -> str | None: + """Build the command prefix that sets or clears the channel-user-id env var. + + Returns ``None`` for a non-IM run (no ``channel_user_id`` key in context) so + the command is left untouched. For an IM run the prefix is always emitted: + + - valid id (non-empty str within the length cap) → ``export VAR=; `` + - unusable id (empty / non-str / over the cap) → ``unset VAR; `` + + The id deliberately rides the command string instead of the + ``execute_command(env=...)`` channel: a non-empty ``env`` switches + ``AioSandbox`` to the ``bash.exec`` API (fresh session per call, image + >= 1.9.3 required), which is reserved for request-scoped secrets. Emitting an + explicit ``export``-or-``unset`` on every IM command makes per-call identity + correct **without depending on the AIO shell's session semantics**: the AIO + no-env path reuses a persistent shell session (the reason for the class lock, + #1433), so a bare command could otherwise resolve a stale value exported by + an earlier sender in a shared group-chat sandbox. The ``unset`` closes the + window the length/type guard would otherwise open — a sender whose id is + dropped inherits the previous sender's value. Values are identifiers, not + secrets, so keeping them in the audit-visible command string is fine. + """ + context = getattr(runtime, "context", None) + if not isinstance(context, dict) or _CHANNEL_USER_ID_CONTEXT_KEY not in context: + return None + channel_user_id = context.get(_CHANNEL_USER_ID_CONTEXT_KEY) + if isinstance(channel_user_id, str) and 0 < len(channel_user_id) <= _CHANNEL_USER_ID_MAX_LEN: + return f"export {CHANNEL_USER_ID_ENV}={shlex.quote(channel_user_id)}; " + return f"unset {CHANNEL_USER_ID_ENV}; " + + @tool("bash", parse_docstring=True) def bash_tool(runtime: Runtime, description: str, command: str) -> str: """Execute a bash command in a Linux environment. @@ -1566,6 +1614,7 @@ def bash_tool(runtime: Runtime, description: str, command: str) -> str: # Request-scoped secrets resolved for the active skill (#3861); injected as # per-call env into the subprocess, never placed in the command string. injected_env = read_active_secrets(getattr(runtime, "context", None)) or None + identity_prefix = _channel_identity_prefix(runtime) if is_local_sandbox(runtime): if not is_host_bash_allowed(): return f"Error: {LOCAL_HOST_BASH_DISABLED_MESSAGE}" @@ -1574,6 +1623,10 @@ def bash_tool(runtime: Runtime, description: str, command: str) -> str: validate_local_bash_command_paths(command, thread_data) command = replace_virtual_paths_in_command(command, thread_data) command = _apply_cwd_prefix(command, thread_data) + # POSIX-only: the Windows local sandbox may execute via + # PowerShell/cmd.exe where `export` is not valid syntax. + if identity_prefix and not _is_windows(): + command = identity_prefix + command try: from deerflow.config.app_config import get_app_config @@ -1589,6 +1642,8 @@ def bash_tool(runtime: Runtime, description: str, command: str) -> str: max_chars, ) ensure_thread_directories_exist(runtime) + if identity_prefix: + command = identity_prefix + command try: from deerflow.config.app_config import get_app_config diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 1a85f610e..a1f3dbd3a 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -295,6 +295,7 @@ class SubagentExecutor: oauth_provider: str | None = None, oauth_id: str | None = None, run_id: str | None = None, + channel_user_id: str | None = None, deerflow_trace_id: str | None = None, ): """Initialize the executor. @@ -342,6 +343,10 @@ class SubagentExecutor: self.oauth_provider = oauth_provider self.oauth_id = oauth_id self.run_id = run_id + # IM-channel sender identity captured at task_tool dispatch: group + # chats share one thread across senders, so delegated bash commands + # must export the dispatching turn's id, not none at all. + self.channel_user_id = channel_user_id self.deerflow_trace_id = deerflow_trace_id self._base_tools = _filter_tools( @@ -604,6 +609,8 @@ class SubagentExecutor: context["oauth_provider"] = self.oauth_provider context["oauth_id"] = self.oauth_id context["run_id"] = self.run_id + if self.channel_user_id: + context["channel_user_id"] = self.channel_user_id if self.deerflow_trace_id: context[DEERFLOW_TRACE_METADATA_KEY] = self.deerflow_trace_id context["is_subagent"] = True diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 88bac1d99..7b94f269d 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -327,6 +327,9 @@ async def task_tool( oauth_provider = parent_context.get("oauth_provider") oauth_id = parent_context.get("oauth_id") run_id = parent_context.get("run_id") + # IM-channel sender identity: group chats share one thread across senders, + # so delegated bash commands need the dispatching turn's channel_user_id. + channel_user_id = parent_context.get("channel_user_id") deerflow_trace_id = normalize_trace_id(parent_context.get(DEERFLOW_TRACE_METADATA_KEY)) or normalize_trace_id(metadata.get(DEERFLOW_TRACE_METADATA_KEY)) or get_current_trace_id() parent_available_skills = metadata.get("available_skills") @@ -371,6 +374,7 @@ async def task_tool( "oauth_provider": oauth_provider, "oauth_id": oauth_id, "run_id": run_id, + "channel_user_id": channel_user_id, "deerflow_trace_id": deerflow_trace_id, } if resolved_app_config is not None: diff --git a/backend/tests/test_channel_user_id_env.py b/backend/tests/test_channel_user_id_env.py new file mode 100644 index 000000000..a2de0dd90 --- /dev/null +++ b/backend/tests/test_channel_user_id_env.py @@ -0,0 +1,200 @@ +"""Tests for exposing the IM-channel platform user id to sandbox commands (#3914). + +Two halves: +- Gateway: ``merge_run_context_overrides`` forwards ``channel_user_id`` from + ``body.context`` into ``config['context']`` (runtime context) only — never + into ``configurable`` (which is checkpointed). +- Sandbox: ``bash_tool`` exposes the id as the fixed env var + ``DEERFLOW_CHANNEL_USER_ID`` via an ``export`` prefix on the command string. + It must NOT ride the ``env=`` parameter: on ``AioSandbox`` a non-empty env + switches execution to the ``bash.exec`` API, which requires image >= 1.9.3 + and abandons the persistent shell session — that channel is reserved for + request-scoped secrets. +""" + +from types import SimpleNamespace + +from deerflow.sandbox.tools import ( + CHANNEL_USER_ID_ENV, + _channel_identity_prefix, + bash_tool, +) + +_THREAD_DATA = { + "workspace_path": "/tmp/deer-flow/threads/t1/user-data/workspace", + "uploads_path": "/tmp/deer-flow/threads/t1/user-data/uploads", + "outputs_path": "/tmp/deer-flow/threads/t1/user-data/outputs", +} + + +def _aio_runtime(context: dict) -> SimpleNamespace: + return SimpleNamespace( + state={"sandbox": {"sandbox_id": "aio-sandbox-1"}, "thread_data": _THREAD_DATA.copy()}, + context=context, + ) + + +class _CapturingSandbox: + def __init__(self, output: str = "ok"): + self.calls: list[dict] = [] + self._output = output + + def execute_command(self, command: str, env=None, timeout=None) -> str: + self.calls.append({"command": command, "env": env}) + return self._output + + +def _run_bash(monkeypatch, runtime, command: str = "echo hi") -> _CapturingSandbox: + sandbox = _CapturingSandbox() + monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox) + monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None) + bash_tool.func(runtime=runtime, description="test", command=command) + return sandbox + + +class TestMergeRunContextOverridesChannelUserId: + def test_channel_user_id_propagates_to_runtime_context_only(self): + from app.gateway.services import build_run_config, merge_run_context_overrides + + config = build_run_config("thread-1", None, None) + merge_run_context_overrides(config, {"channel_user_id": "ou_feishu_123"}) + + assert config["context"]["channel_user_id"] == "ou_feishu_123" + # Never into configurable: that mapping is checkpointed with the thread. + assert "channel_user_id" not in config["configurable"] + + def test_existing_runtime_context_value_wins(self): + """setdefault semantics: a server-side value stamped earlier must not be + overridden by the client-supplied body.context.""" + from app.gateway.services import build_run_config, merge_run_context_overrides + + config = build_run_config("thread-1", None, None) + config.setdefault("context", {})["channel_user_id"] = "server-stamped" + merge_run_context_overrides(config, {"channel_user_id": "client-supplied"}) + + assert config["context"]["channel_user_id"] == "server-stamped" + + def test_absent_channel_user_id_adds_nothing(self): + from app.gateway.services import build_run_config, merge_run_context_overrides + + config = build_run_config("thread-1", None, None) + merge_run_context_overrides(config, {"model_name": "gpt"}) + + assert "channel_user_id" not in config.get("context", {}) + + +class TestBashToolChannelIdentityPrefix: + def test_identity_exported_and_env_stays_none(self, monkeypatch): + """The id rides the command string; env must stay None so AioSandbox + keeps the legacy persistent-shell path (regression guard for the + #3921/#3922 bash.exec capability gap).""" + sandbox = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "ou_feishu_123"})) + + assert len(sandbox.calls) == 1 + assert sandbox.calls[0]["command"] == f"export {CHANNEL_USER_ID_ENV}=ou_feishu_123; echo hi" + assert sandbox.calls[0]["env"] is None + + def test_no_channel_user_id_leaves_command_unchanged(self, monkeypatch): + sandbox = _run_bash(monkeypatch, _aio_runtime({"thread_id": "t1"})) + + assert sandbox.calls[0]["command"] == "echo hi" + assert sandbox.calls[0]["env"] is None + + def test_per_call_identity_follows_current_context(self, monkeypatch): + """Group chats share one thread/sandbox: each message's run carries that + sender's id, so consecutive commands must each export their own value.""" + first = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "sender-a"})) + second = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "sender-b"})) + + assert "sender-a" in first.calls[0]["command"] + assert "sender-b" in second.calls[0]["command"] + + def test_value_is_shell_quoted(self, monkeypatch): + """A hostile platform id must not be able to inject shell syntax.""" + sandbox = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "x'; rm -rf /tmp/y; '"})) + + command = sandbox.calls[0]["command"] + assert command.endswith("; echo hi") + # shlex.quote wraps the value; the raw injection payload must not appear + # as executable syntax outside the quoted region. + assert "export " + CHANNEL_USER_ID_ENV + "='x'\"'\"'; rm -rf /tmp/y; '\"'\"''; echo hi" == command + + def test_secrets_and_identity_compose(self, monkeypatch): + """Active skill secrets keep the env= channel; the identity keeps the + command-string channel. They must not mix.""" + runtime = _aio_runtime( + { + "channel_user_id": "ou_1", + "__active_skill_secrets": {"ERP_TOKEN": "secret-value"}, + } + ) + sandbox = _run_bash(monkeypatch, runtime) + + call = sandbox.calls[0] + assert call["env"] == {"ERP_TOKEN": "secret-value"} + assert call["command"].startswith(f"export {CHANNEL_USER_ID_ENV}=ou_1; ") + assert "secret-value" not in call["command"] + + def test_non_im_run_leaves_command_untouched(self): + """No channel_user_id key at all → non-IM run → prefix is None so the + command (the vast majority: Web/API/subagent) is unchanged.""" + assert _channel_identity_prefix(SimpleNamespace(context={"thread_id": "t1"})) is None + assert _channel_identity_prefix(SimpleNamespace(context={})) is None + assert _channel_identity_prefix(SimpleNamespace(context=None)) is None + + def test_unusable_value_emits_unset_not_none(self, monkeypatch): + """An IM run whose id is unusable (empty / non-str / over the cap) must + emit ``unset`` — not skip the prefix. Skipping would let a bare command + resolve a stale value left in the AIO persistent shell by an earlier + sender (willem-bd's group-chat leak window).""" + for bad in ("", 123, "x" * 5000, None): + prefix = _channel_identity_prefix(SimpleNamespace(context={"channel_user_id": bad})) + assert prefix == f"unset {CHANNEL_USER_ID_ENV}; ", f"value={bad!r}" + + def test_group_chat_dropped_id_clears_previous_sender(self, monkeypatch): + """Sender A (valid) then sender B (over-cap id, dropped): B's command must + carry ``unset`` so it cannot inherit A's exported id in a shared + persistent-shell sandbox — per-call correctness independent of session + persistence.""" + a = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "sender-a"})) + b = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "b" * 5000})) + + assert a.calls[0]["command"] == f"export {CHANNEL_USER_ID_ENV}=sender-a; echo hi" + assert b.calls[0]["command"] == f"unset {CHANNEL_USER_ID_ENV}; echo hi" + assert b.calls[0]["env"] is None + + def test_windows_local_sandbox_skips_prefix(self, monkeypatch): + """On Windows the local sandbox may execute via PowerShell/cmd.exe where + POSIX ``export`` is not valid syntax — skip injection rather than break + every IM-channel command.""" + runtime = SimpleNamespace( + state={"sandbox": {"sandbox_id": "local"}, "thread_data": _THREAD_DATA.copy()}, + context={"channel_user_id": "ou_1", "thread_id": "t1"}, + ) + sandbox = _CapturingSandbox() + monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox) + monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None) + monkeypatch.setattr("deerflow.sandbox.tools.is_host_bash_allowed", lambda: True) + monkeypatch.setattr("deerflow.sandbox.tools._is_windows", lambda: True) + + bash_tool.func(runtime=runtime, description="test", command="echo hi") + + assert len(sandbox.calls) == 1 + assert "export" not in sandbox.calls[0]["command"] + + def test_posix_local_sandbox_gets_prefix(self, monkeypatch): + runtime = SimpleNamespace( + state={"sandbox": {"sandbox_id": "local"}, "thread_data": _THREAD_DATA.copy()}, + context={"channel_user_id": "ou_1", "thread_id": "t1"}, + ) + sandbox = _CapturingSandbox() + monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox) + monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None) + monkeypatch.setattr("deerflow.sandbox.tools.is_host_bash_allowed", lambda: True) + + bash_tool.func(runtime=runtime, description="test", command="echo hi") + + assert len(sandbox.calls) == 1 + command = sandbox.calls[0]["command"] + assert command.startswith(f"export {CHANNEL_USER_ID_ENV}=ou_1; ") + assert command.endswith("echo hi") diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 31e04d3a3..1473ff42e 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -2417,6 +2417,43 @@ class TestSubagentGuardrailAttribution: assert context.get("run_id") == "run-42" assert context.get("is_subagent") is True + @pytest.mark.anyio + async def test_aexecute_propagates_channel_user_id_to_subagent_context( + self, + classes, + executor_module, + monkeypatch, + ): + """The IM-channel sender identity captured at task_tool must reach the + subagent's ``astream`` context so delegated bash commands export the + dispatching turn's ``DEERFLOW_CHANNEL_USER_ID`` (group chats share one + thread across senders).""" + SubagentExecutor = classes["SubagentExecutor"] + SubagentConfig = classes["SubagentConfig"] + executor = SubagentExecutor( + config=SubagentConfig( + name="general-purpose", + description="Channel identity test agent", + system_prompt="You are a channel identity test agent.", + max_turns=5, + timeout_seconds=30, + ), + tools=[], + parent_model="test-model", + thread_id="thread-channel-1", + trace_id="trace-channel-1", + channel_user_id="ou_group_sender_1", + ) + fake_agent = _FakeStreamAgent() + monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state) + monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent) + + await executor._aexecute("do something") + + context = fake_agent.captured_context + assert context is not None + assert context.get("channel_user_id") == "ou_group_sender_1" + @pytest.mark.anyio async def test_aexecute_context_defaults_to_none_when_attribution_absent( self, diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index fef64e4d3..831119347 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -186,6 +186,47 @@ def test_task_tool_returns_error_for_unknown_subagent(monkeypatch): assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == "Unknown subagent type 'general-purpose'. Available: general-purpose" +def test_task_tool_forwards_channel_user_id_to_executor(monkeypatch): + """The IM-channel sender identity must survive delegation: in group chats + one thread serves many senders, so a subagent's bash commands need the + dispatching turn's channel_user_id (same propagation rule as user_role / + oauth attribution).""" + runtime = _make_runtime() + runtime.context["channel_user_id"] = "ou_group_sender_1" + captured = {} + + class DummyExecutor: + def __init__(self, **kwargs): + captured["executor_kwargs"] = kwargs + + def execute_async(self, prompt, task_id=None): + return task_id or "generated-task-id" + + monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus) + monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config()) + monkeypatch.setattr( + task_tool_module, + "get_background_task_result", + lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"), + ) + monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _event: None) + monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep) + monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: []) + + output = _run_task_tool( + runtime=runtime, + description="运行子任务", + prompt="collect diagnostics", + subagent_type="general-purpose", + tool_call_id="tc-channel-id", + ) + + message = _task_tool_message(output) + assert message.content == "Task Succeeded. Result: done" + assert captured["executor_kwargs"]["channel_user_id"] == "ou_group_sender_1" + + def test_task_tool_rejects_bash_subagent_when_host_bash_disabled(monkeypatch): monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config()) monkeypatch.setattr(task_tool_module, "is_host_bash_allowed", lambda: False)