mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
This reverts commit 9c071c7053379590b63618498fecd0a626ac95e6.
This commit is contained in:
parent
e9133787ff
commit
e5a72b5009
@ -17,12 +17,10 @@ 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
|
||||
@ -76,7 +74,6 @@ 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,
|
||||
@ -1807,101 +1804,73 @@ class DeerFlowClient:
|
||||
|
||||
try:
|
||||
for src_path, dest_name in resolved_files:
|
||||
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:
|
||||
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)
|
||||
try:
|
||||
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
|
||||
# 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, upload_source, md_output).result()
|
||||
converted = conversion_pool.submit(_convert_in_thread, src_path, md_output).result()
|
||||
else:
|
||||
converted = asyncio.run(convert_file_to_markdown(upload_source, 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
|
||||
# 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)
|
||||
|
||||
@ -43,10 +43,6 @@ 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`.
|
||||
|
||||
@ -2,13 +2,10 @@
|
||||
|
||||
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
|
||||
@ -22,7 +19,6 @@ 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
|
||||
@ -3238,100 +3234,6 @@ 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:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user