diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 297ff360c..4e5de1c6f 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -729,7 +729,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply. - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle) - `service.py` - Manages lifecycle of all configured channels from `config.yaml` -- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, and edits the "Working on it..." stream target in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured) +- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, and edits the "Working on it..." stream target in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`) - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR - `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs - `deerflow.persistence.channel_connections` - SQL-backed user-owned connection, optional credential, connect state, and conversation store diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 3e4254d6b..9d6f5ecc8 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -17,6 +17,10 @@ from app.channels.base import Channel from app.channels.commands import is_known_channel_command from app.channels.connection_identity import attach_connection_identity from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment +from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.sandbox.sandbox_provider import get_sandbox_provider +from deerflow.uploads.manager import UnsafeUploadPathError, claim_unique_filename, normalize_filename, write_upload_file_no_symlink logger = logging.getLogger(__name__) @@ -29,6 +33,11 @@ _CONVERSATION_TYPE_GROUP = "2" _MAX_UPLOAD_SIZE_BYTES = 20 * 1024 * 1024 +# Inbound attachments are buffered in memory before being persisted and synced +# into the sandbox, so bound them. DingTalk chats accept far larger files than +# an agent can usefully read; oversized ones surface as a failed-load marker. +_MAX_INBOUND_FILE_SIZE_BYTES = 50 * 1024 * 1024 + def _normalize_conversation_type(raw: Any) -> str: """Normalize ``conversationType`` to ``"1"`` (P2P) or ``"2"`` (group). @@ -63,6 +72,16 @@ def _is_dingtalk_command(text: str) -> bool: return is_known_channel_command(text) +def _display_filename(filename: str) -> str: + """Collapse whitespace and cap length for embedding a filename in message text. + + ``fileName`` is webhook-supplied data: embedded raw in a failed-load marker, + a newline could forge a standalone ``/mnt/user-data/uploads/...`` line and an + over-long name would bloat the message text. + """ + return re.sub(r"\s+", " ", filename).strip()[:80] + + def _extract_text_from_rich_text(rich_text_list: list) -> str: parts: list[str] = [] for item in rich_text_list: @@ -137,6 +156,9 @@ class DingTalkChannel(Channel): self._incoming_messages: dict[str, Any] = {} self._incoming_messages_lock = threading.Lock() self._card_repliers: dict[str, Any] = {} + # Serialize inbound-file writes into the uploads directory to avoid + # racing writers clobbering one another (mirrors FeishuChannel). + self._file_write_lock = threading.Lock() @property def supports_streaming(self) -> bool: @@ -366,8 +388,9 @@ class DingTalkChannel(Channel): sender_nick = message.sender_nick or "" text = self._extract_text(message) - if not text: - logger.info("[DingTalk] empty text, ignoring message") + files = self._extract_files(message) + if not text and not files: + logger.info("[DingTalk] empty message with no files, ignoring") return connect_code = self._pending_connect_code(text) @@ -396,12 +419,13 @@ class DingTalkChannel(Channel): # INFO logs, and only after the allowed_users gate so blocked senders are # not logged at all. logger.info( - "[DingTalk] parsed message: conv_type=%s, msg_id=%s, sender=%s(%s), text_len=%d", + "[DingTalk] parsed message: conv_type=%s, msg_id=%s, sender=%s(%s), text_len=%d, files=%d", conversation_type, msg_id, sender_staff_id, sender_nick, len(text or ""), + len(files), ) if _is_dingtalk_command(text): @@ -435,6 +459,7 @@ class DingTalkChannel(Channel): text=text, msg_type=msg_type, thread_ts=msg_id, + files=files, metadata={ "conversation_type": conversation_type, "conversation_id": conversation_id, @@ -471,6 +496,237 @@ class DingTalkChannel(Channel): return _extract_text_from_rich_text(message.rich_text_content.rich_text_list).strip() return "" + # -- inbound: file attachments ----------------------------------------- + + @staticmethod + def _extract_files(message: Any) -> list[dict[str, Any]]: + """Extract inbound file/image descriptors from a DingTalk message. + + Returns a list of dicts shaped for ``InboundMessage.files``; each carries + ``type`` (``"image"`` or ``"file"``), ``download_code``, and ``filename``. + + Images arrive as ``picture`` messages (a single ``downloadCode``) or as + inline images inside a ``richText`` message. Documents arrive as ``file`` + messages, which ``dingtalk_stream.ChatbotMessage.from_dict`` does not + parse — the descriptor (``downloadCode`` / ``fileName``) is read from the + raw callback payload stashed on the message as ``_df_raw_data`` by + ``_DingTalkMessageHandler.process``. + """ + msg_type = getattr(message, "message_type", None) + files: list[dict[str, Any]] = [] + + if msg_type == "picture": + image_content = getattr(message, "image_content", None) + code = getattr(image_content, "download_code", None) + if code: + files.append({"type": "image", "download_code": code, "filename": "image.png"}) + elif msg_type == "richText": + try: + codes = message.get_image_list() or [] + except Exception: + # Log rather than swallow: without this, a richText message whose + # SDK image parse fails looks identical to one that simply has no + # inline images, which is hard to diagnose if the SDK shape changes. + logger.warning("[DingTalk] failed to read inline images from richText message", exc_info=True) + codes = [] + for idx, code in enumerate(codes): + if code: + files.append({"type": "image", "download_code": code, "filename": f"image_{idx}.png"}) + elif msg_type == "file": + raw = getattr(message, "_df_raw_data", None) + content = raw.get("content") if isinstance(raw, dict) else None + if isinstance(content, dict): + code = content.get("downloadCode") + if code: + filename = content.get("fileName") or "file.bin" + files.append({"type": "file", "download_code": code, "filename": filename}) + + return files + + async def receive_file(self, msg: InboundMessage, thread_id: str, *, user_id: str | None = None) -> InboundMessage: + """Download inbound DingTalk files into the thread uploads directory. + + Mirrors :meth:`FeishuChannel.receive_file`: each descriptor in + ``msg.files`` is downloaded by its ``downloadCode``, persisted under the + thread's uploads bucket, synced into a non-local sandbox, and its sandbox + virtual path is prepended to ``msg.text`` so downstream models can read + the file by path. Descriptors are then cleared so the generic + URL-based ``_ingest_inbound_files`` pass does not try to re-fetch them + (DingTalk files are downloaded by code, not by URL). + + An attachment that fails to download contributes a short + ``[failed to load ...]`` marker instead of a path, so the agent can tell + the user something was sent but could not be read rather than silently + ignoring it. + """ + if not msg.files: + return msg + + virtual_paths: list[str] = [] + failures: list[str] = [] + for f in msg.files: + if not isinstance(f, dict): + continue + download_code = f.get("download_code") + if not download_code: + continue + file_type = "image" if f.get("type") == "image" else "file" + filename = f.get("filename") if isinstance(f.get("filename"), str) else "" + # Per-attachment isolation: the manager awaits receive_file without a + # try, so anything escaping here would kill the chat turn with no + # reply — the silent-drop failure mode this feature exists to fix. + try: + virtual_path = await self._receive_single_file(download_code, file_type, filename, thread_id, user_id=user_id) + except Exception: + logger.exception("[DingTalk] unexpected error receiving inbound %s", file_type) + virtual_path = "" + if virtual_path: + virtual_paths.append(virtual_path) + else: + display = _display_filename(filename) + failures.append(f"[failed to load {file_type}: {display}]" if display else f"[failed to load {file_type}]") + + prefix_lines = virtual_paths + failures + if prefix_lines: + block = "\n".join(prefix_lines) + msg.text = f"{block}\n\n{msg.text}".strip() if msg.text else block + + msg.files = [] + return msg + + async def _receive_single_file( + self, + download_code: str, + file_type: str, + filename: str, + thread_id: str, + *, + user_id: str | None = None, + ) -> str: + """Download one file by ``downloadCode`` and persist it into uploads. + + Returns the sandbox virtual path on success, or ``""`` on failure. + """ + content = await self._download_by_code(download_code) + if not content: + return "" + + paths = get_paths() + effective_user_id = user_id or get_effective_user_id() + + default_ext = "png" if file_type == "image" else "bin" + # ``download_code`` is attacker-controllable webhook data, so restrict it to + # a safe character set before embedding it in the fallback name. This keeps + # the fallback safe by construction instead of relying on a later write + # failure to reject a name that escaped the uploads directory. + code_token = re.sub(r"[^A-Za-z0-9_-]", "", download_code)[-12:] or "attachment" + fallback_name = f"dingtalk_{code_token}.{default_ext}" + # normalize_filename strips directory components and rejects traversal + # patterns ("..", backslash paths, over-long names); fall back to the + # sanitized generated name if the platform-supplied filename is rejected. + try: + safe_filename = normalize_filename(filename or fallback_name) + except ValueError: + safe_filename = fallback_name + + def _persist() -> Path: + # Directory prep, the uniqueness claim, and the write are blocking + # filesystem IO — the whole sequence stays off the event loop. The + # claim and the write share one lock because generated names repeat + # across messages ("image.png" for every picture message): without a + # claim a later attachment silently overwrites an earlier one whose + # path was already handed to the agent, and letting the claim and + # write interleave would resolve two attachments to the same free name. + paths.ensure_thread_dirs(thread_id, user_id=effective_user_id) + uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve() + with self._file_write_lock: + seen = {entry.name for entry in uploads_dir.iterdir() if entry.is_file()} + unique_name = claim_unique_filename(safe_filename, seen) + # write_upload_file_no_symlink refuses a symlinked destination: + # uploads dirs can be mounted into local sandboxes, so a sandbox + # process could otherwise redirect this privileged write outside + # the bucket. + return write_upload_file_no_symlink(uploads_dir, unique_name, content) + + try: + resolved_target = await asyncio.to_thread(_persist) + except (OSError, UnsafeUploadPathError): + logger.exception("[DingTalk] failed to persist downloaded file: %s", safe_filename) + return "" + + virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}" + + try: + sandbox_provider = get_sandbox_provider() + # acquire_async keeps provider lifecycle work (Docker discovery, + # readiness polls) off the event loop; update_file is blocking + # transport IO on remote sandboxes, so it is offloaded too. + sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id) + if sandbox_id != "local": + sandbox = sandbox_provider.get(sandbox_id) + if sandbox is None: + # Mirror Feishu: the agent's non-local sandbox cannot see this + # file, so returning the virtual path would hand the model a + # path that reads as nothing — surface a failed-load marker. + logger.warning("[DingTalk] sandbox %s not found after acquire, dropping attachment: %s", sandbox_id, virtual_path) + return "" + await asyncio.to_thread(sandbox.update_file, virtual_path, content) + except Exception: + # Same failure mode as the sandbox-is-None branch: the bytes never + # reached the agent's sandbox, so the virtual path would read as + # nothing. Mirror Feishu and surface a failed-load marker. + logger.exception("[DingTalk] failed to sync downloaded file into non-local sandbox: %s", virtual_path) + return "" + + return virtual_path + + async def _download_by_code(self, download_code: str) -> bytes | None: + """Exchange a DingTalk ``downloadCode`` for the raw file bytes. + + Two steps per the DingTalk robot OpenAPI: + 1. ``POST /v1.0/robot/messageFiles/download`` -> ``{downloadUrl}`` + 2. ``GET downloadUrl`` -> binary content + """ + try: + # Token acquisition stays inside the try: the manager awaits + # receive_file without a try, so a token failure escaping here would + # abort the whole chat turn instead of degrading to a failed-load marker. + token = await self._get_access_token() + async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client: + response = await client.post( + f"{DINGTALK_API_BASE}/v1.0/robot/messageFiles/download", + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json={"downloadCode": download_code, "robotCode": self._client_id}, + ) + if response.status_code != 200: + logger.warning("[DingTalk] messageFiles/download failed: status=%d, body=%s", response.status_code, response.text[:300]) + return None + download_url = response.json().get("downloadUrl") + if not download_url: + logger.warning("[DingTalk] messageFiles/download returned no downloadUrl") + return None + + # Stream with a size cap: the bytes are buffered in memory before + # being persisted, so an oversized attachment must be refused + # before it is fully read, not after. + chunks: list[bytes] = [] + total = 0 + async with client.stream("GET", download_url, follow_redirects=True) as file_response: + file_response.raise_for_status() + async for chunk in file_response.aiter_bytes(): + total += len(chunk) + if total > _MAX_INBOUND_FILE_SIZE_BYTES: + logger.warning("[DingTalk] inbound file exceeds %d bytes, dropping", _MAX_INBOUND_FILE_SIZE_BYTES) + return None + chunks.append(chunk) + return b"".join(chunks) + except (httpx.HTTPError, ValueError): + logger.exception("[DingTalk] failed to download file by code") + return None + async def _prepare_inbound(self, chat_id: str, inbound: InboundMessage) -> None: inbound = await self._attach_connection_identity(inbound) # Running reply must finish before publish_inbound so AI card tracks are @@ -831,5 +1087,9 @@ class _DingTalkMessageHandler: import dingtalk_stream incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data) + # Stash the raw callback payload: dingtalk_stream does not parse ``file`` + # (document) messages, so DingTalkChannel._extract_files reads the file + # descriptor (downloadCode / fileName) from it. + incoming_message._df_raw_data = callback.data self._channel._on_chatbot_message(incoming_message) return dingtalk_stream.AckMessage.STATUS_OK, "OK" diff --git a/backend/tests/blocking_io/test_dingtalk_receive_file.py b/backend/tests/blocking_io/test_dingtalk_receive_file.py new file mode 100644 index 000000000..e9b00d37b --- /dev/null +++ b/backend/tests/blocking_io/test_dingtalk_receive_file.py @@ -0,0 +1,56 @@ +"""Regression anchor: DingTalk ``receive_file`` must not block the event loop. + +``_receive_single_file`` prepares the thread directories, resolves the uploads +dir, scans it for the uniqueness claim, and writes the attachment — all blocking +filesystem IO that must run inside ``asyncio.to_thread`` (and sandbox sync must +go through ``acquire_async`` + an offloaded ``update_file``). This anchor drives +the real ``receive_file`` under the strict Blockbuster gate; if any of that +regresses back onto the event loop, Blockbuster raises ``BlockingError``. + +The ``Paths`` construction is offloaded only because ``Paths.__init__`` resolves +paths synchronously; the surface under test (``receive_file``'s persist path) is +exercised on the event loop, not bypassed. The download itself is mocked — the +network leg is httpx-async and not the subject here. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +pytestmark = pytest.mark.asyncio + + +async def test_receive_file_persist_does_not_block_event_loop(tmp_path, monkeypatch) -> None: + from app.channels.dingtalk import DingTalkChannel + from app.channels.message_bus import MessageBus + from deerflow.config.paths import Paths + + paths = await asyncio.to_thread(Paths, str(tmp_path)) + monkeypatch.setattr("app.channels.dingtalk.get_paths", lambda: paths) + + async def _acquire_async(thread_id, user_id=None): + return "local" + + monkeypatch.setattr( + "app.channels.dingtalk.get_sandbox_provider", + lambda: SimpleNamespace(acquire_async=_acquire_async, get=lambda sid: None), + ) + + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"DATA") + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert "/uploads/a.pdf" in out.text + assert out.files == [] diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 6138d62dc..098edbb9d 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import logging from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -22,6 +23,7 @@ from app.channels.dingtalk import ( _normalize_conversation_type, ) from app.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage +from deerflow.config.paths import VIRTUAL_PATH_PREFIX def _run(coro): @@ -1691,3 +1693,841 @@ class TestCardMode: assert channel._card_track_ids == {} _run(go()) + + +# --------------------------------------------------------------------------- +# Inbound file support +# --------------------------------------------------------------------------- + + +def _make_picture_message(*, download_code: str = "dc_img", **kwargs): + """Build a mock DingTalk ``picture`` message carrying an ImageContent.""" + msg = _make_chatbot_message(text="", message_type="picture", **kwargs) + msg.image_content = SimpleNamespace(download_code=download_code) + return msg + + +def _make_file_message(*, download_code: str = "dc_file", file_name: str | None = "report.xlsx", **kwargs): + """Build a mock DingTalk ``file`` (document) message. + + dingtalk_stream does not parse ``file`` messages, so the descriptor lives on + the raw callback payload the handler stashes as ``_df_raw_data``. + """ + msg = _make_chatbot_message(text="", message_type="file", **kwargs) + content: dict = {"downloadCode": download_code} + if file_name is not None: + content["fileName"] = file_name + msg._df_raw_data = {"msgtype": "file", "content": content} + return msg + + +def _patch_uploads(monkeypatch, uploads_dir, *, sandbox_id="local", sandbox=None): + monkeypatch.setattr( + "app.channels.dingtalk.get_paths", + lambda: SimpleNamespace( + ensure_thread_dirs=lambda thread_id, user_id=None: None, + sandbox_uploads_dir=lambda thread_id, user_id=None: uploads_dir, + ), + ) + monkeypatch.setattr("app.channels.dingtalk.get_effective_user_id", lambda: "default") + + async def _acquire_async(thread_id, user_id=None): + return sandbox_id + + monkeypatch.setattr( + "app.channels.dingtalk.get_sandbox_provider", + lambda: SimpleNamespace(acquire_async=_acquire_async, get=lambda sid: sandbox), + ) + + +class TestExtractFiles: + def test_text_message_has_no_files(self): + assert DingTalkChannel._extract_files(_make_chatbot_message(text="hi")) == [] + + def test_picture_message(self): + files = DingTalkChannel._extract_files(_make_picture_message(download_code="dc_1")) + assert files == [{"type": "image", "download_code": "dc_1", "filename": "image.png"}] + + def test_picture_message_missing_download_code_ignored(self): + msg = _make_chatbot_message(text="", message_type="picture") + msg.image_content = SimpleNamespace(download_code=None) + assert DingTalkChannel._extract_files(msg) == [] + + def test_rich_text_images(self): + msg = _make_chatbot_message(text="", message_type="richText") + msg.get_image_list = lambda: ["dc_a", "dc_b"] + files = DingTalkChannel._extract_files(msg) + assert files == [ + {"type": "image", "download_code": "dc_a", "filename": "image_0.png"}, + {"type": "image", "download_code": "dc_b", "filename": "image_1.png"}, + ] + + def test_rich_text_get_image_list_failure_is_logged_not_silent(self, caplog): + """An SDK parse failure must be distinguishable from "no inline images".""" + msg = _make_chatbot_message(text="", message_type="richText") + + def _boom(): + raise RuntimeError("sdk failure") + + msg.get_image_list = _boom + with caplog.at_level(logging.WARNING, logger="app.channels.dingtalk"): + assert DingTalkChannel._extract_files(msg) == [] + + assert any("failed to read inline images" in r.message for r in caplog.records) + + def test_file_message_from_raw_data(self): + files = DingTalkChannel._extract_files(_make_file_message(download_code="dc_doc", file_name="报价.xlsx")) + assert files == [{"type": "file", "download_code": "dc_doc", "filename": "报价.xlsx"}] + + def test_file_message_default_filename(self): + files = DingTalkChannel._extract_files(_make_file_message(download_code="dc_doc", file_name=None)) + assert files == [{"type": "file", "download_code": "dc_doc", "filename": "file.bin"}] + + def test_file_message_without_raw_data_ignored(self): + msg = _make_chatbot_message(text="", message_type="file") + assert DingTalkChannel._extract_files(msg) == [] + + +class TestOnChatbotMessageFiles: + def test_picture_message_publishes_inbound_with_files(self): + async def go(): + bus = MessageBus() + bus.publish_inbound = AsyncMock() + channel = DingTalkChannel(bus, config={}) + channel._client_id = "test_key" + channel._main_loop = asyncio.get_event_loop() + channel._running = True + channel._send_running_reply = AsyncMock() + + channel._on_chatbot_message(_make_picture_message(download_code="dc_pic", sender_staff_id="u1", message_id="m1")) + await asyncio.sleep(0.1) + + bus.publish_inbound.assert_awaited_once() + inbound = bus.publish_inbound.await_args.args[0] + assert inbound.msg_type == InboundMessageType.CHAT + assert inbound.files == [{"type": "image", "download_code": "dc_pic", "filename": "image.png"}] + + _run(go()) + + def test_file_message_publishes_inbound_with_files(self): + async def go(): + bus = MessageBus() + bus.publish_inbound = AsyncMock() + channel = DingTalkChannel(bus, config={}) + channel._client_id = "test_key" + channel._main_loop = asyncio.get_event_loop() + channel._running = True + channel._send_running_reply = AsyncMock() + + channel._on_chatbot_message(_make_file_message(download_code="dc_doc", file_name="data.csv", sender_staff_id="u1", message_id="m1")) + await asyncio.sleep(0.1) + + bus.publish_inbound.assert_awaited_once() + inbound = bus.publish_inbound.await_args.args[0] + assert inbound.files == [{"type": "file", "download_code": "dc_doc", "filename": "data.csv"}] + + _run(go()) + + def test_message_without_text_or_files_dropped(self): + async def go(): + bus = MessageBus() + bus.publish_inbound = AsyncMock() + channel = DingTalkChannel(bus, config={}) + channel._client_id = "test_key" + channel._main_loop = asyncio.get_event_loop() + channel._running = True + + # picture message whose image carries no download_code -> no files, no text + msg = _make_chatbot_message(text="", message_type="picture") + msg.image_content = SimpleNamespace(download_code=None) + channel._on_chatbot_message(msg) + await asyncio.sleep(0.1) + + bus.publish_inbound.assert_not_awaited() + + _run(go()) + + +class TestReceiveFile: + def test_no_files_is_noop(self): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + msg = channel._make_inbound(chat_id="c", user_id="u", text="hi", thread_ts="m") + out = await channel.receive_file(msg, "t1") + assert out is msg + assert out.text == "hi" + + _run(go()) + + def test_downloads_persists_and_prepends_path(self, tmp_path, monkeypatch): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._client_id = "robot" + channel._download_by_code = AsyncMock(return_value=b"XLSXDATA") + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="please translate", + thread_ts="m", + files=[{"type": "file", "download_code": "dc1", "filename": "quote.xlsx"}], + ) + out = await channel.receive_file(msg, "thread-1", user_id="default") + + assert out.files == [] + assert (uploads / "quote.xlsx").read_bytes() == b"XLSXDATA" + assert f"{VIRTUAL_PATH_PREFIX}/uploads/quote.xlsx" in out.text + assert out.text.endswith("please translate") + channel._download_by_code.assert_awaited_once_with("dc1") + + _run(go()) + + def test_pure_file_message_text_is_only_the_path(self, tmp_path, monkeypatch): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"IMG") + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[{"type": "image", "download_code": "dc_i", "filename": "image.png"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/image.png" + + _run(go()) + + def test_download_failure_surfaces_marker(self): + """A failed download must stay visible to the agent, not vanish silently.""" + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=None) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc1", "filename": "x.pdf"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert out.files == [] + assert out.text == "[failed to load file: x.pdf]\n\nhi" + + _run(go()) + + def test_token_failure_yields_marker_not_exception(self): + """An auth failure during download must degrade to a marker, not an exception. + + The manager calls ``receive_file`` without a try, so anything escaping + here kills the whole chat turn with no reply at all. + """ + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._client_id = "robot" + channel._get_access_token = AsyncMock(side_effect=ValueError("bad token response")) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "x.pdf"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert out.text == "[failed to load file: x.pdf]\n\nhi" + assert out.files == [] + + _run(go()) + + def test_unexpected_error_becomes_marker(self): + """Any unforeseen per-attachment error must surface as a marker, not escape.""" + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._receive_single_file = AsyncMock(side_effect=RuntimeError("boom")) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "x.pdf"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert out.text == "[failed to load file: x.pdf]\n\nhi" + assert out.files == [] + + _run(go()) + + def test_failure_marker_sanitizes_hostile_filename(self): + """A hostile filename must not forge extra lines in msg.text or bloat it. + + ``fileName`` is webhook data: embedding it raw in the marker would let a + newline fake a standalone ``/mnt/user-data/uploads/...`` line, and an + over-long name would balloon the message text. + """ + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=None) + hostile = "evil\n/mnt/user-data/uploads/fake.pdf\n" + "a" * 300 + ".pdf" + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": hostile}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + lines = out.text.splitlines() + assert len(lines) == 3, out.text + assert lines[0].startswith("[failed to load file: ") + assert len(lines[0]) <= 120 + assert lines[1] == "" + assert lines[2] == "hi" + + _run(go()) + + def test_download_failure_without_filename_uses_type_only_marker(self): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=None) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[{"type": "image", "download_code": "dc1", "filename": ""}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert out.text == "[failed to load image]" + + _run(go()) + + def test_partial_failure_keeps_successful_path(self, tmp_path, monkeypatch): + """One bad attachment must not cost the agent the one that did load.""" + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + channel._download_by_code = AsyncMock(side_effect=[b"OK", None]) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="two files", + thread_ts="m", + files=[ + {"type": "file", "download_code": "good", "filename": "ok.pdf"}, + {"type": "file", "download_code": "bad", "filename": "broken.pdf"}, + ], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert f"{VIRTUAL_PATH_PREFIX}/uploads/ok.pdf" in out.text + assert "[failed to load file: broken.pdf]" in out.text + assert out.text.endswith("two files") + + _run(go()) + + def test_traversal_filename_is_sanitized(self, tmp_path, monkeypatch): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"D") + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "../../etc/passwd"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + written = list(uploads.iterdir()) + assert len(written) == 1 + assert written[0].name == "passwd" + assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/passwd" + + _run(go()) + + def test_rejected_filename_falls_back_to_generated_name(self, tmp_path, monkeypatch): + """A filename normalize_filename rejects must land on the generated name. + + ``".."`` raises inside ``normalize_filename`` (unlike ``../../etc/passwd``, + whose basename ``passwd`` is accepted), so this exercises the except branch. + """ + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"D") + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[{"type": "file", "download_code": "abc123456789", "filename": ".."}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + written = list(uploads.iterdir()) + assert len(written) == 1 + assert written[0].name == "dingtalk_abc123456789.bin" + assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/dingtalk_abc123456789.bin" + + _run(go()) + + def test_fallback_name_sanitizes_download_code(self, tmp_path, monkeypatch): + """download_code is webhook data: it must not inject path separators.""" + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"D") + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[{"type": "image", "download_code": "../../evil", "filename": ".."}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + written = list(uploads.iterdir()) + assert len(written) == 1 + assert "/" not in written[0].name + assert written[0].name == "dingtalk_evil.png" + assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/dingtalk_evil.png" + + _run(go()) + + def test_second_picture_does_not_overwrite_first(self, tmp_path, monkeypatch): + """Two picture messages both generate "image.png" and must not collide. + + The path handed to the agent must still hold the bytes it referred to. + """ + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + + channel._download_by_code = AsyncMock(return_value=b"FIRST") + first = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m1", + files=[{"type": "image", "download_code": "dc1", "filename": "image.png"}], + ) + out_first = await channel.receive_file(first, "t1", user_id="default") + + channel._download_by_code = AsyncMock(return_value=b"SECOND") + second = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m2", + files=[{"type": "image", "download_code": "dc2", "filename": "image.png"}], + ) + out_second = await channel.receive_file(second, "t1", user_id="default") + + assert out_first.text != out_second.text + first_name = out_first.text.rsplit("/", 1)[-1] + second_name = out_second.text.rsplit("/", 1)[-1] + # Each advertised path must still resolve to its own bytes. + assert (uploads / first_name).read_bytes() == b"FIRST" + assert (uploads / second_name).read_bytes() == b"SECOND" + + _run(go()) + + def test_multiple_images_in_one_message_get_distinct_paths(self, tmp_path, monkeypatch): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + channel._download_by_code = AsyncMock(side_effect=[b"A", b"B"]) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[ + {"type": "image", "download_code": "dc1", "filename": "image_0.png"}, + {"type": "image", "download_code": "dc2", "filename": "image_0.png"}, + ], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + paths = out.text.split("\n") + assert len(paths) == 2 + assert paths[0] != paths[1] + assert len(list(uploads.iterdir())) == 2 + assert {p.read_bytes() for p in uploads.iterdir()} == {b"A", b"B"} + + _run(go()) + + def test_duplicate_document_names_do_not_collide(self, tmp_path, monkeypatch): + """The same real filename sent twice must not clobber the earlier upload.""" + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + + channel._download_by_code = AsyncMock(return_value=b"V1") + m1 = channel._make_inbound(chat_id="c", user_id="u", text="", thread_ts="m1", files=[{"type": "file", "download_code": "d1", "filename": "quote.xlsx"}]) + await channel.receive_file(m1, "t1", user_id="default") + + channel._download_by_code = AsyncMock(return_value=b"V2") + m2 = channel._make_inbound(chat_id="c", user_id="u", text="", thread_ts="m2", files=[{"type": "file", "download_code": "d2", "filename": "quote.xlsx"}]) + await channel.receive_file(m2, "t1", user_id="default") + + assert (uploads / "quote.xlsx").read_bytes() == b"V1" + assert len(list(uploads.iterdir())) == 2 + + _run(go()) + + def test_write_does_not_follow_planted_symlink(self, tmp_path, monkeypatch): + """A symlink planted at the destination must not be written through. + + Upload dirs can be mounted into local sandboxes, so a sandbox process can + leave a symlink at a future upload name; following it would let a + gateway-privileged write land outside the bucket. + """ + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + uploads = tmp_path / "uploads" + uploads.mkdir() + outside = tmp_path / "outside.txt" + (uploads / "image.png").symlink_to(outside) + _patch_uploads(monkeypatch, uploads) + channel._download_by_code = AsyncMock(return_value=b"PWNED") + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[{"type": "image", "download_code": "dc", "filename": "image.png"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert not outside.exists() + assert "[failed to load image: image.png]" in out.text + + _run(go()) + + def test_missing_sandbox_after_acquire_yields_marker(self, tmp_path, monkeypatch): + """A non-local sandbox that cannot be resolved must not yield a path. + + The agent's sandbox cannot see the file in that case, so handing it the + virtual path would point at nothing readable; mirror Feishu and surface + a failed-load marker instead. + """ + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"BYTES") + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=None) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert out.text == "[failed to load file: a.pdf]\n\nhi" + + _run(go()) + + def test_update_file_failure_yields_marker(self, tmp_path, monkeypatch): + """A failed non-local sandbox sync must not yield a path either. + + Same failure mode as the missing-sandbox case: the bytes never reached + the agent's sandbox, so the virtual path would read as nothing. Mirrors + Feishu, whose sync except-branch returns the failure marker. + """ + + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"BYTES") + uploads = tmp_path / "uploads" + uploads.mkdir() + broken_sandbox = MagicMock() + broken_sandbox.update_file.side_effect = RuntimeError("sandbox transport down") + _patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=broken_sandbox) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}], + ) + out = await channel.receive_file(msg, "t1", user_id="default") + + assert out.text == "[failed to load file: a.pdf]\n\nhi" + + _run(go()) + + def test_non_local_sandbox_is_synced(self, tmp_path, monkeypatch): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"BYTES") + uploads = tmp_path / "uploads" + uploads.mkdir() + fake_sandbox = MagicMock() + _patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=fake_sandbox) + + msg = channel._make_inbound( + chat_id="c", + user_id="u", + text="", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}], + ) + await channel.receive_file(msg, "t1", user_id="default") + + fake_sandbox.update_file.assert_called_once_with(f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf", b"BYTES") + + _run(go()) + + +class TestDownloadByCode: + def test_two_step_download(self): + async def go(): + from unittest.mock import patch + + channel = DingTalkChannel(MessageBus(), config={}) + channel._client_id = "robot_x" + channel._get_access_token = AsyncMock(return_value="tok") + + class PostResponse: + status_code = 200 + + @staticmethod + def json(): + return {"downloadUrl": "https://dl.dingtalk/xyz"} + + captured: dict = {} + + class FakeStream: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + pass + + def raise_for_status(self): + pass + + async def aiter_bytes(self): + yield b"FILE" + yield b"BYTES" + + class FakeClient: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + pass + + async def post(self, url, **kwargs): + captured["post_url"] = url + captured["json"] = kwargs.get("json") + return PostResponse() + + def stream(self, method, url, **kwargs): + captured["get_url"] = url + return FakeStream() + + with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()): + data = await channel._download_by_code("dl_code") + + assert data == b"FILEBYTES" + assert captured["post_url"].endswith("/v1.0/robot/messageFiles/download") + assert captured["json"] == {"downloadCode": "dl_code", "robotCode": "robot_x"} + assert captured["get_url"] == "https://dl.dingtalk/xyz" + + _run(go()) + + def test_oversized_download_is_dropped(self, monkeypatch): + """Inbound bytes are buffered in memory; a file over the cap must be refused. + + Outbound uploads already enforce a size cap — without an inbound one, a + single large chat attachment balloons gateway memory. + """ + + async def go(): + from unittest.mock import patch + + channel = DingTalkChannel(MessageBus(), config={}) + channel._client_id = "robot_x" + channel._get_access_token = AsyncMock(return_value="tok") + monkeypatch.setattr("app.channels.dingtalk._MAX_INBOUND_FILE_SIZE_BYTES", 10) + + class PostResponse: + status_code = 200 + + @staticmethod + def json(): + return {"downloadUrl": "https://dl.dingtalk/big"} + + class FakeStream: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + pass + + def raise_for_status(self): + pass + + async def aiter_bytes(self): + for _ in range(4): + yield b"x" * 8 + + class FakeClient: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + pass + + async def post(self, url, **kwargs): + return PostResponse() + + def stream(self, method, url, **kwargs): + return FakeStream() + + with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()): + assert await channel._download_by_code("dl") is None + + _run(go()) + + def test_post_non_200_returns_none(self): + async def go(): + from unittest.mock import patch + + channel = DingTalkChannel(MessageBus(), config={}) + channel._client_id = "robot_x" + channel._get_access_token = AsyncMock(return_value="tok") + + class PostResponse: + status_code = 403 + text = "forbidden" + + class FakeClient: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + pass + + async def post(self, url, **kwargs): + return PostResponse() + + with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()): + assert await channel._download_by_code("dl") is None + + _run(go()) + + def test_missing_download_url_returns_none(self): + async def go(): + from unittest.mock import patch + + channel = DingTalkChannel(MessageBus(), config={}) + channel._client_id = "robot_x" + channel._get_access_token = AsyncMock(return_value="tok") + + class PostResponse: + status_code = 200 + + @staticmethod + def json(): + return {} + + class FakeClient: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + pass + + async def post(self, url, **kwargs): + return PostResponse() + + with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()): + assert await channel._download_by_code("dl") is None + + _run(go()) + + +class TestHandlerStashesRawData: + def test_process_stashes_raw_callback_payload(self): + pytest.importorskip("dingtalk_stream") + + async def go(): + bus = MessageBus() + channel = DingTalkChannel(bus, config={}) + captured: dict = {} + channel._on_chatbot_message = lambda m: captured.update(msg=m) + handler = _DingTalkMessageHandler(channel) + cb = MagicMock() + cb.data = { + "msgtype": "file", + "content": {"downloadCode": "dc_doc", "fileName": "a.xlsx"}, + "senderStaffId": "u1", + "conversationType": "1", + "msgId": "m1", + } + await handler.process(cb) + + msg = captured["msg"] + assert getattr(msg, "_df_raw_data", None) == cb.data + # _extract_files can now read the document descriptor from the stash. + assert DingTalkChannel._extract_files(msg) == [{"type": "file", "download_code": "dc_doc", "filename": "a.xlsx"}] + + _run(go())