diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 59d88e299..80f0f23b5 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -705,6 +705,7 @@ Multi-file upload with automatic document conversion: - Reuses one conversion worker per request when called from an active event loop - Files stored in thread-isolated directories under the resolving user's bucket (`users/{user_id}/threads/{thread_id}/user-data/uploads`). For IM channels the owner is threaded explicitly via the `user_id=` kwarg (see IM Channels → Owner-scoped file storage); HTTP/embedded callers resolve it from `get_effective_user_id()` - Duplicate filenames in a single upload request are auto-renamed with `_N` suffixes so later files do not truncate earlier files +- Gateway HTTP uploads stage bytes as `.upload-*.part` files and atomically replace the destination only after size validation. These staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools, and swept on Gateway startup if a hard crash leaves one behind. - Agent receives uploaded file list via `UploadsMiddleware` See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details. diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 528d2cbe4..9633f3708 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -31,6 +31,7 @@ from app.gateway.routers import ( ) from deerflow.config import app_config as deerflow_app_config from deerflow.config.app_config import apply_logging_level +from deerflow.uploads.manager import cleanup_stale_upload_staging_files AppConfig = deerflow_app_config.AppConfig get_app_config = deerflow_app_config.get_app_config @@ -207,6 +208,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.warning("tiktoken warm-up skipped", exc_info=True) + try: + removed_upload_staging_files = await asyncio.to_thread(cleanup_stale_upload_staging_files) + if removed_upload_staging_files: + logger.info("Removed %d stale upload staging file(s)", removed_upload_staging_files) + except Exception: + logger.warning("Upload staging file cleanup skipped", exc_info=True) + # Initialize LangGraph runtime components (StreamBridge, RunManager, checkpointer, store) async with langgraph_runtime(app, startup_config): logger.info("LangGraph runtime initialised") diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py index 6ecd547fd..267b9f8e1 100644 --- a/backend/app/gateway/routers/uploads.py +++ b/backend/app/gateway/routers/uploads.py @@ -3,6 +3,8 @@ import logging import os import stat +import tempfile +from pathlib import Path from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile from pydantic import BaseModel, Field @@ -14,6 +16,8 @@ from deerflow.config.paths import get_paths from deerflow.runtime.user_context import get_effective_user_id from deerflow.sandbox.sandbox_provider import SandboxProvider, get_sandbox_provider from deerflow.uploads.manager import ( + UPLOAD_STAGING_PREFIX, + UPLOAD_STAGING_SUFFIX, PathTraversalError, UnsafeUploadPathError, claim_unique_filename, @@ -23,9 +27,9 @@ from deerflow.uploads.manager import ( get_uploads_dir, list_files_in_dir, normalize_filename, - open_upload_file_no_symlink, upload_artifact_url, upload_virtual_path, + validate_upload_destination, ) from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown @@ -173,7 +177,11 @@ async def _write_upload_file_with_limits( total_size: int, ) -> tuple[os.PathLike[str] | str, int, int]: file_size = 0 - file_path, fh = open_upload_file_no_symlink(uploads_dir, display_filename) + uploads_dir_path = Path(uploads_dir) + file_path = validate_upload_destination(uploads_dir_path, display_filename) + temp_fd, temp_path_str = tempfile.mkstemp(prefix=UPLOAD_STAGING_PREFIX, suffix=UPLOAD_STAGING_SUFFIX, dir=uploads_dir_path) + temp_path = Path(temp_path_str) + fh = os.fdopen(temp_fd, "wb") try: while chunk := await file.read(UPLOAD_CHUNK_SIZE): file_size += len(chunk) @@ -186,12 +194,20 @@ async def _write_upload_file_with_limits( except Exception: fh.close() try: - os.unlink(file_path) + os.unlink(temp_path) except FileNotFoundError: pass raise else: fh.close() + try: + os.replace(temp_path, file_path) + except Exception: + try: + os.unlink(temp_path) + except FileNotFoundError: + pass + raise return file_path, file_size, total_size diff --git a/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py index 1d0d7a03f..480749f36 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py @@ -12,6 +12,7 @@ from langgraph.runtime import Runtime from deerflow.config.paths import Paths, get_paths from deerflow.runtime.user_context import get_effective_user_id +from deerflow.uploads.manager import is_upload_staging_file from deerflow.utils.file_conversion import extract_outline from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, message_content_to_text @@ -172,7 +173,7 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]): if not isinstance(f, dict): continue filename = f.get("filename") or "" - if not filename or Path(filename).name != filename: + if not filename or Path(filename).name != filename or is_upload_staging_file(filename): continue if uploads_dir is not None and not (uploads_dir / filename).is_file(): continue @@ -234,6 +235,8 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]): historical_files: list[dict] = [] if uploads_dir and uploads_dir.exists(): for file_path in sorted(uploads_dir.iterdir()): + if is_upload_staging_file(file_path.name): + continue if file_path.is_file() and file_path.name not in new_filenames: stat = file_path.stat() outline, preview = _extract_outline_for_file(file_path) diff --git a/backend/packages/harness/deerflow/sandbox/search.py b/backend/packages/harness/deerflow/sandbox/search.py index a5cf1314e..b4b7b9608 100644 --- a/backend/packages/harness/deerflow/sandbox/search.py +++ b/backend/packages/harness/deerflow/sandbox/search.py @@ -43,6 +43,7 @@ IGNORE_PATTERNS = [ "*.log", "*.tmp", "*.temp", + ".upload-*.part", "*.bak", "*.cache", ".cache", diff --git a/backend/packages/harness/deerflow/uploads/__init__.py b/backend/packages/harness/deerflow/uploads/__init__.py index d38859798..f0747f9c6 100644 --- a/backend/packages/harness/deerflow/uploads/__init__.py +++ b/backend/packages/harness/deerflow/uploads/__init__.py @@ -1,10 +1,14 @@ from .manager import ( + UPLOAD_STAGING_PREFIX, + UPLOAD_STAGING_SUFFIX, PathTraversalError, claim_unique_filename, + cleanup_stale_upload_staging_files, delete_file_safe, enrich_file_listing, ensure_uploads_dir, get_uploads_dir, + is_upload_staging_file, list_files_in_dir, normalize_filename, upload_artifact_url, @@ -18,7 +22,11 @@ __all__ = [ "ensure_uploads_dir", "normalize_filename", "PathTraversalError", + "UPLOAD_STAGING_PREFIX", + "UPLOAD_STAGING_SUFFIX", "claim_unique_filename", + "cleanup_stale_upload_staging_files", + "is_upload_staging_file", "validate_path_traversal", "list_files_in_dir", "delete_file_safe", diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index 3605cd5b2..691cbc478 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -5,6 +5,7 @@ Both Gateway and Client delegate to these functions. """ import errno +import logging import os import re import stat @@ -23,8 +24,12 @@ class UnsafeUploadPathError(ValueError): """Raised when an upload destination is not a safe regular file path.""" +logger = logging.getLogger(__name__) + # thread_id must be alphanumeric, hyphens, underscores, or dots only. _SAFE_THREAD_ID = re.compile(r"^[a-zA-Z0-9._-]+$") +UPLOAD_STAGING_PREFIX = ".upload-" +UPLOAD_STAGING_SUFFIX = ".part" def validate_thread_id(thread_id: str) -> None: @@ -103,6 +108,11 @@ def claim_unique_filename(name: str, seen: set[str]) -> str: return candidate +def is_upload_staging_file(filename: str) -> bool: + """Return whether *filename* is a transient Gateway upload staging file.""" + return filename.startswith(UPLOAD_STAGING_PREFIX) and filename.endswith(UPLOAD_STAGING_SUFFIX) + + def validate_path_traversal(path: Path, base: Path) -> None: """Verify that *path* is inside *base*. @@ -115,6 +125,56 @@ def validate_path_traversal(path: Path, base: Path) -> None: raise PathTraversalError("Path traversal detected") from None +def validate_upload_destination(base_dir: Path, filename: str) -> Path: + """Validate an upload destination without mutating an existing file.""" + safe_name = normalize_filename(filename) + dest = base_dir / safe_name + + try: + st = os.lstat(dest) + except FileNotFoundError: + st = None + + if st is not None and not stat.S_ISREG(st.st_mode): + raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}") + if st is not None and st.st_nlink > 1: + raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}") + + validate_path_traversal(dest, base_dir) + return dest + + +def _iter_upload_dirs(base_dir: Path): + yield from base_dir.glob("threads/*/user-data/uploads") + yield from base_dir.glob("users/*/threads/*/user-data/uploads") + + +def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> int: + """Remove orphaned Gateway upload staging files left by a hard crash.""" + root = Path(base_dir) if base_dir is not None else get_paths().base_dir + removed = 0 + for uploads_dir in _iter_upload_dirs(root): + if not uploads_dir.is_dir(): + continue + try: + with os.scandir(uploads_dir) as entries: + for entry in entries: + if not is_upload_staging_file(entry.name) or not entry.is_file(follow_symlinks=False): + continue + try: + os.unlink(entry.path) + removed += 1 + except FileNotFoundError: + pass + except OSError: + logger.warning("Failed to remove stale upload staging file: %s", entry.path, exc_info=True) + except FileNotFoundError: + continue + except OSError: + logger.warning("Failed to scan uploads directory for stale staging files: %s", uploads_dir, exc_info=True) + return removed + + def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, object]: """Open an upload destination for safe streaming writes. @@ -128,18 +188,12 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob validation prevents escapes from *base_dir* in both cases. """ safe_name = normalize_filename(filename) - dest = base_dir / safe_name - + dest = validate_upload_destination(base_dir, safe_name) try: st = os.lstat(dest) except FileNotFoundError: st = None - if st is not None and not stat.S_ISREG(st.st_mode): - raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}") - - validate_path_traversal(dest, base_dir) - has_nofollow = hasattr(os, "O_NOFOLLOW") if has_nofollow: @@ -234,6 +288,8 @@ def list_files_in_dir(directory: Path) -> dict: files = [] with os.scandir(directory) as entries: for entry in sorted(entries, key=lambda e: e.name): + if is_upload_staging_file(entry.name): + continue if not entry.is_file(follow_symlinks=False): continue st = entry.stat(follow_symlinks=False) diff --git a/backend/tests/test_gateway_lifespan_shutdown.py b/backend/tests/test_gateway_lifespan_shutdown.py index 4481ed041..c1835bdf7 100644 --- a/backend/tests/test_gateway_lifespan_shutdown.py +++ b/backend/tests/test_gateway_lifespan_shutdown.py @@ -11,6 +11,7 @@ from __future__ import annotations import asyncio from contextlib import asynccontextmanager +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from fastapi import FastAPI @@ -32,17 +33,18 @@ async def _run_lifespan_with_hanging_stop() -> float: await asyncio.sleep(3600) app = FastAPI() + startup_config = SimpleNamespace(log_level="INFO", memory=SimpleNamespace(token_counting="char")) fake_service = MagicMock() fake_service.get_status = MagicMock(return_value={}) - async def fake_start(): + async def fake_start(_startup_config): return fake_service close_oidc_service = AsyncMock() with ( - patch("app.gateway.app.get_app_config"), + patch("app.gateway.app.get_app_config", return_value=startup_config), patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)), patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime), patch("app.gateway.app.auth.close_oidc_service", close_oidc_service), @@ -70,3 +72,40 @@ def test_shutdown_is_bounded_when_channel_stop_hangs(): assert elapsed < _SHUTDOWN_HOOK_TIMEOUT_SECONDS + 2.0, f"Lifespan shutdown took {elapsed:.2f}s; expected <= {_SHUTDOWN_HOOK_TIMEOUT_SECONDS + 2.0:.1f}s" # Lower bound: the wait_for should actually have waited. assert elapsed >= _SHUTDOWN_HOOK_TIMEOUT_SECONDS - 0.5, f"Lifespan exited too quickly ({elapsed:.2f}s); wait_for may not have been invoked." + + +async def _run_lifespan_with_upload_staging_cleanup(): + from app.gateway.app import lifespan + + app = FastAPI() + startup_config = SimpleNamespace(log_level="INFO", memory=SimpleNamespace(token_counting="char")) + fake_service = MagicMock() + fake_service.get_status = MagicMock(return_value={}) + cleanup_upload_staging_files = MagicMock(return_value=2) + close_oidc_service = AsyncMock() + stop_channel_service = AsyncMock() + + async def fake_start(_startup_config): + return fake_service + + with ( + patch("app.gateway.app.get_app_config", return_value=startup_config), + patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)), + patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime), + patch("app.gateway.app.cleanup_stale_upload_staging_files", cleanup_upload_staging_files), + patch("app.gateway.app.auth.close_oidc_service", close_oidc_service), + patch("app.channels.service.start_channel_service", side_effect=fake_start), + patch("app.channels.service.stop_channel_service", stop_channel_service), + ): + async with lifespan(app): + pass + + return cleanup_upload_staging_files, close_oidc_service, stop_channel_service + + +def test_lifespan_sweeps_upload_staging_files_on_startup(): + cleanup_upload_staging_files, close_oidc_service, stop_channel_service = asyncio.run(_run_lifespan_with_upload_staging_cleanup()) + + cleanup_upload_staging_files.assert_called_once_with() + close_oidc_service.assert_awaited_once() + stop_channel_service.assert_awaited_once() diff --git a/backend/tests/test_sandbox_search_tools.py b/backend/tests/test_sandbox_search_tools.py index 88e87a783..30e9d384e 100644 --- a/backend/tests/test_sandbox_search_tools.py +++ b/backend/tests/test_sandbox_search_tools.py @@ -459,3 +459,23 @@ def test_ls_tool_returns_empty_for_empty_directory(tmp_path, monkeypatch) -> Non ) assert result == "(empty)" + + +def test_ls_tool_filters_upload_staging_files(tmp_path, monkeypatch) -> None: + runtime = _make_runtime(tmp_path) + uploads = tmp_path / "uploads" + (uploads / "report.txt").write_text("ready\n", encoding="utf-8") + (uploads / ".upload-active.part").write_text("partial\n", encoding="utf-8") + (uploads / ".upload-note.txt").write_text("intentional\n", encoding="utf-8") + + monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: LocalSandbox(id="local")) + + result = ls_tool.func( + runtime=runtime, + description="list uploads", + path="/mnt/user-data/uploads", + ) + + assert "/mnt/user-data/uploads/report.txt" in result + assert "/mnt/user-data/uploads/.upload-note.txt" in result + assert ".upload-active.part" not in result diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index 733e9e402..28364e951 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -10,6 +10,7 @@ from deerflow.uploads.manager import ( PathTraversalError, UnsafeUploadPathError, claim_unique_filename, + cleanup_stale_upload_staging_files, delete_file_safe, list_files_in_dir, normalize_filename, @@ -187,6 +188,49 @@ class TestListFilesInDir: assert result["count"] == 1 assert result["files"][0]["filename"] == "file.txt" + def test_filters_only_upload_staging_files(self, tmp_path): + (tmp_path / ".env").write_text("intentional dotfile") + (tmp_path / ".upload-active.part").write_text("partial") + (tmp_path / ".upload-note.txt").write_text("intentional upload") + (tmp_path / "draft.part").write_text("intentional upload") + (tmp_path / "visible.txt").write_text("visible") + + result = list_files_in_dir(tmp_path) + + assert result["count"] == 4 + assert [f["filename"] for f in result["files"]] == [".env", ".upload-note.txt", "draft.part", "visible.txt"] + + +# --------------------------------------------------------------------------- +# cleanup_stale_upload_staging_files +# --------------------------------------------------------------------------- + + +class TestCleanupStaleUploadStagingFiles: + def test_removes_only_stale_staging_files_from_all_upload_layouts(self, tmp_path): + legacy_uploads = tmp_path / "threads" / "thread-legacy" / "user-data" / "uploads" + user_uploads = tmp_path / "users" / "owner-1" / "threads" / "thread-owned" / "user-data" / "uploads" + unrelated_uploads = tmp_path / "misc" / "thread-other" / "user-data" / "uploads" + for uploads_dir in (legacy_uploads, user_uploads, unrelated_uploads): + uploads_dir.mkdir(parents=True) + + (legacy_uploads / ".upload-old.part").write_text("legacy partial") + (user_uploads / ".upload-new.part").write_text("user partial") + (unrelated_uploads / ".upload-ignore.part").write_text("outside layout") + (legacy_uploads / ".env").write_text("intentional dotfile") + (legacy_uploads / ".upload-note.txt").write_text("intentional upload") + (legacy_uploads / "draft.part").write_text("intentional upload") + + removed = cleanup_stale_upload_staging_files(tmp_path) + + assert removed == 2 + assert not (legacy_uploads / ".upload-old.part").exists() + assert not (user_uploads / ".upload-new.part").exists() + assert (unrelated_uploads / ".upload-ignore.part").exists() + assert (legacy_uploads / ".env").exists() + assert (legacy_uploads / ".upload-note.txt").exists() + assert (legacy_uploads / "draft.part").exists() + # --------------------------------------------------------------------------- # delete_file_safe diff --git a/backend/tests/test_uploads_middleware_core_logic.py b/backend/tests/test_uploads_middleware_core_logic.py index d0482bb71..b69ed6648 100644 --- a/backend/tests/test_uploads_middleware_core_logic.py +++ b/backend/tests/test_uploads_middleware_core_logic.py @@ -146,6 +146,11 @@ class TestFilesFromKwargs: assert result is not None assert result[0]["size"] == 0 + def test_skips_upload_staging_filenames(self, tmp_path): + mw = _middleware(tmp_path) + msg = _human("hi", files=[{"filename": ".upload-active.part", "size": 5, "path": "/mnt/user-data/uploads/.upload-active.part"}]) + assert mw._files_from_kwargs(msg) is None + # --------------------------------------------------------------------------- # _create_files_message @@ -362,6 +367,22 @@ class TestBeforeAgent: assert "previous messages" in content assert "old.txt" in content + def test_historical_files_ignore_upload_staging_files(self, tmp_path): + mw = _middleware(tmp_path) + uploads_dir = _uploads_dir(tmp_path) + (uploads_dir / "old.txt").write_bytes(b"old") + (uploads_dir / ".upload-active.part").write_bytes(b"partial") + (uploads_dir / ".env").write_bytes(b"intentional") + + msg = _human("go") + result = mw.before_agent(self._state(msg), _runtime()) + + assert result is not None + content = result["messages"][-1].content + assert "old.txt" in content + assert ".env" in content + assert ".upload-active.part" not in content + def test_no_historical_section_when_upload_dir_is_empty(self, tmp_path): mw = _middleware(tmp_path) uploads_dir = _uploads_dir(tmp_path) diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index 7fdac6252..79d42f12a 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -653,6 +653,38 @@ def test_upload_files_overwrites_existing_regular_file(tmp_path): assert existing_file.stat().st_nlink == 1 +def test_upload_files_oversized_replacement_preserves_existing_regular_file(tmp_path): + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + existing_file = thread_uploads_dir / "a.txt" + existing_file.write_bytes(b"original bytes") + + provider = MagicMock() + provider.uses_thread_data_mounts = True + + 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), + ): + file = ChunkedUpload("a.txt", [b"tiny", b"x" * 8]) + + with pytest.raises(HTTPException) as exc_info: + asyncio.run( + call_unwrapped( + uploads.upload_files, + "thread-local", + request=MagicMock(), + files=[file], + config=SimpleNamespace(uploads={"max_file_size": 10}), + ) + ) + + assert exc_info.value.status_code == 413 + assert existing_file.read_bytes() == b"original bytes" + assert [path.name for path in thread_uploads_dir.iterdir()] == ["a.txt"] + + def test_delete_uploaded_file_removes_generated_markdown_companion(tmp_path): thread_uploads_dir = tmp_path / "uploads" thread_uploads_dir.mkdir(parents=True)