From c9d3e48a4adc2512b5cd328eaef4118c83bdb9d2 Mon Sep 17 00:00:00 2001 From: XuRuuuy <2948622198@qq.com> Date: Sat, 24 Oct 2026 20:57:27 +0800 Subject: [PATCH] fix(uploads): never truncate an existing upload on same-name collision (#7388) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(uploads): never truncate an existing upload on same-name collision Same-name uploads silently destroyed the earlier file on disk: the shared write path (open_upload_file_no_symlink) opened the destination with O_WRONLY|O_CREAT (no O_EXCL) and ftruncate(0), so re-uploading report.pdf opened and emptied the report.pdf already stored for the thread, and two concurrent same-name uploads could race on one inode. The Gateway ingestion pipeline already refuses overwrites (staged no-overwrite link commit), but the embedded client upload path (copy_upload_file_no_symlink) and the channel ingress paths (write_upload_file_no_symlink) still went through the truncating open. Open every upload destination with O_EXCL. On FileExistsError, claim the next free _N suffix against the directory's current contents via the existing claim_unique_filename helper and retry — the same deterministic behavior the Gateway pipeline documents (name_1.ext). The existing file is never opened, so its bytes survive intact under any future naming policy. Embedded-client and channel inbound-file responses now report the name the bytes actually landed under instead of the requested name. * test(uploads): cover concurrent same-name writers * fix(uploads): skip known copy collisions before staging --------- Co-authored-by: Willem Jiang --- README.md | 3 + backend/app/channels/manager.py | 5 +- backend/packages/harness/deerflow/client.py | 14 +- .../harness/deerflow/uploads/AGENTS.md | 14 +- .../harness/deerflow/uploads/manager.py | 166 +++++++++------- backend/tests/AGENTS.md | 5 + backend/tests/test_client.py | 28 ++- ...test_client_upload_companion_collisions.py | 15 +- backend/tests/test_client_upload_snapshot.py | 29 ++- backend/tests/test_uploads_manager.py | 185 +++++++++++++++--- 10 files changed, 349 insertions(+), 115 deletions(-) diff --git a/README.md b/README.md index d7f0e1ce0d..2731b7f98e 100644 --- a/README.md +++ b/README.md @@ -1125,6 +1125,9 @@ 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. +Embedded client uploads also keep earlier same-name files under distinct names. +Renamed originals and Markdown companions retain private source snapshots and +become visible only after the complete copy is published. Feishu and DingTalk attachment transfers return standalone temporary sandboxes to the pool after synchronization, without interrupting an active run. diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 2e8a001a2d..620dc99a30 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -1530,9 +1530,10 @@ async def _ingest_inbound_files_impl(thread_id: str, msg: InboundMessage, *, use created.append( { - "filename": safe_name, + # Report the returned path rather than the pre-scan claim. + "filename": dest.name, "size": len(data), - "path": f"/mnt/user-data/uploads/{safe_name}", + "path": f"/mnt/user-data/uploads/{dest.name}", "is_image": ftype == "image", } ) diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 0cfcdeb343..67780e13e5 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -1841,22 +1841,26 @@ class DeerFlowClient: skipped_files.append(dest_name) continue invalidate_overwritten_upload(dest) + # Same-name collisions land under the next free ``_N`` suffix; + # report the name the bytes actually landed under, never the + # requested one (issue #3750). + resolved_name = dest.name info: dict[str, Any] = { - "filename": dest_name, + "filename": resolved_name, "size": dest.stat().st_size, "path": str(dest), - "virtual_path": upload_virtual_path(dest_name), - "artifact_url": upload_artifact_url(thread_id, dest_name), + "virtual_path": upload_virtual_path(resolved_name), + "artifact_url": upload_artifact_url(thread_id, resolved_name), } - if dest_name != src_path.name: + if resolved_name != src_path.name: info["original_filename"] = src_path.name if src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS: # Reserve companion .md name before convert so two stems # that collapse to the same .md (or a prior .md upload) # cannot silently overwrite each other. - provisional_md_name = Path(dest_name).with_suffix(".md").name + provisional_md_name = Path(resolved_name).with_suffix(".md").name unique_md_name = claim_unique_filename(provisional_md_name, seen_names) try: # Convert the same private capture used to publish the diff --git a/backend/packages/harness/deerflow/uploads/AGENTS.md b/backend/packages/harness/deerflow/uploads/AGENTS.md index 9e47dfff86..45e6b33434 100644 --- a/backend/packages/harness/deerflow/uploads/AGENTS.md +++ b/backend/packages/harness/deerflow/uploads/AGENTS.md @@ -15,12 +15,18 @@ errors, keep its files and continue elsewhere. A vanished directory stays quiet. Embedded first-time uploads use `publish_upload_copy` for suffix retries from the normalized original name, for originals and companions. With `snapshot_path`, capture the opened source once before retries and retain its original `fstat` -metadata; every attempt reads the held private capture. Replacement copies share -this capture helper. Never copy snapshot permission bits onto the saved upload. +metadata; every attempt reads the held private capture. Collision copies share +this capture helper and publish complete staged copies under the next free name. +Validate the requested destination before seeding copy suffixes from a directory +snapshot, so unsafe links still fail and known collisions do not stage twice. +Atomic no-replace links remain the concurrency boundary; retry after a lost claim. +Never copy snapshot permission bits onto the saved upload. Close the caller's descriptor after capture and finish conversion workers before removing captures. Never reopen the caller/sandbox upload path for conversion. -Original upload names already present in the call's initial regular-file snapshot -retain replacement behavior; these replacements are not serialized. Converted +Original names already present in the initial snapshot also take a free suffix; +never truncate an earlier upload. Default open/write helpers claim with O_EXCL +and retry suffixes, while explicit `exclusive=True` preserves exact-name +FileExistsError so callers can own their retry policy. Converted Markdown always uses no-replace publication, including across upload calls, and registers the actual saved name. An existing basename does not establish companion ownership. Apply source timestamps after new-file publication so cleanup cannot sweep an in-flight copy diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index ce27ace744..ad55503b1c 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -168,21 +168,13 @@ def validate_path_traversal(path: Path, base: Path) -> None: def _replacement_destination_is_hardlinked(st: os.stat_result) -> bool: - """Whether a *replacement* destination shows an untrusted extra hard link. + """Preserve the pre-existing hardlink guard where link counts are reliable. - Only used on the non-exclusive replacement path, where the module is about to - overwrite a file it published itself. On POSIX ``st_nlink`` is authoritative, - so ``st_nlink > 1`` means a second name — possibly outside the uploads tree — - aliases the destination and overwriting it would mutate that other file. - - Windows cannot answer that question with ``st_nlink``: NTFS leaves the count at - 2 once one of the two names of a published upload is unlinked, and unlinking a - hard link does not decrement it either. The count there is therefore *always* - at least 2 for a file this module just published, which made the guard reject - every legitimate replacement upload (``copy_upload_file_no_symlink``) and made - the alias-reclaiming sweep unable to clear it. Treat the count as unreliable on - Windows; the symlink and regular-file guards on the same destination, and the - ``exclusive`` (first-claim) path used by every untrusted caller, are unchanged. + Default destination validation rejects POSIX entries with another alias. + Windows can retain a stale count of two after a staging alias is removed, + so that count cannot reject an otherwise ordinary published upload there. + Collision writes never open the old entry, and both O_EXCL creation modes + require a newly opened file to have exactly one link. """ return os.name != "nt" and st.st_nlink > 1 @@ -319,6 +311,15 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None, *, mi return removed +def _existing_dir_names(base_dir: Path) -> set[str]: + """Snapshot the names currently in *base_dir* (empty set when it cannot be scanned).""" + try: + with os.scandir(base_dir) as entries: + return {entry.name for entry in entries} + except OSError: + return set() + + 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. @@ -326,14 +327,22 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: boo therefore leave a symlink at a future upload filename. Normal ``Path.write_bytes`` follows that link and can overwrite files outside the uploads directory with gateway privileges. This helper rejects symlink destinations using ``O_NOFOLLOW`` - on POSIX. On Windows (which lacks ``O_NOFOLLOW``), it uses dual ``lstat`` checks - 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. + on POSIX. On Windows (which lacks ``O_NOFOLLOW``), dual ``lstat`` checks + and ``O_EXCL`` refuse occupied destination names, with ``fstat`` validation + after ``open()``. Path-traversal validation applies 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. + Same-name collisions never destroy the existing file: the destination is opened + with ``O_EXCL`` (so a pre-existing file can never be opened and truncated), and + on :class:`FileExistsError` the next free ``_N`` suffix is claimed against the + directory's current contents via :func:`claim_unique_filename` — the same helper + the Gateway ingestion pipeline and the channel ingress paths use — and the open + is retried. Two concurrent uploads of one name therefore land as ``name.ext`` + and ``name_1.ext`` with both byte streams intact, instead of racing on one inode + (issue #3750). The returned path is the name the bytes actually landed under; + callers must report it (not the requested name) back to users. + + ``exclusive=True`` instead raises ``FileExistsError`` on a collision, so + callers that own their filename retry policy retain an exact-name claim. """ safe_name = normalize_filename(filename) dest = validate_upload_destination(base_dir, safe_name, exclusive=exclusive) @@ -346,24 +355,37 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: boo 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 + # O_EXCL makes open() fail with EEXIST if dest already exists, so an + # upload can never open and truncate an earlier file of the same name. + flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_EXCL if hasattr(os, "O_NONBLOCK"): flags |= os.O_NONBLOCK - try: - fd = os.open(dest, flags, 0o600) - except OSError as exc: - if exc.errno in {errno.ELOOP, errno.EISDIR, errno.ENOTDIR, errno.ENXIO, errno.EAGAIN}: - raise UnsafeUploadPathError(f"Unsafe upload destination: {safe_name}") from exc - raise + seen: set[str] | None = None + while True: + try: + fd = os.open(dest, flags, 0o600) + break + except FileExistsError: + if exclusive: + raise + # Regular-file collision: claim the next free ``_N`` suffix + # against the directory's current contents and retry. The + # existing file is never opened, so its bytes survive intact, + # and the O_EXCL retry keeps concurrent uploads off one inode. + if seen is None: + seen = _existing_dir_names(base_dir) + seen.add(dest.name) + dest = base_dir / claim_unique_filename(safe_name, seen) + except OSError as exc: + if exc.errno in {errno.ELOOP, errno.EISDIR, errno.ENOTDIR, errno.ENXIO, errno.EAGAIN}: + raise UnsafeUploadPathError(f"Unsafe upload destination: {safe_name}") from exc + raise try: opened_stat = os.fstat(fd) if not stat.S_ISREG(opened_stat.st_mode) or opened_stat.st_nlink != 1: raise UnsafeUploadPathError(f"Upload destination is not an exclusive regular file: {safe_name}") - os.ftruncate(fd, 0) fh = os.fdopen(fd, "wb") fd = -1 finally: @@ -371,17 +393,12 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: boo os.close(fd) return dest, fh - # Windows: no O_NOFOLLOW available. Uses a second lstat immediately before open() - # to narrow the TOCTOU window, then fstat after open() as a further defence. - # 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. + # Windows: retain pre-open validation and reject an occupied destination + # atomically with O_EXCL, then validate the newly claimed descriptor. if not exclusive and st is not None and _replacement_destination_is_hardlinked(st): raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}") - flags = os.O_WRONLY | os.O_CREAT - if exclusive: - flags |= os.O_EXCL + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, "O_BINARY"): flags |= os.O_BINARY @@ -395,21 +412,31 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: boo if not exclusive and pre_open_st is not None and _replacement_destination_is_hardlinked(pre_open_st): raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}") - try: - fd = os.open(dest, flags, 0o600) - except OSError as exc: - if exc.errno in {errno.EISDIR, errno.ENOTDIR, errno.ENXIO, errno.EAGAIN}: - raise UnsafeUploadPathError(f"Unsafe upload destination: {safe_name}") from exc - raise + seen: set[str] | None = None + while True: + try: + fd = os.open(dest, flags, 0o600) + break + except FileExistsError: + if exclusive: + raise + # Regular-file collision: same contract as the POSIX branch — + # claim the next free ``_N`` suffix and retry under O_EXCL so an + # existing file is never opened and truncated (issue #3750). + if seen is None: + seen = _existing_dir_names(base_dir) + seen.add(dest.name) + dest = base_dir / claim_unique_filename(safe_name, seen) + except OSError as exc: + if exc.errno in {errno.EISDIR, errno.ENOTDIR, errno.ENXIO, errno.EAGAIN}: + raise UnsafeUploadPathError(f"Unsafe upload destination: {safe_name}") from exc + raise try: opened_stat = os.fstat(fd) - # The exclusive path keeps the strict post-open single-link check on every - # platform (a just-exclusively-created file must have exactly one link); - # only the non-exclusive replacement path defers to the Windows-aware guard. - if not stat.S_ISREG(opened_stat.st_mode) or (exclusive and opened_stat.st_nlink != 1) or (not exclusive and _replacement_destination_is_hardlinked(opened_stat)): + # Both modes create new files under O_EXCL and require one link. + if not stat.S_ISREG(opened_stat.st_mode) or opened_stat.st_nlink != 1: raise UnsafeUploadPathError(f"Upload destination is not an exclusive regular file: {safe_name}") - os.ftruncate(fd, 0) fh = os.fdopen(fd, "wb") fd = -1 finally: @@ -465,32 +492,37 @@ def copy_upload_file_no_symlink(base_dir: Path, filename: str, src: Path, *, sna """Copy *src* into an upload destination without following a destination symlink. Matches ``shutil.copy2`` for content, permission bits and timestamps, but - opens the destination through :func:`open_upload_file_no_symlink` and - applies the metadata to that descriptor, never to the name. The source is - opened first, so a missing source leaves an existing destination intact. + publishes a complete staged copy without replacing an existing file. + Metadata is applied to the staged descriptor, never to a re-resolved name. + The source is opened first, so a missing source leaves an existing destination intact. Where descriptor-based ``chmod``/``utime`` are unavailable (Windows), the destination keeps its default mode and the copy time. Copying a file onto itself raises :class:`shutil.SameFileError` as - ``copy2`` does, and does so before the destination is opened: opening it - truncates, which would otherwise leave the caller copying an emptied file - over itself. Re-uploading a file that already sits in the uploads - directory takes exactly that path. + ``copy2`` does, before capturing or publishing any destination bytes. Re-uploading + a file that already sits in the uploads directory takes exactly that path. A same-name collision with a different + file never destroys the earlier copy: the copy lands under the next free + ``_N`` suffix claimed by :func:`claim_unique_filename`, and the + returned path is the name the bytes actually landed under (issue #3750). With ``snapshot_path``, first capture the opened source there and publish those captured bytes. The caller owns this path in a private directory outside the sandbox tree and can convert it without reopening ``src``. """ - with _open_upload_copy_source(base_dir, filename, src, snapshot_path=snapshot_path) as (captured, src_stat): - dest, fh = open_upload_file_no_symlink(base_dir, filename) - with fh: - shutil.copyfileobj(captured, fh) - fh.flush() - if os.chmod in os.supports_fd: - os.chmod(fh.fileno(), stat.S_IMODE(src_stat.st_mode)) - if os.utime in os.supports_fd: - os.utime(fh.fileno(), ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns)) - return dest + safe_name = normalize_filename(filename) + with _open_upload_copy_source(base_dir, safe_name, src, snapshot_path=snapshot_path) as (captured, src_stat): + validate_upload_destination(base_dir, safe_name) + # Skip known collisions before staging; the atomic link still decides + # whether this candidate is free when concurrent writers publish. + seen = _existing_dir_names(base_dir) + candidate = claim_unique_filename(safe_name, seen) + while True: + try: + captured.seek(0) + return _publish_upload_copy_source(base_dir, candidate, captured, src_stat) + except FileExistsError: + seen.update(_existing_dir_names(base_dir)) + candidate = claim_unique_filename(safe_name, seen) def _validate_upload_copy_identity(path: Path, expected: os.stat_result) -> None: diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index 9d269ce214..5734c9e1b1 100644 --- a/backend/tests/AGENTS.md +++ b/backend/tests/AGENTS.md @@ -89,6 +89,11 @@ Upload case-collision coverage uses separate HTTP requests and preserves both reported payloads. Observe real filename-claim inputs to pin the disk seed; case-insensitive hosts can otherwise mask a missing seed through link retries. +Same-name upload concurrency tests synchronize the first real claim: `os.open` +for direct writes and `os.link` for staged copies. Cover native/fallback options, +fresh/occupied copy names, every caller's exact bytes and no staging leftovers. +Drain workers and abort barriers on failure; do not rely on scheduling or sleeps. + Channel reload cancellation regressions must assert the worker returned the stale snapshot before checking that newer runtime config survived. Completion alone cannot prove the race: loader exceptions are caught and return `None`. diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index 3b8bbd6ebc..d2a3a9780f 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -3235,13 +3235,20 @@ class TestUploads: patch("deerflow.client.get_uploads_dir", return_value=uploads_dir), patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir), ): - client.upload_files("thread-1", [authored]) + result = client.upload_files("thread-1", [authored]) from deerflow.utils.file_outline import extract_outline_for_file - assert resolve_companion(uploads_dir / "a.pdf") is None - assert extract_outline_for_file(uploads_dir / "a.pdf") == ([], []) - assert extract_outline_for_file(uploads_dir / "a_1.md")[0] == [{"title": "My notes", "line": 1}] + # Same-name uploads can no longer clobber an earlier file in place + # (issue #3750): the authored copy lands under the next free + # suffix, and the a.pdf companion record stays valid because its + # markdown was never destroyed. + assert result["files"][0]["filename"] == "a_1_1.md" + assert resolve_companion(uploads_dir / "a.pdf") == uploads_dir / "a_1.md" + assert (uploads_dir / "a_1.md").read_text(encoding="utf-8") == "FROM:a.pdf" + assert (uploads_dir / "a_1_1.md").read_text(encoding="utf-8") == "# My notes" + assert extract_outline_for_file(uploads_dir / "a.pdf") == ([], ["FROM:a.pdf"]) + assert extract_outline_for_file(uploads_dir / "a_1_1.md")[0] == [{"title": "My notes", "line": 1}] def test_upload_files_failed_conversion_releases_the_claimed_markdown_name(self, client): """A conversion that writes nothing must not reserve stem.md against a later companion. @@ -5378,7 +5385,7 @@ class TestUploadDuplicateFilenames: assert companion.read_bytes() == source.read_bytes() assert len(list(uploads_dir.iterdir())) == 4 - def test_sequential_upload_keeps_replacement_behavior(self, client, tmp_path): + def test_sequential_upload_preserves_existing_file_and_reports_claimed_name(self, client, tmp_path): uploads_dir = tmp_path / "uploads" uploads_dir.mkdir() first = tmp_path / "first" / "data.txt" @@ -5392,9 +5399,14 @@ class TestUploadDuplicateFilenames: client.upload_files("t-dup", [first]) result = client.upload_files("t-dup", [second]) - assert result["files"][0]["filename"] == "data.txt" - assert (uploads_dir / "data.txt").read_bytes() == b"second" - assert len(list(uploads_dir.iterdir())) == 1 + info = result["files"][0] + assert info["filename"] == "data_1.txt" + assert info["original_filename"] == "data.txt" + assert info["virtual_path"] == "/mnt/user-data/uploads/data_1.txt" + assert info["artifact_url"].endswith("/uploads/data_1.txt") + assert (uploads_dir / "data.txt").read_bytes() == b"first" + assert (uploads_dir / "data_1.txt").read_bytes() == b"second" + assert len(list(uploads_dir.iterdir())) == 2 def test_duplicate_filenames_auto_renamed(self, client): """Two files with same basename → second gets _1 suffix.""" diff --git a/backend/tests/test_client_upload_companion_collisions.py b/backend/tests/test_client_upload_companion_collisions.py index 3e2657b12e..752fed4615 100644 --- a/backend/tests/test_client_upload_companion_collisions.py +++ b/backend/tests/test_client_upload_companion_collisions.py @@ -78,11 +78,12 @@ def test_separate_uploads_preserve_existing_markdown(upload_case, tmp_path, in_l assert resolve_companion(uploads / first.name) == original_markdown -def test_reupload_replaces_original_but_preserves_occupied_companion_names(upload_case, tmp_path): +def test_reupload_preserves_original_and_occupied_companion_names(upload_case, tmp_path): client, uploads = upload_case source = tmp_path / "report.xlsx" _document(source, "FIRST DOCUMENT") client.upload_files("companion-thread", [source]) + first_original = (uploads / source.name).read_bytes() first_markdown = (uploads / "report.md").read_bytes() occupied = uploads / "report_1.md" occupied.write_text("INDEPENDENT NOTES", encoding="utf-8") @@ -92,11 +93,13 @@ def test_reupload_replaces_original_but_preserves_occupied_companion_names(uploa assert response["success"] info = response["files"][0] - assert info["filename"] == "report.xlsx" - assert (uploads / source.name).read_bytes() == source.read_bytes() - assert info["markdown_file"] == "report_2.md" + assert info["filename"] == "report_1.xlsx" + assert (uploads / source.name).read_bytes() == first_original + assert (uploads / info["filename"]).read_bytes() == source.read_bytes() + assert info["markdown_file"] == "report_1_1.md" assert (uploads / "report.md").read_bytes() == first_markdown assert occupied.read_text(encoding="utf-8") == "INDEPENDENT NOTES" - companion = uploads / "report_2.md" + companion = uploads / info["markdown_file"] assert "REPLACEMENT DOCUMENT" in companion.read_text(encoding="utf-8") - assert resolve_companion(uploads / source.name) == companion + assert resolve_companion(uploads / info["filename"]) == companion + assert resolve_companion(uploads / source.name) == uploads / "report.md" diff --git a/backend/tests/test_client_upload_snapshot.py b/backend/tests/test_client_upload_snapshot.py index 793c2a1a14..efacef8e6b 100644 --- a/backend/tests/test_client_upload_snapshot.py +++ b/backend/tests/test_client_upload_snapshot.py @@ -178,7 +178,10 @@ def test_real_xlsx_conversion_uses_captured_version(upload_case, monkeypatch, tm response = call_upload(client, source, in_loop) assert response["success"] assert saves == [source] - saved = uploads / "report.xlsx" + saved = uploads / response["files"][0]["filename"] + if existing: + assert saved.name == "report_1.xlsx" + assert (uploads / "report.xlsx").read_bytes() == b"OLD UPLOAD" saved_workbook = load_workbook(saved) try: assert saved_workbook.active["A1"].value == "VERSION A" @@ -341,3 +344,27 @@ def test_collision_retries_keep_one_capture(upload_case, monkeypatch, tmp_path, assert not captures[0].is_relative_to(uploads) assert not captures[0].parent.exists() assert not list(uploads.glob(".upload-*.part")) + + +def test_colliding_original_is_published_only_after_copy_completes(upload_case, monkeypatch, tmp_path): + import deerflow.uploads.manager as manager + + client, uploads = upload_case + source = tmp_path / "notes.txt" + source.write_bytes(b"NEW UPLOAD") + (uploads / source.name).write_bytes(b"OLD UPLOAD") + observed = [] + real_copy = manager.shutil.copyfileobj + + def inspect_copy(src, dst, *args, **kwargs): + real_copy(src, dst, *args, **kwargs) + observed.append({item["filename"] for item in manager.list_files_in_dir(uploads)["files"]}) + + monkeypatch.setattr(manager.shutil, "copyfileobj", inspect_copy) + response = client.upload_files("snapshot-thread", [source]) + + assert observed + assert all(names == {"notes.txt"} for names in observed) + assert response["files"][0]["filename"] == "notes_1.txt" + assert (uploads / "notes.txt").read_bytes() == b"OLD UPLOAD" + assert (uploads / "notes_1.txt").read_bytes() == b"NEW UPLOAD" diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index 9523bec386..9936a0fe80 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -6,7 +6,9 @@ import os import shutil import stat import time +from concurrent.futures import ThreadPoolExecutor from datetime import timedelta +from threading import Barrier, local from unittest.mock import patch import pytest @@ -245,13 +247,15 @@ class TestWriteUploadFileNoSymlink: 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, where the - # platform maintains a trustworthy link count. Windows keeps a published - # upload at ``st_nlink == 2`` forever, so the replacement overwrites the - # destination (and, being a hard link, ``source`` with it). + # Default collision retries keep both aliases intact. POSIX still + # rejects a pre-existing multi-link destination; Windows cannot use its + # stale link count to reject files published through a staging link. if os.name == "nt": - assert write_upload_file_no_symlink(tmp_path, "notes.txt", b"replacement").read_bytes() == b"replacement" - assert source.read_bytes() == b"replacement" + renamed = write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment") + assert renamed.name == "notes_1.txt" + assert renamed.read_bytes() == b"new attachment" + assert source.read_bytes() == b"existing upload" + assert (tmp_path / "notes.txt").read_bytes() == b"existing upload" else: with pytest.raises(UnsafeUploadPathError, match="multiple links"): write_upload_file_no_symlink(tmp_path, "notes.txt", b"replacement") @@ -286,16 +290,69 @@ class TestWriteUploadFileNoSymlink: assert dest == tmp_path / "notes.txt" assert dest.read_bytes() == b"hello" - def test_overwrites_existing_regular_file_with_single_link(self, tmp_path): - dest = tmp_path / "notes.txt" - dest.write_bytes(b"old contents") - assert os.stat(dest).st_nlink == 1 + def test_collision_appends_suffix_and_preserves_existing_file(self, tmp_path): + """A same-name upload must never truncate the file already on disk (issue #3750).""" + existing = tmp_path / "notes.txt" + existing.write_bytes(b"old contents") + assert os.stat(existing).st_nlink == 1 result = write_upload_file_no_symlink(tmp_path, "notes.txt", b"new contents") - assert result == dest - assert dest.read_bytes() == b"new contents" - assert os.stat(dest).st_nlink == 1 + assert result == tmp_path / "notes_1.txt" + assert existing.read_bytes() == b"old contents" + assert result.read_bytes() == b"new contents" + + def test_repeated_same_name_uploads_keep_every_version(self, tmp_path): + write_upload_file_no_symlink(tmp_path, "notes.txt", b"v1") + write_upload_file_no_symlink(tmp_path, "notes.txt", b"v2") + + result = write_upload_file_no_symlink(tmp_path, "notes.txt", b"v3") + + assert result == tmp_path / "notes_2.txt" + assert {p.read_bytes() for p in tmp_path.glob("notes*.txt")} == {b"v1", b"v2", b"v3"} + + @pytest.mark.parametrize("without_nofollow", [False, True], ids=["native", "without-nofollow"]) + def test_concurrent_same_name_uploads_keep_every_payload(self, tmp_path, monkeypatch, without_nofollow): + if without_nofollow: + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + filename = "notes.txt" + payloads = [f"upload-{index}:".encode() + bytes([index]) * 4096 for index in range(12)] + first_claim = Barrier(len(payloads), timeout=15) + real_open = os.open + + def synchronized_open(path, flags, *args, **kwargs): + if os.path.basename(path) == filename and flags & os.O_CREAT: + # All writers complete their pre-open checks before any can + # create the shared name. Keep the actual filesystem open. + first_claim.wait() + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", synchronized_open) + with ThreadPoolExecutor(max_workers=len(payloads)) as executor: + futures = [executor.submit(write_upload_file_no_symlink, tmp_path, filename, payload) for payload in payloads] + try: + paths = [future.result(timeout=20) for future in futures] + finally: + first_claim.abort() + + assert len(set(paths)) == len(payloads) + assert {path.name for path in paths} == {filename, *(f"notes_{index}.txt" for index in range(1, len(payloads)))} + assert set(tmp_path.iterdir()) == set(paths) + for path, payload in zip(paths, payloads, strict=True): + assert path.read_bytes() == payload + + def test_collision_fallback_without_no_follow_support_preserves_existing_file(self, tmp_path, monkeypatch): + # When O_NOFOLLOW is absent (Windows), the function falls back to a + # dual-lstat + fstat approach — which must also refuse to truncate. + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + existing = tmp_path / "notes.txt" + existing.write_bytes(b"old contents") + + result = write_upload_file_no_symlink(tmp_path, "notes.txt", b"new contents") + + assert result == tmp_path / "notes_1.txt" + assert existing.read_bytes() == b"old contents" + assert result.read_bytes() == b"new contents" def test_fallback_without_no_follow_support_succeeds(self, tmp_path, monkeypatch): monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) @@ -306,11 +363,9 @@ class TestWriteUploadFileNoSymlink: assert result == tmp_path / "notes.txt" assert (tmp_path / "notes.txt").read_bytes() == b"hello" - def test_exclusive_write_keeps_post_open_link_guard_without_no_follow(self, tmp_path, monkeypatch): - """Without ``O_NOFOLLOW`` (the Windows fallback branch), the exclusive - path keeps its strict post-open single-link check: a file that opens - with more than one link is rejected on every platform. Only the - non-exclusive replacement path defers to the Windows-aware guard.""" + @pytest.mark.parametrize("exclusive", [False, True]) + def test_write_keeps_post_open_link_guard_without_no_follow(self, tmp_path, monkeypatch, exclusive): + """Both O_EXCL creation modes require a new file with one link.""" monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) real_fstat = os.fstat @@ -321,7 +376,7 @@ class TestWriteUploadFileNoSymlink: monkeypatch.setattr(os, "fstat", twin_linked_fstat) with pytest.raises(UnsafeUploadPathError, match="not an exclusive regular file"): - write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True) + write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=exclusive) def test_open_uses_nonblocking_flag_when_available(self, tmp_path): if not hasattr(os, "O_NONBLOCK"): @@ -367,7 +422,8 @@ class TestCopyUploadFileNoSymlink: if os.utime in os.supports_fd: assert os.stat(dest).st_mtime_ns == 1_700_000_000_000_000_000 - def test_overwrites_existing_regular_file(self, tmp_path): + def test_collision_copies_to_unique_name_and_preserves_existing(self, tmp_path): + """Re-uploading an existing name must not destroy the earlier copy (issue #3750).""" uploads = tmp_path / "uploads" uploads.mkdir() (uploads / "notes.txt").write_bytes(b"old contents") @@ -376,8 +432,91 @@ class TestCopyUploadFileNoSymlink: dest = copy_upload_file_no_symlink(uploads, "notes.txt", src) + assert dest == uploads / "notes_1.txt" + assert (uploads / "notes.txt").read_bytes() == b"old contents" assert dest.read_bytes() == b"new contents" + @pytest.mark.parametrize("capture", [False, True], ids=["direct", "snapshot"]) + def test_known_collision_stages_payload_once(self, tmp_path, capture): + uploads = tmp_path / "uploads" + uploads.mkdir() + existing = {"notes.txt": b"first upload", "NOTES_1.TXT": b"second upload"} + for name, data in existing.items(): + (uploads / name).write_bytes(data) + src = tmp_path / "source.txt" + src.write_bytes(b"new contents" * 4096) + snapshot = tmp_path / "snapshot" if capture else None + + with patch("deerflow.uploads.manager.shutil.copyfileobj", wraps=shutil.copyfileobj) as copy: + dest = copy_upload_file_no_symlink(uploads, "notes.txt", src, snapshot_path=snapshot) + + assert copy.call_count == 1 + int(capture) + assert dest.name == "notes_2.txt" + assert dest.read_bytes() == src.read_bytes() + for name, data in existing.items(): + assert (uploads / name).read_bytes() == data + assert {path.name for path in uploads.iterdir()} == {*existing, dest.name} + if snapshot is not None: + assert snapshot.read_bytes() == src.read_bytes() + + @pytest.mark.parametrize("existing", [False, True], ids=["fresh", "occupied"]) + @pytest.mark.parametrize("nofollow", [False, True], ids=["link-fallback", "link-nofollow"]) + def test_concurrent_same_name_copies_keep_every_payload(self, tmp_path, monkeypatch, existing, nofollow): + uploads = tmp_path / "uploads" + uploads.mkdir() + if existing: + (uploads / "notes.txt").write_bytes(b"original upload") + payloads = [f"copy-{index}:".encode() + bytes([index]) * 4096 for index in range(8)] + sources = [tmp_path / f"source-{index}.txt" for index in range(len(payloads))] + for source, payload in zip(sources, payloads, strict=True): + source.write_bytes(payload) + first_publication = Barrier(len(payloads), timeout=15) + worker_state = local() + real_link = os.link + if nofollow and real_link not in os.supports_follow_symlinks: + pytest.skip("host does not support no-follow hard links") + + def synchronized_link(staging, dest, **kwargs): + if not getattr(worker_state, "published", False): + worker_state.published = True + first_publication.wait() + return real_link(staging, dest, **kwargs) + + monkeypatch.setattr(os, "link", synchronized_link) + monkeypatch.setattr(os, "supports_follow_symlinks", {synchronized_link} if nofollow else set()) + with ThreadPoolExecutor(max_workers=len(payloads)) as executor: + futures = [executor.submit(copy_upload_file_no_symlink, uploads, "notes.txt", source) for source in sources] + try: + paths = [future.result(timeout=20) for future in futures] + finally: + first_publication.abort() + + assert len(set(paths)) == len(payloads) + start = int(existing) + expected = {"notes.txt" if index == 0 else f"notes_{index}.txt" for index in range(start, start + len(payloads))} + assert {path.name for path in paths} == expected + assert {path.name for path in uploads.iterdir()} == expected | ({"notes.txt"} if existing else set()) + for path, payload in zip(paths, payloads, strict=True): + assert path.read_bytes() == payload + if existing: + assert (uploads / "notes.txt").read_bytes() == b"original upload" + + @pytest.mark.skipif(os.name == "nt", reason="POSIX destination hardlink guard") + def test_existing_hardlinked_destination_is_not_silently_renamed(self, tmp_path): + uploads = tmp_path / "uploads" + uploads.mkdir() + original = tmp_path / "original.txt" + original.write_bytes(b"original upload") + os.link(original, uploads / "notes.txt") + src = tmp_path / "source.txt" + src.write_bytes(b"new contents") + + with pytest.raises(UnsafeUploadPathError, match="multiple links"): + copy_upload_file_no_symlink(uploads, "notes.txt", src) + + assert (uploads / "notes.txt").read_bytes() == b"original upload" + assert [path.name for path in uploads.iterdir()] == ["notes.txt"] + def test_no_replace_copy_preserves_existing_file_and_cleans_stage(self, tmp_path): uploads = tmp_path / "uploads" uploads.mkdir() @@ -857,8 +996,10 @@ class TestCleanupStaleUploadStagingFiles: assert validate_upload_destination(uploads_dir, "notes.txt") == published replacement = tmp_path / "replacement.txt" replacement.write_bytes(b"replacement bytes") - assert copy_upload_file_no_symlink(uploads_dir, "notes.txt", replacement) == published - assert published.read_bytes() == b"replacement bytes" + copied = copy_upload_file_no_symlink(uploads_dir, "notes.txt", replacement) + assert copied == uploads_dir / "notes_1.txt" + assert copied.read_bytes() == b"replacement bytes" + assert published.read_bytes() == b"published bytes" def test_skips_staging_file_that_vanished_before_its_age_was_read(self, tmp_path, caplog): uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads"