534 lines
21 KiB
Python
534 lines
21 KiB
Python
"""Upload router for handling file uploads."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
from weakref import WeakKeyDictionary, WeakValueDictionary
|
|
|
|
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.authz import require_permission
|
|
from app.gateway.deps import get_config
|
|
from app.gateway.routers.upload_policy import LargeFilePolicy, UploadGroups, get_effective_upload_policy
|
|
from deerflow.config.app_config import AppConfig
|
|
from deerflow.config.paths import get_paths
|
|
from deerflow.runtime.thread_paths import aresolve_path_user_id
|
|
from deerflow.sandbox.sandbox_provider import SandboxProvider, get_sandbox_provider
|
|
from deerflow.uploads.manager import (
|
|
PathTraversalError,
|
|
UnsafeUploadPathError,
|
|
delete_file_safe,
|
|
enrich_file_listing,
|
|
ensure_uploads_dir,
|
|
get_uploads_dir,
|
|
list_files_in_dir,
|
|
normalize_filename,
|
|
open_upload_file_no_symlink,
|
|
upload_artifact_url,
|
|
upload_virtual_path,
|
|
validate_path_traversal,
|
|
)
|
|
from deerflow.uploads.policy import classify_upload_filename
|
|
from deerflow.uploads.validation import UploadContentError, validate_upload_content
|
|
from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown_sync
|
|
from deerflow.utils.file_io import FileIOPool, file_io
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/threads/{thread_id}/uploads", tags=["uploads"])
|
|
|
|
UPLOAD_CHUNK_SIZE = 256 * 1024
|
|
|
|
# Upload work must not fill the default executor used by authentication, or the
|
|
# artifact/event pools used by active conversations. Waiting batches use no worker.
|
|
_upload_io = FileIOPool(2, 2, "uploads")
|
|
_upload_locks: WeakKeyDictionary = WeakKeyDictionary()
|
|
|
|
|
|
def _upload_lock(directory: Path) -> asyncio.Lock:
|
|
locks = _upload_locks.setdefault(asyncio.get_running_loop(), WeakValueDictionary())
|
|
key = os.path.normcase(str(directory))
|
|
lock = locks.get(key)
|
|
if lock is None:
|
|
lock = asyncio.Lock()
|
|
locks[key] = lock
|
|
return lock
|
|
|
|
|
|
class _UploadCancelled(Exception):
|
|
"""Cooperative stop inside a worker; the request propagates CancelledError."""
|
|
|
|
|
|
def _check_upload_cancelled(cancelled: threading.Event | None) -> None:
|
|
if cancelled is not None and cancelled.is_set():
|
|
raise _UploadCancelled()
|
|
|
|
|
|
async def _resolve_uploads_path(thread_id: str) -> Path:
|
|
owner = await aresolve_path_user_id(thread_id)
|
|
try:
|
|
return get_paths().sandbox_uploads_dir(thread_id, user_id=owner)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
|
|
class UploadResponse(BaseModel):
|
|
"""Response model for file upload."""
|
|
|
|
success: bool
|
|
files: list[dict[str, str]]
|
|
message: str
|
|
skipped_files: list[str] = Field(default_factory=list)
|
|
policy_revision: int = 0
|
|
|
|
|
|
class UploadLimits(BaseModel):
|
|
"""Application-level upload limits exposed to clients."""
|
|
|
|
max_files: int
|
|
max_file_size: int
|
|
max_total_size: int
|
|
max_files_per_message: int
|
|
max_total_bytes: int
|
|
groups: UploadGroups
|
|
large_file: LargeFilePolicy
|
|
revision: int = 0
|
|
|
|
|
|
def _make_file_sandbox_writable(file_path: os.PathLike[str] | str) -> None:
|
|
"""Ensure uploaded files remain writable when mounted into non-local sandboxes.
|
|
|
|
In AIO sandbox mode, the gateway writes the authoritative host-side file
|
|
first, then the sandbox runtime may rewrite the same mounted path. Granting
|
|
world-writable access here prevents permission mismatches between the
|
|
gateway user and the sandbox runtime user.
|
|
"""
|
|
file_stat = os.lstat(file_path)
|
|
if stat.S_ISLNK(file_stat.st_mode):
|
|
logger.warning("Skipping sandbox chmod for symlinked upload path: %s", file_path)
|
|
return
|
|
|
|
writable_mode = stat.S_IMODE(file_stat.st_mode) | stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH
|
|
chmod_kwargs = {"follow_symlinks": False} if os.chmod in os.supports_follow_symlinks else {}
|
|
os.chmod(file_path, writable_mode, **chmod_kwargs)
|
|
|
|
|
|
def _uses_thread_data_mounts(sandbox_provider: SandboxProvider) -> bool:
|
|
return bool(getattr(sandbox_provider, "uses_thread_data_mounts", False))
|
|
|
|
|
|
def _get_uploads_config_value(app_config: AppConfig, key: str, default: object) -> object:
|
|
"""Read a value from the uploads config, supporting dict and attribute access."""
|
|
uploads_cfg = getattr(app_config, "uploads", None)
|
|
if isinstance(uploads_cfg, dict):
|
|
return uploads_cfg.get(key, default)
|
|
return getattr(uploads_cfg, key, default)
|
|
|
|
|
|
def _get_upload_limits(policy: dict, revision: int = 0) -> UploadLimits:
|
|
groups = UploadGroups.model_validate(policy["groups"])
|
|
max_single = max(group.max_bytes for group in (groups.document, groups.image, groups.spreadsheet))
|
|
return UploadLimits(
|
|
max_files=int(policy["max_files_per_message"]),
|
|
max_file_size=max_single,
|
|
max_total_size=int(policy["max_total_bytes"]),
|
|
max_files_per_message=int(policy["max_files_per_message"]),
|
|
max_total_bytes=int(policy["max_total_bytes"]),
|
|
groups=groups,
|
|
large_file=LargeFilePolicy.model_validate(policy["large_file"]),
|
|
revision=revision,
|
|
)
|
|
|
|
|
|
async def _write_upload_file_with_limits(
|
|
file: UploadFile,
|
|
*,
|
|
uploads_dir: os.PathLike[str] | str,
|
|
display_filename: str,
|
|
max_single_file_size: int,
|
|
max_total_size: int,
|
|
total_size: int,
|
|
) -> tuple[os.PathLike[str] | str, int, int]:
|
|
cancelled = threading.Event()
|
|
return await _upload_io.run_until_complete(
|
|
_copy_upload_file_with_limits,
|
|
file,
|
|
uploads_dir=uploads_dir,
|
|
display_filename=display_filename,
|
|
max_single_file_size=max_single_file_size,
|
|
max_total_size=max_total_size,
|
|
total_size=total_size,
|
|
cancelled=cancelled,
|
|
on_cancel=cancelled.set,
|
|
)
|
|
|
|
|
|
def _copy_upload_file_with_limits(
|
|
file: UploadFile,
|
|
*,
|
|
uploads_dir: os.PathLike[str] | str,
|
|
display_filename: str,
|
|
max_single_file_size: int,
|
|
max_total_size: int,
|
|
total_size: int,
|
|
cancelled: threading.Event | None = None,
|
|
) -> tuple[Path, int, int]:
|
|
"""Copy the parsed multipart spool entirely in the dedicated upload worker."""
|
|
_check_upload_cancelled(cancelled)
|
|
file_size = 0
|
|
file_path, fh = open_upload_file_no_symlink(uploads_dir, display_filename)
|
|
try:
|
|
while chunk := file.file.read(UPLOAD_CHUNK_SIZE):
|
|
_check_upload_cancelled(cancelled)
|
|
file_size += len(chunk)
|
|
total_size += len(chunk)
|
|
if file_size > max_single_file_size:
|
|
raise HTTPException(status_code=413, detail=f"File too large: {display_filename}")
|
|
if total_size > max_total_size:
|
|
raise HTTPException(status_code=413, detail="Total upload size too large")
|
|
fh.write(chunk)
|
|
_check_upload_cancelled(cancelled)
|
|
except Exception:
|
|
fh.close()
|
|
try:
|
|
os.unlink(file_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
raise
|
|
else:
|
|
fh.close()
|
|
return file_path, file_size, total_size
|
|
|
|
|
|
def _commit_staged_files(staging_dir: Path, uploads_dir: Path, names: list[str]) -> list[Path]:
|
|
"""Commit a validated batch and restore every old file on failure."""
|
|
|
|
if len(names) != len(set(names)):
|
|
raise HTTPException(status_code=409, detail="Upload batch contains colliding filenames")
|
|
backup_dir = staging_dir / ".backup"
|
|
backup_dir.mkdir()
|
|
backups: list[tuple[Path, Path]] = []
|
|
committed: list[Path] = []
|
|
try:
|
|
for index, name in enumerate(names):
|
|
staged = staging_dir / name
|
|
destination = uploads_dir / name
|
|
validate_path_traversal(destination, uploads_dir)
|
|
if destination.exists() or destination.is_symlink():
|
|
existing_stat = os.lstat(destination)
|
|
if not stat.S_ISREG(existing_stat.st_mode):
|
|
raise UnsafeUploadPathError(f"Upload destination is not a regular file: {name}")
|
|
backup = backup_dir / f"{index}.bak"
|
|
os.replace(destination, backup)
|
|
backups.append((destination, backup))
|
|
os.replace(staged, destination)
|
|
committed.append(destination)
|
|
except Exception:
|
|
for destination in reversed(committed):
|
|
try:
|
|
os.unlink(destination)
|
|
except FileNotFoundError:
|
|
pass
|
|
for destination, backup in reversed(backups):
|
|
if backup.exists():
|
|
os.replace(backup, destination)
|
|
raise
|
|
return committed
|
|
|
|
|
|
def _rollback_committed_files(staging_dir: Path, uploads_dir: Path, names: list[str]) -> None:
|
|
"""Undo :func:`_commit_staged_files` while its backups still exist."""
|
|
|
|
backup_dir = staging_dir / ".backup"
|
|
errors: list[Exception] = []
|
|
for index, name in reversed(list(enumerate(names))):
|
|
destination = uploads_dir / name
|
|
backup = backup_dir / f"{index}.bak"
|
|
try:
|
|
if destination.exists() or destination.is_symlink():
|
|
os.unlink(destination)
|
|
if backup.exists():
|
|
os.replace(backup, destination)
|
|
except Exception as exc: # best-effort through the entire batch
|
|
errors.append(exc)
|
|
logger.exception("Failed to roll back committed upload %s", destination)
|
|
if errors:
|
|
raise RuntimeError("Failed to fully roll back upload batch") from errors[0]
|
|
|
|
|
|
def _auto_convert_documents_enabled(app_config: AppConfig) -> bool:
|
|
"""Return whether automatic host-side document conversion is enabled.
|
|
|
|
The secure default is disabled unless an operator explicitly opts in via
|
|
uploads.auto_convert_documents in config.yaml.
|
|
"""
|
|
try:
|
|
raw = _get_uploads_config_value(app_config, "auto_convert_documents", False)
|
|
if isinstance(raw, str):
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
return bool(raw)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
@router.post("", response_model=UploadResponse)
|
|
@require_permission("threads", "write", owner_check=True, require_existing=False)
|
|
async def upload_files(
|
|
thread_id: str,
|
|
request: Request,
|
|
files: list[UploadFile] = File(...),
|
|
config: AppConfig = Depends(get_config),
|
|
) -> UploadResponse:
|
|
"""Upload multiple files to a thread's uploads directory."""
|
|
if not files:
|
|
raise HTTPException(status_code=400, detail="No files provided")
|
|
|
|
policy, policy_metadata = await get_effective_upload_policy(request)
|
|
limits = _get_upload_limits(policy, int(policy_metadata.get("revision") or 0))
|
|
if len(files) > limits.max_files:
|
|
raise HTTPException(status_code=413, detail=f"Too many files: maximum is {limits.max_files}")
|
|
|
|
prepared: list[tuple[UploadFile, str, str, int]] = []
|
|
claimed_names: set[str] = set()
|
|
for file in files:
|
|
if not file.filename:
|
|
raise HTTPException(status_code=400, detail="Every upload must have a filename")
|
|
try:
|
|
safe_filename = normalize_filename(file.filename)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
folded_name = safe_filename.casefold()
|
|
if folded_name in claimed_names:
|
|
raise HTTPException(status_code=409, detail=f"Duplicate filename in upload batch: {safe_filename}")
|
|
claimed_names.add(folded_name)
|
|
classification = classify_upload_filename(safe_filename, policy)
|
|
if classification is None:
|
|
raise HTTPException(status_code=415, detail=f"File type is not allowed: {safe_filename}")
|
|
category, max_bytes = classification
|
|
prepared.append((file, safe_filename, category, max_bytes))
|
|
|
|
# Resolve the canonical owner asynchronously before any filesystem work.
|
|
sandbox_uploads = await _resolve_uploads_path(thread_id)
|
|
cancelled = threading.Event()
|
|
# Moving work off the event loop introduces real concurrency. Serialize
|
|
# batches for the same directory so rollback cannot undo another upload.
|
|
async with _upload_lock(sandbox_uploads):
|
|
return await _upload_io.run_until_complete(
|
|
_process_upload_batch,
|
|
thread_id,
|
|
sandbox_uploads,
|
|
prepared,
|
|
claimed_names,
|
|
limits,
|
|
_auto_convert_documents_enabled(config),
|
|
cancelled,
|
|
on_cancel=cancelled.set,
|
|
)
|
|
|
|
|
|
def _process_upload_batch(
|
|
thread_id: str,
|
|
sandbox_uploads: Path,
|
|
prepared: list[tuple[UploadFile, str, str, int]],
|
|
claimed_names: set[str],
|
|
limits: UploadLimits,
|
|
auto_convert_documents: bool,
|
|
cancelled: threading.Event,
|
|
) -> UploadResponse:
|
|
"""Own the full staging/commit/rollback lifetime in one bounded worker."""
|
|
_check_upload_cancelled(cancelled)
|
|
try:
|
|
uploads_dir = ensure_uploads_dir(thread_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
uploaded_files: list[dict[str, str]] = []
|
|
sandbox_sync_targets: list[tuple[Path, str]] = []
|
|
total_size = 0
|
|
staging_dir = Path(tempfile.mkdtemp(prefix=".upload-staging-", dir=uploads_dir))
|
|
records: list[dict] = []
|
|
committed_names: list[str] | None = None
|
|
try:
|
|
for file, safe_filename, category, max_bytes in prepared:
|
|
staged_path, file_size, total_size = _copy_upload_file_with_limits(
|
|
file,
|
|
uploads_dir=staging_dir,
|
|
display_filename=safe_filename,
|
|
max_single_file_size=max_bytes,
|
|
max_total_size=limits.max_total_size,
|
|
total_size=total_size,
|
|
cancelled=cancelled,
|
|
)
|
|
file_ext = staged_path.suffix.lower()
|
|
try:
|
|
validate_upload_content(staged_path, file_ext)
|
|
except UploadContentError as exc:
|
|
raise HTTPException(status_code=415, detail=f"Invalid file {safe_filename}: {exc}") from exc
|
|
|
|
record = {
|
|
"filename": safe_filename,
|
|
"size": file_size,
|
|
"category": category,
|
|
"detected_type": file_ext.removeprefix("."),
|
|
}
|
|
if auto_convert_documents and file_ext in CONVERTIBLE_EXTENSIONS:
|
|
_check_upload_cancelled(cancelled)
|
|
md_path = convert_file_to_markdown_sync(staged_path)
|
|
if md_path:
|
|
markdown_folded = md_path.name.casefold()
|
|
if markdown_folded in claimed_names:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"Generated Markdown filename collides with another upload: {md_path.name}",
|
|
)
|
|
claimed_names.add(markdown_folded)
|
|
record["markdown_file"] = md_path.name
|
|
records.append(record)
|
|
|
|
_check_upload_cancelled(cancelled)
|
|
sandbox_provider = get_sandbox_provider()
|
|
sync_to_sandbox = not _uses_thread_data_mounts(sandbox_provider)
|
|
sandbox = None
|
|
if sync_to_sandbox:
|
|
sandbox_id = sandbox_provider.acquire(thread_id)
|
|
sandbox = sandbox_provider.get(sandbox_id)
|
|
if sandbox is None:
|
|
raise HTTPException(status_code=500, detail="Failed to acquire sandbox")
|
|
|
|
commit_names = [str(record["filename"]) for record in records]
|
|
commit_names.extend(str(record["markdown_file"]) for record in records if record.get("markdown_file"))
|
|
_check_upload_cancelled(cancelled)
|
|
_commit_staged_files(staging_dir, uploads_dir, commit_names)
|
|
committed_names = commit_names
|
|
|
|
for record in records:
|
|
safe_filename = str(record["filename"])
|
|
final_path = uploads_dir / safe_filename
|
|
virtual_path = upload_virtual_path(safe_filename)
|
|
file_info = {
|
|
"filename": safe_filename,
|
|
"size": str(record["size"]),
|
|
"category": str(record["category"]),
|
|
"detected_type": str(record["detected_type"]),
|
|
"path": str(sandbox_uploads / safe_filename),
|
|
"virtual_path": virtual_path,
|
|
"artifact_url": upload_artifact_url(thread_id, safe_filename),
|
|
}
|
|
if sync_to_sandbox:
|
|
sandbox_sync_targets.append((final_path, virtual_path))
|
|
markdown_name = record.get("markdown_file")
|
|
if markdown_name:
|
|
markdown_name = str(markdown_name)
|
|
markdown_path = uploads_dir / markdown_name
|
|
markdown_virtual_path = upload_virtual_path(markdown_name)
|
|
file_info.update(
|
|
{
|
|
"markdown_file": markdown_name,
|
|
"markdown_path": str(sandbox_uploads / markdown_name),
|
|
"markdown_virtual_path": markdown_virtual_path,
|
|
"markdown_artifact_url": upload_artifact_url(thread_id, markdown_name),
|
|
}
|
|
)
|
|
if sync_to_sandbox:
|
|
sandbox_sync_targets.append((markdown_path, markdown_virtual_path))
|
|
uploaded_files.append(file_info)
|
|
logger.info("Saved validated %s upload: %s (%d bytes)", record["category"], safe_filename, record["size"])
|
|
|
|
if sync_to_sandbox and sandbox is not None:
|
|
for file_path, virtual_path in sandbox_sync_targets:
|
|
_check_upload_cancelled(cancelled)
|
|
_make_file_sandbox_writable(file_path)
|
|
sandbox.update_file(virtual_path, file_path.read_bytes())
|
|
_check_upload_cancelled(cancelled)
|
|
except (HTTPException, _UploadCancelled):
|
|
if committed_names is not None:
|
|
_rollback_committed_files(staging_dir, uploads_dir, committed_names)
|
|
raise
|
|
except UnsafeUploadPathError as exc:
|
|
if committed_names is not None:
|
|
_rollback_committed_files(staging_dir, uploads_dir, committed_names)
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
except Exception as exc:
|
|
logger.exception("Failed to upload validated batch")
|
|
if committed_names is not None:
|
|
try:
|
|
_rollback_committed_files(staging_dir, uploads_dir, committed_names)
|
|
except Exception:
|
|
logger.critical("Upload batch rollback failed", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=f"Failed to upload files: {exc}") from exc
|
|
finally:
|
|
shutil.rmtree(staging_dir, ignore_errors=True)
|
|
|
|
return UploadResponse(
|
|
success=True,
|
|
files=uploaded_files,
|
|
message=f"Successfully uploaded {len(uploaded_files)} file(s)",
|
|
skipped_files=[],
|
|
policy_revision=limits.revision,
|
|
)
|
|
|
|
|
|
@router.get("/limits", response_model=UploadLimits)
|
|
@require_permission("threads", "read", owner_check=True)
|
|
async def get_upload_limits(
|
|
thread_id: str,
|
|
request: Request,
|
|
) -> UploadLimits:
|
|
"""Return upload limits used by the gateway for this thread."""
|
|
policy, metadata = await get_effective_upload_policy(request)
|
|
return _get_upload_limits(policy, int(metadata.get("revision") or 0))
|
|
|
|
|
|
@router.get("/list", response_model=dict)
|
|
@require_permission("threads", "read", owner_check=True)
|
|
async def list_uploaded_files(thread_id: str, request: Request) -> dict:
|
|
"""List all files in a thread's uploads directory."""
|
|
sandbox_uploads = await _resolve_uploads_path(thread_id)
|
|
async with _upload_lock(sandbox_uploads):
|
|
return await file_io.run_until_complete(_list_uploaded_files, thread_id, sandbox_uploads)
|
|
|
|
|
|
def _list_uploaded_files(thread_id: str, sandbox_uploads: Path) -> dict:
|
|
try:
|
|
uploads_dir = get_uploads_dir(thread_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
result = list_files_in_dir(uploads_dir)
|
|
enrich_file_listing(result, thread_id)
|
|
|
|
# Gateway additionally includes the sandbox-relative path.
|
|
for f in result["files"]:
|
|
f["path"] = str(sandbox_uploads / f["filename"])
|
|
|
|
return result
|
|
|
|
|
|
@router.delete("/{filename}")
|
|
@require_permission("threads", "delete", owner_check=True, require_existing=True)
|
|
async def delete_uploaded_file(thread_id: str, filename: str, request: Request) -> dict:
|
|
"""Delete a file from a thread's uploads directory."""
|
|
sandbox_uploads = await _resolve_uploads_path(thread_id)
|
|
async with _upload_lock(sandbox_uploads):
|
|
return await file_io.run_until_complete(_delete_uploaded_file, thread_id, filename)
|
|
|
|
|
|
def _delete_uploaded_file(thread_id: str, filename: str) -> dict:
|
|
try:
|
|
uploads_dir = get_uploads_dir(thread_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
try:
|
|
return delete_file_safe(uploads_dir, filename, convertible_extensions=CONVERTIBLE_EXTENSIONS)
|
|
except FileNotFoundError:
|
|
raise HTTPException(status_code=404, detail=f"File not found: {filename}")
|
|
except PathTraversalError:
|
|
raise HTTPException(status_code=400, detail="Invalid path")
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete {filename}: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to delete {filename}: {str(e)}")
|