diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index d467d54d06..08cfc23bdd 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -17,10 +17,12 @@ Usage: import asyncio import concurrent.futures +import contextlib import copy import logging import mimetypes import os +import shutil import tempfile import uuid from collections.abc import Generator, Iterator, Mapping, Sequence @@ -74,6 +76,7 @@ from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, bind_trace_id, e from deerflow.tracing import build_tracing_callbacks, inject_langfuse_metadata from deerflow.uploads.manager import ( UnsafeUploadPathError, + _reject_same_file, claim_unique_filename, copy_upload_file_no_symlink, delete_file_safe, @@ -1804,73 +1807,101 @@ class DeerFlowClient: try: for src_path, dest_name in resolved_files: - try: - dest = copy_upload_file_no_symlink(uploads_dir, dest_name, src_path) - except UnsafeUploadPathError: - logger.warning("Skipping upload with unsafe destination: %s", dest_name) - skipped_files.append(dest_name) - continue - invalidate_overwritten_upload(dest) - - info: dict[str, Any] = { - "filename": dest_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), - } - if dest_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 - unique_md_name = claim_unique_filename(provisional_md_name, seen_names) + convertible = src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS + # Snapshot a convertible source before writing anything: the + # saved upload and its Markdown companion must come from one + # read of the caller's file, which can change between the + # upload copy and the conversion (#6600). The snapshot stays + # in a private temp dir, so conversion still never reopens a + # name inside the sandbox-writable uploads dir (#5611). + work_dir_manager = tempfile.TemporaryDirectory() if convertible else contextlib.nullcontext() + with work_dir_manager as work_dir: + upload_source = src_path + if convertible: + # Open the source once and pin both the same-file + # rejection and the snapshot copy to that descriptor: + # a name swapped in between two path resolutions would + # otherwise split identity from content (#5611 shape). + # copyfileobj (not copy2) reads through the held fd, so + # a source readable only via ACL keeps working; mode + # bits do not survive onto the snapshot, which is fine + # since the snapshot is private to this process. The + # mtime is restored from the same fstat so the upload + # copy keeps the source's timestamps. + with open(src_path, "rb") as src_fh: + src_stat = os.fstat(src_fh.fileno()) + _reject_same_file(uploads_dir, dest_name, src_path, src_stat) + upload_source = Path(work_dir) / dest_name + with open(upload_source, "wb") as snap_fh: + shutil.copyfileobj(src_fh, snap_fh) + os.utime(upload_source, ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns)) try: - # 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 + dest = copy_upload_file_no_symlink(uploads_dir, dest_name, upload_source) + except UnsafeUploadPathError: + logger.warning("Skipping upload with unsafe destination: %s", dest_name) + skipped_files.append(dest_name) + continue + invalidate_overwritten_upload(dest) + + info: dict[str, Any] = { + "filename": dest_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), + } + if dest_name != src_path.name: + info["original_filename"] = src_path.name + + if convertible: + # 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 + unique_md_name = claim_unique_filename(provisional_md_name, seen_names) + try: + # Convert the private snapshot, never the copy that + # 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. + md_output = Path(work_dir) / unique_md_name if conversion_pool is not None: - converted = conversion_pool.submit(_convert_in_thread, src_path, md_output).result() + converted = conversion_pool.submit(_convert_in_thread, upload_source, md_output).result() else: - converted = asyncio.run(convert_file_to_markdown(src_path, output_path=md_output)) + converted = asyncio.run(convert_file_to_markdown(upload_source, output_path=md_output)) md_path = None if converted is not None: # copy, not write_bytes: the companion keeps the # converter's permissions, so a sandbox running as # another uid can still read it. md_path = copy_upload_file_no_symlink(uploads_dir, unique_md_name, converted) - except UnsafeUploadPathError: - logger.warning("Skipping markdown companion with unsafe destination: %s", unique_md_name) - md_path = None - except Exception: - logger.warning( - "Failed to convert %s to markdown", - src_path.name, - exc_info=True, - ) - md_path = None + except UnsafeUploadPathError: + logger.warning("Skipping markdown companion with unsafe destination: %s", unique_md_name) + md_path = None + except Exception: + logger.warning( + "Failed to convert %s to markdown", + src_path.name, + exc_info=True, + ) + md_path = None - if md_path is not None: - register_companion(dest, md_path) - info["markdown_file"] = md_path.name - info["markdown_path"] = str(uploads_dir / md_path.name) - info["markdown_virtual_path"] = upload_virtual_path(md_path.name) - info["markdown_artifact_url"] = upload_artifact_url(thread_id, md_path.name) - else: - # No companion was written, so release the claim; - # holding it would rename a later same-stem upload - # against a name this request never filled. - seen_names.discard(unique_md_name) + if md_path is not None: + register_companion(dest, md_path) + info["markdown_file"] = md_path.name + info["markdown_path"] = str(uploads_dir / md_path.name) + info["markdown_virtual_path"] = upload_virtual_path(md_path.name) + info["markdown_artifact_url"] = upload_artifact_url(thread_id, md_path.name) + else: + # No companion was written, so release the claim; + # holding it would rename a later same-stem upload + # against a name this request never filled. + seen_names.discard(unique_md_name) - uploaded_files.append(info) + uploaded_files.append(info) finally: if conversion_pool is not None: conversion_pool.shutdown(wait=True) diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index 09d799a026..28c62ea213 100644 --- a/backend/tests/AGENTS.md +++ b/backend/tests/AGENTS.md @@ -43,6 +43,10 @@ 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. +Embedded upload source-parity regressions overwrite the caller's file before a +converter reading that path consumes it. The test must reject conversion from +the original path even when the upload copy still uses a private snapshot. + 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 55ef31f10e..f59965ba97 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -2,10 +2,13 @@ import asyncio import concurrent.futures +import getpass import json import os import shutil import stat +import subprocess +import sys import tempfile import zipfile from enum import Enum @@ -19,6 +22,7 @@ from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, Sys from langchain_core.tools import StructuredTool from test_mcp_cache_reconciliation import reconciler # noqa: F401 - fixture re-export +import deerflow.client as client_module from app.gateway.routers.mcp import McpConfigResponse from app.gateway.routers.memory import MemoryConfigResponse, MemoryStatusResponse from app.gateway.routers.models import ModelResponse, ModelsListResponse @@ -3234,6 +3238,100 @@ class TestUploads: assert companion.read_bytes() == b"CONVERTED:pdf-bytes" assert b"HOST SECRET" not in companion.read_bytes() + def test_upload_files_conversion_matches_the_saved_upload_version(self, client): + """The upload copy and its companion must read the same source version (#6600).""" + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + source = tmp_path / "report.pdf" + source.write_bytes(b"VERSION A") + + real_copy = client_module.copy_upload_file_no_symlink + + def copy_then_overwrite(base, name, src): + dest = real_copy(base, name, src) + if Path(src) == source: + # The caller's file changes between the copy and conversion. + source.write_bytes(b"VERSION B") + return dest + + async def fake_convert(path, output_path=None): + # Rewrite the caller's file too, so a regression that reads + # src_path again during conversion (instead of the snapshot) + # goes red even though the copy-time hook no longer fires on + # the fixed code path. + if Path(path) == source: + source.write_bytes(b"VERSION B") + 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.client.copy_upload_file_no_symlink", + side_effect=copy_then_overwrite, + ), + patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}), + patch( + "deerflow.utils.file_conversion.convert_file_to_markdown", + side_effect=fake_convert, + ), + ): + result = client.upload_files("thread-1", [source]) + + saved = uploads_dir / result["files"][0]["filename"] + companion = uploads_dir / result["files"][0]["markdown_file"] + assert saved.read_bytes() == b"VERSION A" + assert companion.read_bytes() == b"CONVERTED:VERSION A" + + def test_upload_files_acl_only_readable_source_survives(self, client): + """A source readable only via ACL must upload and convert (#6621 review). + + copy2 drops ACLs; on macOS a mode-000 file with an ACL read grant is + readable, but the snapshot copy made from it was not. Reading through + one held descriptor keeps ACL-authorized sources working. + """ + if sys.platform != "darwin": + pytest.skip("chmod +a ACL grants are macOS-only") + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + source = tmp_path / "report.pdf" + source.write_bytes(b"VERSION A") + source.chmod(0o000) + subprocess.run( + ["chmod", "+a", f"user:{getpass.getuser()} allow read", str(source)], + check=True, + ) + try: + + async def fake_convert(path, output_path=None): + 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=fake_convert, + ), + ): + result = client.upload_files("thread-1", [source]) + + saved = uploads_dir / result["files"][0]["filename"] + companion = uploads_dir / result["files"][0]["markdown_file"] + assert saved.read_bytes() == b"VERSION A" + assert companion.read_bytes() == b"CONVERTED:VERSION A" + finally: + source.chmod(0o644) + 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: