From a0d26e446cf6c3b08b7b3880784aa47b259f680b Mon Sep 17 00:00:00 2001 From: hetaoBackend Date: Thu, 6 Aug 2026 10:22:27 +0800 Subject: [PATCH] fix: make inbound attachments collision-safe --- backend/app/channels/dingtalk.py | 28 +++------- backend/app/channels/manager.py | 28 +++++----- .../tests/blocking_io/test_channels_ingest.py | 7 ++- .../blocking_io/test_dingtalk_receive_file.py | 11 ++-- .../tests/test_channel_file_attachments.py | 51 +++++++++++++++++-- backend/tests/test_dingtalk_channel.py | 31 ++++++++++- 6 files changed, 106 insertions(+), 50 deletions(-) diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 9d6f5ecc8..5a98811b6 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -17,10 +17,11 @@ 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.config.paths import 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 +from deerflow.uploads.layout import upload_virtual_path +from deerflow.uploads.manager import UnsafeUploadPathError, normalize_filename, publish_upload_bytes logger = logging.getLogger(__name__) @@ -156,9 +157,6 @@ 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: @@ -630,23 +628,11 @@ class DingTalkChannel(Channel): 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. + # Directory prep and collision-safe publication are blocking + # filesystem IO, so the whole sequence stays off the event loop. 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) + return publish_upload_bytes(uploads_dir, safe_filename, content) try: resolved_target = await asyncio.to_thread(_persist) @@ -654,7 +640,7 @@ class DingTalkChannel(Channel): logger.exception("[DingTalk] failed to persist downloaded file: %s", safe_filename) return "" - virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}" + virtual_path = upload_virtual_path(resolved_target.name) try: sandbox_provider = get_sandbox_provider() diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index d7982006f..c18661066 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -882,22 +882,20 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: if not msg.files: return [] + from deerflow.uploads.layout import upload_virtual_path from deerflow.uploads.manager import ( UnsafeUploadPathError, - claim_unique_filename, ensure_uploads_dir, normalize_filename, - write_upload_file_no_symlink, + publish_upload_bytes, ) - def _prepare_uploads_dir() -> tuple[Path, set[str]]: - # Worker thread: ensure_uploads_dir's mkdir and the iterdir enumeration are - # blocking filesystem IO that must stay off the event loop. - target = ensure_uploads_dir(thread_id, user_id=user_id) - existing = {entry.name for entry in target.iterdir() if entry.is_file()} - return target, existing + def _prepare_uploads_dir() -> Path: + # Worker thread: directory creation is blocking filesystem IO that must + # stay off the event loop. + return ensure_uploads_dir(thread_id, user_id=user_id) - uploads_dir, seen_names = await asyncio.to_thread(_prepare_uploads_dir) + uploads_dir = await asyncio.to_thread(_prepare_uploads_dir) created: list[dict[str, Any]] = [] file_reader = INBOUND_FILE_READERS.get(msg.channel_name, _read_http_inbound_file) @@ -940,7 +938,7 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: filename = f"{msg.thread_ts or 'msg'}_{idx}{ext}" try: - safe_name = claim_unique_filename(normalize_filename(filename), seen_names) + safe_name = normalize_filename(filename) except ValueError: logger.warning( "[Manager] skipping inbound file with unsafe filename: channel=%s, file=%r", @@ -949,21 +947,21 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: ) continue - dest = uploads_dir / safe_name try: - dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data) + dest = await asyncio.to_thread(publish_upload_bytes, uploads_dir, safe_name, data) except UnsafeUploadPathError: logger.warning("[Manager] skipping inbound file with unsafe destination: %s", safe_name) continue except Exception: - logger.exception("[Manager] failed to write inbound file: %s", dest) + logger.exception("[Manager] failed to write inbound file: %s", safe_name) continue + actual_name = dest.name created.append( { - "filename": safe_name, + "filename": actual_name, "size": len(data), - "path": f"/mnt/user-data/uploads/{safe_name}", + "path": upload_virtual_path(actual_name), "is_image": ftype == "image", } ) diff --git a/backend/tests/blocking_io/test_channels_ingest.py b/backend/tests/blocking_io/test_channels_ingest.py index 7b16c7084..7f35c374c 100644 --- a/backend/tests/blocking_io/test_channels_ingest.py +++ b/backend/tests/blocking_io/test_channels_ingest.py @@ -1,10 +1,9 @@ """Regression anchor: ingesting inbound channel files must not block the event loop. ``ChannelManager``'s ``_ingest_inbound_files`` ensures the thread uploads -directory (``mkdir``), enumerates it (``iterdir`` / ``is_file``) to de-duplicate -filenames, and writes each downloaded attachment to disk -(``write_upload_file_no_symlink``) — all blocking filesystem IO. The async -function offloads the directory prep and every per-file write via +directory (``mkdir``), stages each downloaded attachment, and publishes it +without replacing an existing name — all blocking filesystem IO. The async +function offloads the directory prep and every per-file publication via ``asyncio.to_thread`` while keeping the genuinely async network read (``file_reader``) on the loop. If any of that regresses back onto the event loop, the strict Blockbuster gate raises ``BlockingError`` and this test fails. diff --git a/backend/tests/blocking_io/test_dingtalk_receive_file.py b/backend/tests/blocking_io/test_dingtalk_receive_file.py index e9b00d37b..8e08ec5ab 100644 --- a/backend/tests/blocking_io/test_dingtalk_receive_file.py +++ b/backend/tests/blocking_io/test_dingtalk_receive_file.py @@ -1,11 +1,12 @@ """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``. +dir, stages the attachment, and publishes it without replacing an existing name +— 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 diff --git a/backend/tests/test_channel_file_attachments.py b/backend/tests/test_channel_file_attachments.py index 0b8e27496..62440c922 100644 --- a/backend/tests/test_channel_file_attachments.py +++ b/backend/tests/test_channel_file_attachments.py @@ -282,7 +282,7 @@ class TestInboundFileIngestion: assert (uploads_dir / "report.pdf").read_bytes() == b"pdf bytes" assert "_content" not in msg.files[0] - def test_rejects_preexisting_symlink_destination(self, tmp_path): + def test_renames_around_preexisting_symlink_destination(self, tmp_path): from app.channels import manager uploads_dir = tmp_path / "uploads" @@ -307,11 +307,19 @@ class TestInboundFileIngestion: ): result = _run(manager._ingest_inbound_files("thread-1", msg)) - assert result == [] + assert result == [ + { + "filename": "victim_1.txt", + "size": len(b"attacker data"), + "path": "/mnt/user-data/uploads/victim_1.txt", + "is_image": False, + } + ] assert not outside_file.exists() assert (uploads_dir / "victim.txt").is_symlink() + assert (uploads_dir / "victim_1.txt").read_bytes() == b"attacker data" - def test_rejects_dangling_symlink_destination(self, tmp_path): + def test_renames_around_dangling_symlink_destination(self, tmp_path): from app.channels import manager uploads_dir = tmp_path / "uploads" @@ -336,9 +344,17 @@ class TestInboundFileIngestion: ): result = _run(manager._ingest_inbound_files("thread-1", msg)) - assert result == [] + assert result == [ + { + "filename": "victim_1.txt", + "size": len(b"attacker data"), + "path": "/mnt/user-data/uploads/victim_1.txt", + "is_image": False, + } + ] assert not missing_target.exists() assert (uploads_dir / "victim.txt").is_symlink() + assert (uploads_dir / "victim_1.txt").read_bytes() == b"attacker data" def test_hardlinked_existing_file_is_not_overwritten(self, tmp_path): from app.channels import manager @@ -378,6 +394,33 @@ class TestInboundFileIngestion: assert (uploads_dir / "victim.txt").read_text(encoding="utf-8") == "protected" assert (uploads_dir / "victim_1.txt").read_bytes() == b"new attachment data" + def test_concurrent_inbound_messages_with_same_name_preserve_all_bytes(self, tmp_path): + from app.channels import manager + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + payloads = [f"payload-{index}".encode() for index in range(8)] + messages = [ + InboundMessage( + channel_name="telegram", + chat_id="chat-1", + user_id="user-1", + text="attachment", + files=[{"type": "file", "filename": "same.txt", "_content": payload}], + ) + for payload in payloads + ] + + async def run_all(): + return await asyncio.gather(*(manager._ingest_inbound_files("thread-1", message) for message in messages)) + + with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir): + results = _run(run_all()) + + created = [batch[0] for batch in results] + assert len({item["filename"] for item in created}) == len(payloads) + assert {(uploads_dir / item["filename"]).read_bytes() for item in created} == set(payloads) + # --------------------------------------------------------------------------- # Channel base class _on_outbound with attachments diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 098edbb9d..e05ccfe9f 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -2223,6 +2223,33 @@ class TestReceiveFile: _run(go()) + def test_concurrent_duplicate_document_names_preserve_all_bytes(self, tmp_path, monkeypatch): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + uploads = tmp_path / "uploads" + uploads.mkdir() + _patch_uploads(monkeypatch, uploads) + payloads = [f"version-{index}".encode() for index in range(8)] + channel._download_by_code = AsyncMock(side_effect=payloads) + + paths = await asyncio.gather( + *( + channel._receive_single_file( + f"code-{index}", + "file", + "quote.xlsx", + "thread-1", + user_id="default", + ) + for index in range(len(payloads)) + ) + ) + + assert len(set(paths)) == len(payloads) + assert {(uploads / path.rsplit("/", 1)[-1]).read_bytes() for path in paths} == set(payloads) + + _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. @@ -2250,7 +2277,9 @@ class TestReceiveFile: out = await channel.receive_file(msg, "t1", user_id="default") assert not outside.exists() - assert "[failed to load image: image.png]" in out.text + assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/image_1.png" + assert (uploads / "image.png").is_symlink() + assert (uploads / "image_1.png").read_bytes() == b"PWNED" _run(go())