diff --git a/README.md b/README.md index 6f856d4782..12a96774a7 100644 --- a/README.md +++ b/README.md @@ -867,6 +867,11 @@ DeerFlow can also expose user-owned IM channel connections in the workspace UI. | DingTalk | Stream Push (WebSocket) | Moderate | | Buzz | Nostr relay (WebSocket, NIP-42) | Moderate | +Attachments saved by the shared IM ingestion pipeline or Feishu/DingTalk's +embedded downloads keep distinct filenames, including when concurrent uploads +choose the same name. The final filename is passed to the agent and used for +sandbox sync; an existing conversation file is not overwritten. + **Configuration in `config.yaml`:** Discord's `channels.discord.allowed_guilds` accepts one positive numeric guild diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 4ee0f40cba..fcb36c8258 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -45,7 +45,7 @@ Cursor writes drain before cancellation; token expiry drains cursor/auth clearin **Owner-scoped file storage**: inbound files, uploads, and output artifacts are staged under the DeerFlow owner's bucket so they land where the agent run reads/writes (`users/{user_id}/threads/{thread_id}/user-data/{uploads,outputs}`). `ChannelManager._handle_chat` resolves the storage owner once via `_channel_storage_user_id(msg)` (sanitized owner id, falling back to `safe(msg.user_id)` for unbound auth-enabled channels — mirroring `_resolve_run_params`'s run identity; `None` only when no identity is available) and threads it as the `user_id=` kwarg through the file pipeline: - `Channel.receive_file(msg, thread_id, user_id=...)` — owner-bound channels persist downloaded files under the owner's bucket instead of the default bucket -- `FeishuChannel._receive_single_file(...)` / `DingTalkChannel._receive_single_file(...)` — normalize provider filenames, claim a collision-free basename and write it through `write_upload_file_no_symlink` under the same channel lock; the returned basename drives both the agent-visible virtual path and non-local sandbox sync +- `FeishuChannel._receive_single_file(...)` / `DingTalkChannel._receive_single_file(...)` — normalize filenames and claim/write through `write_upload_file_no_symlink(..., exclusive=True)`, retrying `FileExistsError` under the channel lock; the final basename drives agent-visible paths and sandbox sync - `sandbox_files.py` — non-mounted Feishu/DingTalk syncs acquire unique execution holders, drain blocking `update_file` workers across repeated cancellation, and release only after the last sandbox operation, so a parallel run cannot close the shared client mid-upload. Feishu requests provider release after the final holder exits, parking standalone temporary sandboxes; DingTalk retains its non-releasing holder behavior. - `_ingest_inbound_files(...)` and the underlying `ensure_uploads_dir` / `get_uploads_dir` — owner-scoped via the same kwarg - `_resolve_attachments` / `_prepare_artifact_delivery` — resolve output artifacts from the bound owner's bucket through `app.gateway.path_utils.resolve_outputs_confined_path`, the same outputs-only rule the artifact editor uses, so a sibling `uploads/`/`workspace/` path or a symlink planted in `outputs/` is skipped with a warning diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 016f8b0878..8373e8e302 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -672,22 +672,19 @@ class DingTalkChannel(Channel): 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. + # filesystem IO — the whole sequence stays off the event loop. + # Exclusive creation also protects against publishers outside this + # channel instance's lock. 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()} - 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) + while True: + unique_name = claim_unique_filename(safe_filename, seen) + try: + return write_upload_file_no_symlink(uploads_dir, unique_name, content, exclusive=True) + except FileExistsError: + continue try: resolved_target = await asyncio.to_thread(_persist) diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 0b5c2bce47..ea6f155069 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -495,8 +495,12 @@ class FeishuChannel(Channel): uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve() with self._thread_lock: seen = {entry.name for entry in uploads_dir.iterdir()} - unique_name = claim_unique_filename(safe_filename, seen) - return write_upload_file_no_symlink(uploads_dir, unique_name, content) + while True: + unique_name = claim_unique_filename(safe_filename, seen) + try: + return write_upload_file_no_symlink(uploads_dir, unique_name, content, exclusive=True) + except FileExistsError: + continue try: resolved_target = await asyncio.to_thread(_persist) diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 787c227b77..cbeff25fbe 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -1188,7 +1188,14 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: dest = uploads_dir / safe_name try: - dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data) + while True: + try: + dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data, exclusive=True) + break + except FileExistsError: + # Another upload can claim a name after the directory + # snapshot, including while this attachment downloads. + safe_name = claim_unique_filename(safe_name, seen_names) # Root-written 0o600 files are unreadable to the non-root # sandbox; grant group/other read like the HTTP upload path. await asyncio.to_thread(_make_inbound_file_sandbox_readable, dest) diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index fe9ff227ee..7ce5fa6831 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -163,8 +163,8 @@ def validate_path_traversal(path: Path, base: Path) -> None: raise PathTraversalError("Path traversal detected") from None -def validate_upload_destination(base_dir: Path, filename: str) -> Path: - """Validate an upload destination without mutating an existing file.""" +def validate_upload_destination(base_dir: Path, filename: str, *, exclusive: bool = False) -> Path: + """Validate a destination; exclusive creation treats regular hardlinks as collisions.""" safe_name = normalize_filename(filename) dest = base_dir / safe_name @@ -175,7 +175,7 @@ def validate_upload_destination(base_dir: Path, filename: str) -> Path: if st is not None and not stat.S_ISREG(st.st_mode): raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}") - if st is not None and st.st_nlink > 1: + if not exclusive and st is not None and st.st_nlink > 1: raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}") validate_path_traversal(dest, base_dir) @@ -248,7 +248,7 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None, *, mi return removed -def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, object]: +def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: bool = False) -> tuple[Path, object]: """Open an upload destination for safe streaming writes. Upload directories may be mounted into local sandboxes. A sandbox process can @@ -259,9 +259,13 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob and ``fstat`` validation after ``open()`` to reduce the TOCTOU window; this does not eliminate all races but makes exploitation significantly harder. Path-traversal validation prevents escapes from *base_dir* in both cases. + + ``exclusive=True`` atomically claims a new filename. An existing regular + file raises ``FileExistsError`` without truncation so the caller can retry + under another name. The default retains replacement semantics. """ safe_name = normalize_filename(filename) - dest = validate_upload_destination(base_dir, safe_name) + dest = validate_upload_destination(base_dir, safe_name, exclusive=exclusive) try: st = os.lstat(dest) except FileNotFoundError: @@ -272,6 +276,8 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob if has_nofollow: # POSIX: O_NOFOLLOW makes open() fail with ELOOP if dest is a symlink. flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW + if exclusive: + flags |= os.O_EXCL if hasattr(os, "O_NONBLOCK"): flags |= os.O_NONBLOCK @@ -299,10 +305,12 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob # Note: a narrow race window remains between the pre-open lstat and open(); the # path-traversal check mitigates escapes from base_dir but cannot prevent an # attacker who can atomically replace dest with a symlink after the check. - if st is not None and st.st_nlink > 1: + if not exclusive and st is not None and st.st_nlink > 1: raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}") flags = os.O_WRONLY | os.O_CREAT + if exclusive: + flags |= os.O_EXCL if hasattr(os, "O_BINARY"): flags |= os.O_BINARY @@ -313,7 +321,7 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob if pre_open_st is not None and not stat.S_ISREG(pre_open_st.st_mode): raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}") - if pre_open_st is not None and pre_open_st.st_nlink > 1: + if not exclusive and pre_open_st is not None and pre_open_st.st_nlink > 1: raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}") try: @@ -336,9 +344,9 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob return dest, fh -def write_upload_file_no_symlink(base_dir: Path, filename: str, data: bytes) -> Path: +def write_upload_file_no_symlink(base_dir: Path, filename: str, data: bytes, *, exclusive: bool = False) -> Path: """Write upload bytes without following a pre-existing destination symlink.""" - dest, fh = open_upload_file_no_symlink(base_dir, filename) + dest, fh = open_upload_file_no_symlink(base_dir, filename, exclusive=exclusive) with fh: fh.write(data) return dest diff --git a/backend/tests/test_channel_file_attachments.py b/backend/tests/test_channel_file_attachments.py index 22ecb4443f..bf5395a52c 100644 --- a/backend/tests/test_channel_file_attachments.py +++ b/backend/tests/test_channel_file_attachments.py @@ -282,6 +282,105 @@ class TestResolveAttachments: class TestInboundFileIngestion: + def test_concurrent_same_name_downloads_preserve_both_payloads(self, tmp_path): + from app.channels import manager + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + payloads = [b"first attachment" * 127, b"second attachment" * 91] + + async def scenario(): + # Each reader starts after its call's directory snapshot. Hold + # both downloads there so neither snapshot can see the other file. + downloads_ready = asyncio.Event() + arrivals = 0 + + async def reader(file_info, client): + nonlocal arrivals + arrivals += 1 + if arrivals == 2: + downloads_ready.set() + await asyncio.wait_for(downloads_ready.wait(), timeout=5) + return payloads[file_info["index"]] + + messages = [InboundMessage(channel_name="test-channel", chat_id="chat-1", user_id="user-1", text="attachment", files=[{"filename": "report.txt", "index": index}]) for index in range(2)] + with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir), patch.dict(manager.INBOUND_FILE_READERS, {"test-channel": reader}): + tasks = [asyncio.create_task(manager._ingest_inbound_files("thread-1", msg)) for msg in messages] + try: + return await asyncio.wait_for(asyncio.gather(*tasks), timeout=10) + finally: + downloads_ready.set() + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + results = asyncio.run(scenario()) + assert all(len(batch) == 1 for batch in results) + names = [batch[0]["filename"] for batch in results] + assert len(set(names)) == 2 + for batch, payload in zip(results, payloads, strict=True): + saved = batch[0] + assert (uploads_dir / saved["filename"]).read_bytes() == payload + assert saved["path"] == f"/mnt/user-data/uploads/{saved['filename']}" + assert saved["size"] == len(payload) + + def test_file_created_during_download_is_not_overwritten(self, tmp_path): + from app.channels import manager + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + prior_payload = b"another request published this file" + new_payload = b"this attachment arrived later" + + async def reader(file_info, client): + # A second uploader wins both the original and first suffix + # after this ingestion has taken its directory snapshot. + for name in ("report.txt", "report_1.txt"): + await asyncio.to_thread((uploads_dir / name).write_bytes, prior_payload) + return new_payload + + msg = InboundMessage(channel_name="test-channel", chat_id="chat-1", user_id="user-1", text="attachment", files=[{"filename": "report.txt"}]) + with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir), patch.dict(manager.INBOUND_FILE_READERS, {"test-channel": reader}): + result = asyncio.run(manager._ingest_inbound_files("thread-1", msg)) + + assert len(result) == 1 + assert result[0]["filename"] not in {"report.txt", "report_1.txt"} + assert (uploads_dir / result[0]["filename"]).read_bytes() == new_payload + for name in ("report.txt", "report_1.txt"): + assert (uploads_dir / name).read_bytes() == prior_payload + + def test_gateway_conversion_staging_link_is_a_filename_collision(self, tmp_path): + from app.channels import manager + from app.gateway.routers.uploads import _link_staged_no_overwrite + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + gateway_payload = b"gateway upload still being converted" + inbound_payload = b"independent IM attachment" + + def publish_gateway_upload(): + staged = uploads_dir / ".upload-fixture.part" + staged.write_bytes(gateway_payload) + # Gateway retains the staged link while a conversion descriptor + # owns this inode. This is an ordinary upload, not an unsafe path. + return _link_staged_no_overwrite(staged, uploads_dir, "report.pdf", unlink_staged=False) + + async def reader(file_info, client): + published = await asyncio.to_thread(publish_gateway_upload) + assert (await asyncio.to_thread(published.stat)).st_nlink == 2 + return inbound_payload + + msg = InboundMessage(channel_name="test-channel", chat_id="chat-1", user_id="user-1", text="attachment", files=[{"filename": "report.pdf"}]) + with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir), patch.dict(manager.INBOUND_FILE_READERS, {"test-channel": reader}): + result = asyncio.run(manager._ingest_inbound_files("thread-1", msg)) + + assert len(result) == 1 + assert result[0]["filename"] != "report.pdf" + assert (uploads_dir / result[0]["filename"]).read_bytes() == inbound_payload + assert (uploads_dir / "report.pdf").read_bytes() == gateway_payload + assert (uploads_dir / ".upload-fixture.part").read_bytes() == gateway_payload + def test_case_variant_names_keep_both_inbound_attachments(self, tmp_path): from app.channels import manager @@ -1479,3 +1578,78 @@ class TestWeChatDownloadGuardLabels: assert channel._stage_downloaded_file("photo.bin", b"x") is None assert "no state directory configured" in caplog.text + + +@pytest.mark.parametrize("channel_name", ["feishu", "dingtalk"]) +@pytest.mark.parametrize("mounted", [False, True]) +def test_embedded_upload_retries_claim_collision(tmp_path, monkeypatch, channel_name, mounted): + """A publisher winning after the filename claim must keep its own bytes.""" + import importlib + from io import BytesIO + from types import SimpleNamespace + from unittest.mock import AsyncMock + + from deerflow.config.paths import Paths + from deerflow.uploads.manager import write_upload_file_no_symlink + + paths = Paths(str(tmp_path)) + paths.ensure_thread_dirs("thread-1", user_id="owner") + uploads = paths.sandbox_uploads_dir("thread-1", user_id="owner") + module = importlib.import_module(f"app.channels.{channel_name}") + monkeypatch.setattr(module, "get_paths", lambda: paths) + synced = [] + + class Provider: + uses_thread_data_mounts = mounted + + async def acquire_async(self, thread_id=None, *, user_id=None): + assert not mounted + assert (thread_id, user_id) == ("thread-1", "owner") + return "remote-sandbox" + + def get(self, sandbox_id): + return SimpleNamespace(update_file=lambda path, content: synced.append((path, content)), release_command_scope=lambda scope: None) + + def release(self, sandbox_id): + pass + + provider = Provider() + monkeypatch.setattr(module, "get_sandbox_provider", lambda: provider) + claim = module.claim_unique_filename + other_files = {} + + def publish_after_claim(name, seen): + chosen = claim(name, seen) + if len(other_files) < 2: + content = f"other-publisher-{len(other_files)}".encode() + write_upload_file_no_symlink(uploads, chosen, content, exclusive=True) + other_files[chosen] = content + return chosen + + monkeypatch.setattr(module, "claim_unique_filename", publish_after_claim) + payload = b"CHANNEL ATTACHMENT" + if channel_name == "feishu": + channel = module.FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"}) + channel._GetMessageResourceRequest = MagicMock() + response = MagicMock() + response.success.return_value = True + response.file = BytesIO(payload) + response.file_name = "report.pdf" + channel._api_client = MagicMock() + channel._api_client.im.v1.message_resource.get.return_value = response + files = [{"file_key": "file-key"}] + else: + channel = module.DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=payload) + files = [{"type": "file", "download_code": "file-key", "filename": "report.pdf"}] + message = InboundMessage(channel_name=channel_name, chat_id="chat-1", user_id="sender", thread_ts="message-1", text="[file]", files=files) + + result = _run(channel.receive_file(message, "thread-1", user_id="owner")) + + for name, content in other_files.items(): + assert (uploads / name).read_bytes() == content + assert set(other_files) == {"report.pdf", "report_1.pdf"} + assert (uploads / "report_2.pdf").read_bytes() == payload + advertised = "/mnt/user-data/uploads/report_2.pdf" + assert advertised in result.text + assert synced == ([] if mounted else [(advertised, payload)]) diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py index 2d8517c40a..68f3d8fd00 100644 --- a/backend/tests/test_feishu_parser.py +++ b/backend/tests/test_feishu_parser.py @@ -427,12 +427,12 @@ def test_feishu_receive_file_path_traversal_failure_is_per_attachment(tmp_path, real_write = feishu_module.write_upload_file_no_symlink call_count = 0 - def flaky_write(base_dir, filename, data): + def flaky_write(base_dir, filename, data, *, exclusive=False): nonlocal call_count call_count += 1 if call_count == 1: raise PathTraversalError("Path traversal detected") - return real_write(base_dir, filename, data) + return real_write(base_dir, filename, data, exclusive=exclusive) monkeypatch.setattr(feishu_module, "write_upload_file_no_symlink", flaky_write) msg = InboundMessage( diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index de396431b0..444c92403f 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -233,6 +233,45 @@ class TestValidatePathTraversal: class TestWriteUploadFileNoSymlink: + @pytest.mark.parametrize("without_nofollow", [False, True]) + def test_exclusive_write_treats_existing_hardlink_as_collision(self, tmp_path, monkeypatch, without_nofollow): + if without_nofollow: + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + source = tmp_path / "source.txt" + source.write_bytes(b"existing upload") + os.link(source, tmp_path / "notes.txt") + + with pytest.raises(FileExistsError): + write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True) + # The default replacement path still rejects hardlinks outright. + with pytest.raises(UnsafeUploadPathError, match="multiple links"): + write_upload_file_no_symlink(tmp_path, "notes.txt", b"replacement") + + assert source.read_bytes() == b"existing upload" + assert (tmp_path / "notes.txt").read_bytes() == b"existing upload" + + @pytest.mark.parametrize("without_nofollow", [False, True]) + def test_exclusive_write_preserves_existing_file(self, tmp_path, monkeypatch, without_nofollow): + if without_nofollow: + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + dest = tmp_path / "notes.txt" + dest.write_bytes(b"previous upload") + + with pytest.raises(FileExistsError): + write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True) + + assert dest.read_bytes() == b"previous upload" + + @pytest.mark.parametrize("without_nofollow", [False, True]) + def test_exclusive_write_creates_new_file(self, tmp_path, monkeypatch, without_nofollow): + if without_nofollow: + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + + dest = write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True) + + assert dest == tmp_path / "notes.txt" + assert dest.read_bytes() == b"new attachment" + def test_writes_new_file(self, tmp_path): dest = write_upload_file_no_symlink(tmp_path, "notes.txt", b"hello")