diff --git a/CHANGELOG.md b/CHANGELOG.md index 900aa1208..6e617bd55 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2785,6 +2785,13 @@ This release closes that milestone with **765 merged pull requests**. ### Security +- **uploads:** Document conversion no longer re-opens the upload by name. The + Gateway converted the committed file and the embedded client converted the + copy it had just placed in the thread's uploads directory, so a sandbox that + replaced that name with a symlink in between had a host file converted into + the thread as the `.md` companion. The Gateway now converts a private copy of + the staged bytes, read through the descriptor it wrote, and the client + converts the caller's own source file. ([#5611]) - **client:** `DeerFlowClient.upload_files` no longer writes through a symlink. A symlink planted in the sandbox-writable uploads directory, at an upload's name or its Markdown companion's name, made the embedded client @@ -4323,3 +4330,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#5534]: https://github.com/bytedance/deer-flow/pull/5534 [#5547]: https://github.com/bytedance/deer-flow/pull/5547 [#5578]: https://github.com/bytedance/deer-flow/pull/5578 +[#5611]: https://github.com/bytedance/deer-flow/pull/5611 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 425deeb7a..4b2e4341a 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -2091,6 +2091,10 @@ ### 安全 +- **上传：** 文档转换不再按文件名重新打开上传文件。此前 Gateway 转换的是已提交的文件，嵌入式 + 客户端转换的是刚放入线程 uploads 目录的副本，因此沙箱若在此期间把该文件名替换为符号链接， + 宿主文件的内容就会被转换成该线程的 `.md` 配套文件。现在 Gateway 通过自己写入时持有的文件 + 描述符，转换 uploads 之外的私有副本；客户端则转换调用方提供的源文件。([#5611]) - **客户端：** `DeerFlowClient.upload_files` 不再写穿符号链接。沙箱可写的 uploads 目录中， 若在上传文件名或其 Markdown 配套文件名处放置符号链接，嵌入式客户端此前会覆盖链接指向的宿主 文件并报告成功。现在该文件会被跳过并列入 `skipped_files`，`success` 为 `false`，与 Gateway @@ -3517,3 +3521,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#5534]: https://github.com/bytedance/deer-flow/pull/5534 [#5547]: https://github.com/bytedance/deer-flow/pull/5547 [#5578]: https://github.com/bytedance/deer-flow/pull/5578 +[#5611]: https://github.com/bytedance/deer-flow/pull/5611 diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index a6d36713f..b5ff56f12 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -102,7 +102,7 @@ owner-scoped assistant version selection remains enabled. | **Subagents** (`/api/subagents`) | Admin managed-worker CRUD and listing. | | **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_ready` (whether `lark-cli` will actually be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/config/credentials` - atomically switch the caller's per-user Lark app after validating the new `app_id`/`app_secret` through the official CLI's live tenant-token probe, revoke/remove the previous OAuth tokens, and restore the prior credential tree if the switch fails; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants. Config and auth flows carry a server-issued, per-user generation persisted under the credential lock; a rejected direct switch leaves the current generation unchanged, stale completions return 409, and browser re-registration uses the same token-clearing/revocation transaction as direct credential switches. | | **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data | -| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); non-mounted sandbox sync uses a non-releasing request lease; `GET /list` - list; `DELETE /{filename}` - delete a regular file; a symlink is never followed and 404s like `GET /list` | +| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); non-mounted sandbox sync uses a non-releasing request lease; `GET /list` - list; `DELETE /{filename}` - delete a regular file; a symlink 404s like `GET /list` | | **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - branch a completed assistant turn with a replay checkpoint; inherited titles take next-free displayed sibling suffixes, including explicit/renamed ones, while explicit titles stay unchanged. Durable `branch` admission rejects races. Workspace files are not checkpointed, so the branch only best-effort copies the current workspace when branching from the **latest** turn (`workspace_clone_mode="current_thread_best_effort"`); branching from an older/historical turn skips the copy (`workspace_clone_mode="skipped_historical_turn"`) so the branch never inherits files that only exist in a later timeline. Thread-scoped runtime channels (`sandbox`, `thread_data`) are not copied onto the branch: the parent's `sandbox_id` binds path mappings and the release lifecycle to the parent's workspace, so the branch lazily acquires its own sandbox instead. Branch creation also seeds the new thread's run-event feed from the branch checkpoint's visible messages (`history_seed_mode` in the response): the thread feed reads run_events, not checkpoints, so without the seed the inherited history disappears from the UI after the branch's first run (#4380). Seeded rows are grouped into one synthetic run per inherited turn (`branch-seed-{thread_id}-{n}`, a new turn opening at every persisted human message, including an allowlisted hidden `ask_clarification` reply) because `run_id` is a turn identity to the feed's consumers, not a provenance tag: regenerating an inherited answer supersedes that row's whole `run_id` in `GET /messages/page`, so one shared id for the entire seed deleted the complete inherited history on a branch's first regenerate (#4458); `GET /goal`, `PUT /goal`, `DELETE /goal` - read, set, and clear the active thread goal; `POST /compact` - summarize older active context, deriving memory policy and bucket from the state-producing checkpoint rather than request `agent_name`, and block while a run is in flight; unexpected failures return a generic 500 detail | | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content (`text/html`, `text/xml`, `application/xml`, `text/xsl`, any `+xml` type such as XHTML/SVG; `.skill` members too) is always forced as a download attachment to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update under a request lease. The outputs-only rule is `path_utils.resolve_outputs_confined_path`, shared with IM-channel attachment delivery: it collapses `..` before the prefix check and re-checks the resolved host path against the resolved outputs root, since `resolve_thread_virtual_path` only confines to `user-data/`; a percent-encoded `..` or a symlink planted in `outputs/` must not reach a sibling `uploads/` file. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | diff --git a/backend/app/gateway/upload_ingestion.py b/backend/app/gateway/upload_ingestion.py index 1571b97a2..00715852d 100644 --- a/backend/app/gateway/upload_ingestion.py +++ b/backend/app/gateway/upload_ingestion.py @@ -28,7 +28,11 @@ patches keep binding to the one pipeline both callers use. from __future__ import annotations +import asyncio import logging +import os +import shutil +import tempfile import uuid from collections.abc import AsyncIterator from pathlib import Path @@ -37,7 +41,7 @@ from typing import TYPE_CHECKING, Any from fastapi import HTTPException from deerflow.config.app_config import AppConfig -from deerflow.utils.file_io import run_file_io +from deerflow.utils.file_io import await_drained, run_file_io if TYPE_CHECKING: from fastapi import Request @@ -48,6 +52,58 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _close_fd(fd: int | None) -> None: + """Close a conversion-source descriptor, tolerating an already-closed one.""" + if fd is None: + return + try: + os.close(fd) + except OSError: + logger.warning("Failed to close upload conversion descriptor", exc_info=True) + + +def _close_abandoned_fd(duplication: asyncio.Future) -> None: + """Close a descriptor whose owner was cancelled before it could take it.""" + if duplication.cancelled() or duplication.exception() is not None: + return + _close_fd(duplication.result()) + + +async def _dup_for_conversion(fileno: int) -> int: + """Duplicate *fileno* off-thread so cancellation cannot strand the copy. + + ``run_file_io`` cannot interrupt its worker: cancelling the await only + abandons the result, and here that result is an open descriptor nothing + would ever close — repeated cancellations would exhaust the Gateway's + descriptor limit and pin the staged bytes of every unlinked upload. The + duplication therefore runs as its own task, shielded from the caller's + cancellation, and closes itself if the caller is gone by the time the + worker finishes. + """ + duplication = asyncio.ensure_future(run_file_io(os.dup, fileno)) + try: + return await asyncio.shield(duplication) + except BaseException: + duplication.add_done_callback(_close_abandoned_fd) + raise + + +def _copy_fd_to_path(fd: int, dest: Path) -> None: + """Copy the bytes behind *fd* to *dest*, then close *fd*. + + Reads through the descriptor, not the name it was committed under, so the + copy is the content this request wrote even if the name has since been + replaced. + """ + try: + os.lseek(fd, 0, os.SEEK_SET) + with open(dest, "wb") as out: + while chunk := os.read(fd, 1 << 20): + out.write(chunk) + finally: + _close_fd(fd) + + def _uploads() -> Any: """Return the uploads router module (late binding — see module docstring).""" from app.gateway.routers import uploads @@ -192,6 +248,7 @@ class ThreadUploadIngestionService: file_size = 0 upload_temp = None + convert_source_fd: int | None = None try: upload_temp = await run_file_io(uploads._prepare_upload_destination, self._uploads_dir, safe_filename) async for chunk in chunks: @@ -202,6 +259,13 @@ class ThreadUploadIngestionService: if self._total_size > self._limits.max_total_size: raise HTTPException(status_code=413, detail="Total upload size too large") await run_file_io(uploads._write_upload_chunk, upload_temp, chunk) + if self._auto_convert and Path(safe_filename).suffix.lower() in uploads.CONVERTIBLE_EXTENSIONS: + # Conversion must read the bytes this request staged. Once the + # name is committed a sandbox process can replace it with a + # symlink, and converting by name would then pull a host file + # into this thread's uploads. A descriptor on the staged inode + # cannot be redirected that way. + convert_source_fd = await _dup_for_conversion(upload_temp.handle.fileno()) # Link-commit with collision retry: the FileExistsError arm # leaves the staged part in place for the retry under the next # suffix (the handle's second close is idempotent). @@ -213,10 +277,16 @@ class ThreadUploadIngestionService: safe_filename = uploads.claim_unique_filename(safe_filename, self._seen_filenames) upload_temp = None except uploads.UnsafeUploadPathError as exc: + _close_fd(convert_source_fd) if upload_temp is not None: await run_file_io(uploads._abort_upload_temp, upload_temp) raise UnsafeUploadDestinationError(safe_filename) from exc - except Exception: + except BaseException: + # BaseException, not Exception: a cancellation between the + # duplication and the commit would otherwise leave the descriptor + # open, and nothing downstream owns it yet. + _close_fd(convert_source_fd) + convert_source_fd = None if upload_temp is not None: await run_file_io(uploads._abort_upload_temp, upload_temp) raise @@ -237,7 +307,7 @@ class ThreadUploadIngestionService: file_info["original_filename"] = original_filename logger.info(f"Saved file: {safe_filename} ({file_size} bytes) to {file_info['path']}") - if self._auto_convert and file_path.suffix.lower() in uploads.CONVERTIBLE_EXTENSIONS: + if convert_source_fd is not None: # The companion gets the same atomic no-overwrite commit as the # original: staged under the hidden .part pattern, link-committed # with next-suffix retry — conversion can never silently truncate @@ -246,12 +316,35 @@ class ThreadUploadIngestionService: provisional_md_name = Path(safe_filename).with_suffix(".md").name unique_md_name = uploads.claim_unique_filename(provisional_md_name, self._seen_filenames) md_staging = self._uploads_dir / f"{uploads.UPLOAD_STAGING_PREFIX}{uuid.uuid4().hex}{uploads.UPLOAD_STAGING_SUFFIX}" + # The staged bytes are copied out of the sandbox-writable tree and + # converted there; the uploads dir only ever receives the result. + # Creating that directory belongs inside the cleanup scope: it can + # fail on its own (a full or unwritable temporary filesystem), and + # the descriptor is already owned here — leaking it would also hold + # the unlinked staged bytes until the process exits. + private_dir: Path | None = None try: - md_staged = await uploads.convert_file_to_markdown(file_path, output_path=md_staging) + private_dir = Path(await run_file_io(tempfile.mkdtemp, "-deerflow-convert")) + conversion_source = private_dir / safe_filename + # Hand the descriptor over before the call: the copy closes it + # even when it fails, so this scope must not close it again and + # risk closing an unrelated descriptor that reused the number. + # await_drained, not a bare await: a cancelled await would + # cancel the queued executor job before its worker — and its + # closing finally — ever ran, and draining also keeps the + # worker from writing into a private directory this scope has + # already removed. + staged_fd, convert_source_fd = convert_source_fd, None + await await_drained(run_file_io(_copy_fd_to_path, staged_fd, conversion_source)) + md_staged = await uploads.convert_file_to_markdown(conversion_source, output_path=md_staging) except Exception: self._seen_filenames.discard(unique_md_name) await run_file_io(md_staging.unlink, True) raise + finally: + _close_fd(convert_source_fd) + if private_dir is not None: + await run_file_io(shutil.rmtree, private_dir, True) if not md_staged: # Conversion failed and wrote nothing (or a partial staged # file, removed here): release the claim; holding it would diff --git a/backend/docs/FILE_UPLOAD.md b/backend/docs/FILE_UPLOAD.md index 1d02630e9..d506cd6ec 100644 --- a/backend/docs/FILE_UPLOAD.md +++ b/backend/docs/FILE_UPLOAD.md @@ -237,6 +237,7 @@ backend/.deer-flow/threads/ - 文件名安全性：系统会自动验证文件路径，防止目录遍历攻击 - 删除只作用于普通文件：上传目录中的符号链接不会被跟随，删除请求按文件不存在（404）处理 - 上传（HTTP 与嵌入式 `DeerFlowClient`）不会写穿符号链接：目标名已是符号链接的文件会被跳过并列入 `skipped_files`，转换生成的 Markdown 也不会写入同名符号链接 +- 转换读取的是本次上传写入的字节，而非落盘后的文件名：HTTP 上传在 uploads 之外的私有副本上转换，嵌入式客户端转换调用方提供的源文件，因此沙箱替换该文件名无法让宿主文件内容被转换进 uploads - 线程隔离：每个线程的上传文件相互隔离，无法跨线程访问 - 自动文档转换默认关闭；如需启用，需在 `config.yaml` 中显式设置 `uploads.auto_convert_documents: true` diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 33b333f64..c885f6777 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -1687,14 +1687,18 @@ class DeerFlowClient: provisional_md_name = Path(dest_name).with_suffix(".md").name unique_md_name = claim_unique_filename(provisional_md_name, seen_names) try: - # Convert outside the sandbox-writable uploads dir, then - # publish without following a symlink at the companion name. + # Convert the caller's own file, not the copy that just + # landed in the sandbox-writable uploads dir: a sandbox + # that swaps that name for a symlink would otherwise have + # a host file converted into this thread's uploads. Write + # the result outside uploads too, then publish it without + # following a symlink at the companion name. with tempfile.TemporaryDirectory() as md_dir: md_output = Path(md_dir) / unique_md_name if conversion_pool is not None: - converted = conversion_pool.submit(_convert_in_thread, dest, md_output).result() + converted = conversion_pool.submit(_convert_in_thread, src_path, md_output).result() else: - converted = asyncio.run(convert_file_to_markdown(dest, output_path=md_output)) + converted = asyncio.run(convert_file_to_markdown(src_path, output_path=md_output)) md_path = None if converted is not None: # copy, not write_bytes: the companion keeps the diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index 78c7d9bff..fd472114e 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -2712,6 +2712,83 @@ class TestUploads: assert (uploads_dir / "a.md").read_text(encoding="utf-8") == "FROM:a.pdf" assert not (uploads_dir / "a_1.md").exists() + def test_upload_files_converts_the_source_not_the_landed_copy(self, client): + """A sandbox swapping the landed upload must not redirect conversion at a host file.""" + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + host_file = tmp_path / "host-secret.pdf" + host_file.write_bytes(b"HOST SECRET") + pdf = tmp_path / "report.pdf" + pdf.write_bytes(b"pdf-bytes") + + async def racing_convert(path: Path, output_path: Path | None = None) -> Path: + # The sandbox wins the race: the landed upload now points outside uploads. + landed = uploads_dir / "report.pdf" + if landed.exists() and not landed.is_symlink(): + landed.unlink() + try: + landed.symlink_to(host_file) + except OSError as exc: + if getattr(exc, "winerror", None) == 1314: + pytest.skip("symlink creation requires Developer Mode or elevated privileges on Windows") + raise + md_path = output_path if output_path is not None else path.with_suffix(".md") + md_path.write_bytes(b"CONVERTED:" + path.read_bytes()) + return md_path + + with ( + patch("deerflow.client.get_uploads_dir", return_value=uploads_dir), + patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir), + patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}), + patch("deerflow.utils.file_conversion.convert_file_to_markdown", side_effect=racing_convert), + ): + result = client.upload_files("thread-1", [pdf]) + + companion = uploads_dir / result["files"][0]["markdown_file"] + assert companion.read_bytes() == b"CONVERTED:pdf-bytes" + assert b"HOST SECRET" not in companion.read_bytes() + + def test_upload_files_converts_the_source_inside_an_event_loop_too(self, client): + """The pooled conversion branch reads the source file as well.""" + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + host_file = tmp_path / "host-secret.pdf" + host_file.write_bytes(b"HOST SECRET") + pdf = tmp_path / "report.pdf" + pdf.write_bytes(b"pdf-bytes") + + async def racing_convert(path: Path, output_path: Path | None = None) -> Path: + landed = uploads_dir / "report.pdf" + if landed.exists() and not landed.is_symlink(): + landed.unlink() + try: + landed.symlink_to(host_file) + except OSError as exc: + if getattr(exc, "winerror", None) == 1314: + pytest.skip("symlink creation requires Developer Mode or elevated privileges on Windows") + raise + md_path = output_path if output_path is not None else path.with_suffix(".md") + md_path.write_bytes(b"CONVERTED:" + path.read_bytes()) + return md_path + + async def call_upload() -> dict: + return client.upload_files("thread-async", [pdf]) + + with ( + patch("deerflow.client.get_uploads_dir", return_value=uploads_dir), + patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir), + patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}), + patch("deerflow.utils.file_conversion.convert_file_to_markdown", side_effect=racing_convert), + ): + result = asyncio.run(call_upload()) + + companion = uploads_dir / result["files"][0]["markdown_file"] + assert companion.read_bytes() == b"CONVERTED:pdf-bytes" + def test_upload_files_rejects_reuploading_a_file_already_in_the_thread(self, client): """Uploading an existing upload onto itself must not destroy its bytes.""" with tempfile.TemporaryDirectory() as tmp: diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index 46252f818..28b0431c3 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -12,6 +12,7 @@ from _router_auth_helpers import call_unwrapped, make_authed_test_app from fastapi import HTTPException, UploadFile from fastapi.testclient import TestClient +from app.gateway import upload_ingestion from app.gateway.deps import get_config from app.gateway.routers import uploads from deerflow.sandbox.lease import get_sandbox_lease_manager @@ -935,6 +936,316 @@ def test_upload_files_oversized_replacement_preserves_existing_regular_file(tmp_ assert [path.name for path in thread_uploads_dir.iterdir()] == ["a.txt"] +def test_upload_files_converts_the_bytes_it_wrote_not_the_committed_name(tmp_path): + """A sandbox swapping the landed upload must not redirect conversion at a host file.""" + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + host_file = tmp_path / "host-secret.pdf" + host_file.write_bytes(b"HOST SECRET") + + provider = MagicMock() + provider.uses_thread_data_mounts = True + + async def racing_convert(file_path: Path, output_path: Path | None = None) -> Path: + # The sandbox wins the race: the committed name now points outside uploads. + landed = thread_uploads_dir / "report.pdf" + if landed.exists() and not landed.is_symlink(): + landed.unlink() + _symlink_to_or_skip(landed, host_file) + md_path = output_path if output_path is not None else file_path.with_suffix(".md") + md_path.write_bytes(b"CONVERTED:" + file_path.read_bytes()) + return md_path + + with ( + patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(uploads, "convert_file_to_markdown", AsyncMock(side_effect=racing_convert)), + ): + file = ChunkedUpload("report.pdf", [b"pdf-bytes"]) + result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-race", request=MagicMock(), files=[file], config=SimpleNamespace())) + + companion = thread_uploads_dir / result.files[0].markdown_file + assert companion.read_bytes() == b"CONVERTED:pdf-bytes" + assert b"HOST SECRET" not in companion.read_bytes() + + +def test_upload_files_conversion_source_survives_a_swap_before_it_is_read(tmp_path): + """The swap lands after the upload commits and before conversion reads it.""" + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + host_file = tmp_path / "host-secret.pdf" + host_file.write_bytes(b"HOST SECRET") + + provider = MagicMock() + provider.uses_thread_data_mounts = True + real_mkdtemp = upload_ingestion.tempfile.mkdtemp + + def swap_then_mkdtemp(*args, **kwargs): + # Runs between the link-commit and the read of the staged bytes. + landed = thread_uploads_dir / "report.pdf" + if landed.exists() and not landed.is_symlink(): + landed.unlink() + _symlink_to_or_skip(landed, host_file) + return real_mkdtemp(*args, **kwargs) + + async def fake_convert(file_path: Path, output_path: Path | None = None) -> Path: + md_path = output_path if output_path is not None else file_path.with_suffix(".md") + md_path.write_bytes(b"CONVERTED:" + file_path.read_bytes()) + return md_path + + with ( + patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(uploads, "convert_file_to_markdown", AsyncMock(side_effect=fake_convert)), + patch.object(upload_ingestion.tempfile, "mkdtemp", side_effect=swap_then_mkdtemp), + ): + file = ChunkedUpload("report.pdf", [b"pdf-bytes"]) + result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-race2", request=MagicMock(), files=[file], config=SimpleNamespace())) + + companion = thread_uploads_dir / result.files[0].markdown_file + assert companion.read_bytes() == b"CONVERTED:pdf-bytes" + assert b"HOST SECRET" not in companion.read_bytes() + + +def test_upload_files_closes_conversion_descriptor_when_private_dir_fails(tmp_path): + """A failed temp-dir creation must not strand the duplicated descriptor.""" + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + + provider = MagicMock() + provider.uses_thread_data_mounts = True + + duplicated: list[int] = [] + closed: list[int] = [] + real_dup, real_close = os.dup, os.close + + def tracking_dup(fd: int) -> int: + new_fd = real_dup(fd) + duplicated.append(new_fd) + return new_fd + + def tracking_close(fd: int) -> None: + closed.append(fd) + real_close(fd) + + with ( + patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(uploads, "convert_file_to_markdown", AsyncMock()), + patch.object(upload_ingestion.os, "dup", side_effect=tracking_dup), + patch.object(upload_ingestion.os, "close", side_effect=tracking_close), + patch.object(upload_ingestion.tempfile, "mkdtemp", side_effect=OSError("No space left on device")), + ): + file = ChunkedUpload("report.pdf", [b"pdf-bytes"]) + with pytest.raises(HTTPException) as exc_info: + asyncio.run(call_unwrapped(uploads.upload_files, "thread-fd", request=MagicMock(), files=[file], config=SimpleNamespace())) + + assert exc_info.value.status_code == 500 + assert len(duplicated) == 1 + assert duplicated[0] in closed + + +def test_upload_files_closes_conversion_descriptor_when_ingestion_is_cancelled(tmp_path): + """Cancelling mid-duplication must not strand the descriptor the worker still produces.""" + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + + provider = MagicMock() + provider.uses_thread_data_mounts = True + + started = threading.Event() + release = threading.Event() + duplicated: list[int] = [] + closed: list[int] = [] + real_dup, real_close = os.dup, os.close + + def slow_dup(fd: int) -> int: + # Allocate first, then stall: the descriptor exists while the caller + # is cancelled, which is exactly what must not be abandoned. + new_fd = real_dup(fd) + duplicated.append(new_fd) + started.set() + release.wait(5) + return new_fd + + def tracking_close(fd: int) -> None: + closed.append(fd) + real_close(fd) + + async def scenario() -> None: + service = upload_ingestion.ThreadUploadIngestionService(request=None, thread_id="thread-cancel", user_id="u", app_config=SimpleNamespace()) + await service.open() + + async def chunks(): + yield b"pdf-bytes" + + task = asyncio.create_task(service.ingest_chunks(chunks(), display_name="report.pdf")) + await asyncio.to_thread(started.wait, 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + # The worker cannot be interrupted; it finishes after the cancellation. + release.set() + for _ in range(100): + if duplicated and duplicated[0] in closed: + break + await asyncio.sleep(0.05) + await service.aclose() + + with ( + patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(uploads, "convert_file_to_markdown", AsyncMock()), + patch.object(upload_ingestion.os, "dup", side_effect=slow_dup), + patch.object(upload_ingestion.os, "close", side_effect=tracking_close), + ): + asyncio.run(scenario()) + + assert len(duplicated) == 1, "the duplication worker must have run" + assert duplicated[0] in closed, "the abandoned descriptor was never closed" + + +def test_upload_files_closes_conversion_descriptor_when_cancelled_during_commit(tmp_path): + """Cancellation after the descriptor is owned must still close it.""" + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + + provider = MagicMock() + provider.uses_thread_data_mounts = True + + started = threading.Event() + release = threading.Event() + duplicated: list[int] = [] + closed: list[int] = [] + real_dup, real_close = os.dup, os.close + real_commit = uploads._commit_upload_temp_no_overwrite + + def tracking_dup(fd: int) -> int: + new_fd = real_dup(fd) + duplicated.append(new_fd) + return new_fd + + def tracking_close(fd: int) -> None: + closed.append(fd) + real_close(fd) + + def slow_commit(*args, **kwargs): + started.set() + release.wait(5) + return real_commit(*args, **kwargs) + + async def scenario() -> None: + service = upload_ingestion.ThreadUploadIngestionService(request=None, thread_id="thread-cancel-commit", user_id="u", app_config=SimpleNamespace()) + await service.open() + + async def chunks(): + yield b"pdf-bytes" + + task = asyncio.create_task(service.ingest_chunks(chunks(), display_name="report.pdf")) + await asyncio.to_thread(started.wait, 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + release.set() + for _ in range(100): + if duplicated and duplicated[0] in closed: + break + await asyncio.sleep(0.05) + await service.aclose() + + with ( + patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(uploads, "convert_file_to_markdown", AsyncMock()), + patch.object(uploads, "_commit_upload_temp_no_overwrite", side_effect=slow_commit), + patch.object(upload_ingestion.os, "dup", side_effect=tracking_dup), + patch.object(upload_ingestion.os, "close", side_effect=tracking_close), + ): + asyncio.run(scenario()) + + assert len(duplicated) == 1 + assert duplicated[0] in closed, "the owned descriptor was not closed on cancellation" + + +def test_upload_files_closes_conversion_descriptor_when_cancelled_while_copy_is_queued(tmp_path): + """A copy job cancelled before its worker starts must not strand the descriptor.""" + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + + provider = MagicMock() + provider.uses_thread_data_mounts = True + + duplicated: list[int] = [] + closed: list[int] = [] + real_dup, real_close = os.dup, os.close + real_run_file_io = upload_ingestion.run_file_io + state: dict[str, object] = {} + + def tracking_dup(fd: int) -> int: + new_fd = real_dup(fd) + duplicated.append(new_fd) + return new_fd + + def tracking_close(fd: int) -> None: + closed.append(fd) + real_close(fd) + + async def queueing_run_file_io(func, *args, **kwargs): + if func is not upload_ingestion._copy_fd_to_path: + return await real_run_file_io(func, *args, **kwargs) + # The pool is busy: the job is queued, not running. + state["queued"].set() + await state["release"].wait() + # Only a job that was never cancelled reaches its worker. + return func(*args, **kwargs) + + async def scenario() -> None: + state["queued"] = asyncio.Event() + state["release"] = asyncio.Event() + service = upload_ingestion.ThreadUploadIngestionService(request=None, thread_id="thread-cancel-copy", user_id="u", app_config=SimpleNamespace()) + await service.open() + + async def chunks(): + yield b"pdf-bytes" + + task = asyncio.create_task(service.ingest_chunks(chunks(), display_name="report.pdf")) + await asyncio.wait_for(state["queued"].wait(), 5) + task.cancel() + state["release"].set() + with pytest.raises(asyncio.CancelledError): + await task + for _ in range(100): + if duplicated and duplicated[0] in closed: + break + await asyncio.sleep(0.05) + await service.aclose() + + with ( + patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(uploads, "convert_file_to_markdown", AsyncMock()), + patch.object(upload_ingestion, "run_file_io", side_effect=queueing_run_file_io), + patch.object(upload_ingestion.os, "dup", side_effect=tracking_dup), + patch.object(upload_ingestion.os, "close", side_effect=tracking_close), + ): + asyncio.run(scenario()) + + assert len(duplicated) == 1 + assert duplicated[0] in closed, "the descriptor handed to the queued copy was never closed" + + def test_delete_uploaded_file_removes_generated_markdown_companion(tmp_path): thread_uploads_dir = tmp_path / "uploads" thread_uploads_dir.mkdir(parents=True)