From 7a8051c0d9211abddc61b1e6031641a6a3c2216e Mon Sep 17 00:00:00 2001 From: yu1128-L <2241397554@qq.com> Date: Mon, 9 Nov 2026 21:18:57 +0800 Subject: [PATCH] fix(artifacts): bind ranged reads to validated generation (#7978) * fix(artifacts): bind ranged reads to validated generation * fix(artifacts): drain pinned reads without Windows hard links --------- Co-authored-by: Willem Jiang --- README.md | 5 +- backend/app/gateway/routers/AGENTS.md | 6 + backend/app/gateway/routers/artifacts.py | 208 +++++++++++++++++++++-- backend/tests/test_artifacts_router.py | 129 ++++++++++++++ 4 files changed, 331 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index be75aaa3cc..6bed016eb8 100644 --- a/README.md +++ b/README.md @@ -3367,8 +3367,9 @@ external image URLs keep their existing behavior. If the sample cuts a CRLF line ending in half, the preview keeps the earlier complete rows and omits the incomplete final record, including when its last field is quoted. -Text artifacts are streamed with HTTP byte-range support. The Web UI initially -loads at most 1 MiB, shows the preview size when a file is larger, and waits for +Text artifacts are streamed with HTTP byte-range support. A resumed download +validates its ETag against the file version that supplies its response bytes. +The Web UI initially loads at most 1 MiB, shows the preview size for larger files, and waits for an explicit **Load full file** action before fetching the remainder or mounting the full code editor. Editing becomes available once the complete file has loaded; unsaved-change detection uses the full content, including when reverting edits diff --git a/backend/app/gateway/routers/AGENTS.md b/backend/app/gateway/routers/AGENTS.md index a57e7590f3..7b1ca367bf 100644 --- a/backend/app/gateway/routers/AGENTS.md +++ b/backend/app/gateway/routers/AGENTS.md @@ -1,5 +1,11 @@ # Router contracts +Artifact range validators and bytes come from one opened file. POSIX streams +the descriptor. Windows copies only the selected response bytes into a +private temporary file and closes the public handle before sending headers. +Preserve the ZIP single-link checks and drain preparation before cancellation +can release an artifact response's resources. + Trash listing resolves the live retention configuration off-loop before its best-effort sweep. Configuration failures skip automatic cleanup and preserve the listing; never substitute a default policy for an unreadable one. Resolve diff --git a/backend/app/gateway/routers/artifacts.py b/backend/app/gateway/routers/artifacts.py index 2887ff59a8..a8a00d9dde 100644 --- a/backend/app/gateway/routers/artifacts.py +++ b/backend/app/gateway/routers/artifacts.py @@ -4,17 +4,21 @@ import hashlib import logging import mimetypes import os +import secrets import stat import tempfile +import threading import zipfile from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path +from typing import BinaryIO from urllib.parse import quote from fastapi import APIRouter, HTTPException, Request from fastapi.responses import FileResponse, Response from pydantic import BaseModel, Field +from starlette.datastructures import MutableHeaders from app.gateway.authz import SandboxRequestLease, require_permission, try_acquire_sandbox_for_request from app.gateway.deps import get_run_manager @@ -41,6 +45,8 @@ _SKILL_ARCHIVE_READ_CHUNK_SIZE = 64 * 1024 MAX_EDITABLE_ARTIFACT_BYTES = 2 * 1024 * 1024 MAX_VIEWED_IMAGE_BLOB_BYTES = 20 * 1024 * 1024 _ARTIFACT_EDIT_TEMP_PREFIX = ".artifact-edit-" +_ARTIFACT_COPY_BEFORE_STREAM = os.name == "nt" +_artifact_digest_source = threading.local() class ArtifactUpdateRequest(BaseModel): @@ -334,7 +340,7 @@ def _read_artifact_payload(actual_path: Path, path: str, download: bool) -> tupl return ("inline_file", mime_type) -def _sha256_of_file(path: Path) -> str: +def _sha256_of_file(path: Path, *, pinned: tuple[BinaryIO, os.stat_result] | None = None) -> str: """Return the hex SHA-256 digest of *path* without loading it whole. Computing the digest on the Gateway lets the browser skip its own @@ -346,45 +352,217 @@ def _sha256_of_file(path: Path) -> str: requests a browser issues while scrubbing/paginating a preview do not each re-hash a potentially huge artifact from scratch (raised in PR review). """ - stat = path.stat() + file_stat = path.stat() if pinned is None else pinned[1] # Sandbox syncs and other writers can atomically replace a same-size file # while preserving its mtime. Identity separates those generations; ctime # also invalidates in-place writes where the filesystem exposes change time. - return _sha256_of_file_cached( + key = ( str(path), - stat.st_dev, - stat.st_ino, - stat.st_ctime_ns, - stat.st_mtime_ns, - stat.st_size, + file_stat.st_dev, + file_stat.st_ino, + file_stat.st_ctime_ns, + file_stat.st_mtime_ns, + file_stat.st_size, ) + if pinned is None: + return _sha256_of_file_cached(*key) + + previous = getattr(_artifact_digest_source, "pinned", None) + _artifact_digest_source.pinned = (key, pinned[0]) + try: + return _sha256_of_file_cached(*key) + finally: + _artifact_digest_source.pinned = previous @functools.lru_cache(maxsize=256) def _sha256_of_file_cached(path: str, device: int, inode: int, ctime_ns: int, mtime_ns: int, size: int) -> str: """Cached SHA-256 of *path*; metadata arguments separate file generations.""" - digest = hashlib.sha256() - with open(path, "rb") as handle: - for chunk in iter(lambda: handle.read(1 << 20), b""): - digest.update(chunk) - return digest.hexdigest() + pinned = getattr(_artifact_digest_source, "pinned", None) + # A nested lookup for a different file must hash its own bytes. + if pinned is None or pinned[0] != (path, device, inode, ctime_ns, mtime_ns, size): + with open(path, "rb") as handle: + return hashlib.file_digest(handle, "sha256").hexdigest() + + handle = pinned[1] + previous_offset = handle.tell() + try: + handle.seek(0) + return hashlib.file_digest(handle, "sha256").hexdigest() + finally: + handle.seek(previous_offset) def _artifact_etag(path: Path) -> str: - """Keep content revisions for editable files and metadata-only tags for large files.""" + """Keep content revisions for editable files and metadata tags for large files.""" file_stat = path.stat() if file_stat.st_size <= MAX_EDITABLE_ARTIFACT_BYTES: - return f'"{_sha256_of_file(path)}"' + # Populate the shared cache only with metadata from the hashed handle. + with path.open("rb") as handle: + return _artifact_etag_from_handle(path, handle, os.fstat(handle.fileno())) + return _artifact_stat_etag(file_stat) + +def _artifact_stat_etag(file_stat: os.stat_result) -> str: # Do not read oversized artifacts just to validate a small byte range. # The prefix keeps this opaque validator distinct from an editable SHA-256. identity = f"{file_stat.st_dev}:{file_stat.st_ino}:{file_stat.st_ctime_ns}:{file_stat.st_mtime_ns}:{file_stat.st_size}" return f'"stat-{hashlib.sha256(identity.encode("ascii")).hexdigest()}"' +def _artifact_etag_from_handle(path: Path, handle: BinaryIO, file_stat: os.stat_result) -> str: + """Build the validator from the same file generation the response will read.""" + if file_stat.st_size <= MAX_EDITABLE_ARTIFACT_BYTES: + digest = _sha256_of_file(path, pinned=(handle, file_stat)) + return f'"{digest}"' + return _artifact_stat_etag(file_stat) + + +def _prepare_artifact_stream(path: Path) -> tuple[BinaryIO, os.stat_result, str]: + """Pin one file generation before response headers authorize a byte range.""" + handle = path.open("rb") + try: + file_stat = os.fstat(handle.fileno()) + if not stat.S_ISREG(file_stat.st_mode): + raise RuntimeError(f"File at path {path} is not a file.") + return handle, file_stat, _artifact_etag_from_handle(path, handle, file_stat) + except BaseException: + _close_artifact_handle(handle) + raise + + +def _copy_artifact_ranges(handle: BinaryIO, ranges: list[tuple[int, int]]) -> BinaryIO: + """Pack the selected ranges into a private temporary file in send order.""" + snapshot = tempfile.TemporaryFile(mode="w+b") + try: + for start, end in ranges: + handle.seek(start) + remaining = end - start + while remaining: + chunk = handle.read(min(1 << 20, remaining)) + if not chunk: + raise OSError("Artifact was truncated while preparing the response") + snapshot.write(chunk) + remaining -= len(chunk) + snapshot.seek(0) + return snapshot + except BaseException: + _close_artifact_handle(snapshot) + raise + + +def _close_artifact_handle(handle: BinaryIO) -> None: + try: + handle.close() + except OSError: + logger.warning("Could not close artifact response handle", exc_info=True) + + class _ArtifactFileResponse(FileResponse): """Require an ETag to authorize conditional ranges of replaceable artifacts.""" + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self._artifact_handle: BinaryIO | None = None + + async def __call__(self, scope, receive, send) -> None: + async def prepare() -> None: + self._artifact_handle, self.stat_result, etag = await asyncio.to_thread(_prepare_artifact_stream, Path(self.path)) + self.headers["etag"] = etag + self.set_stat_headers(self.stat_result) + + try: + await await_drained(prepare()) + await super().__call__(scope, receive, send) + finally: + if self._artifact_handle is not None: + handle, self._artifact_handle = self._artifact_handle, None + await await_drained(asyncio.to_thread(_close_artifact_handle, handle)) + + async def _snapshot_windows_body(self, ranges: list[tuple[int, int]], send_header_only: bool) -> bool: + if not _ARTIFACT_COPY_BEFORE_STREAM: + return False + + source = self._artifact_handle + assert source is not None + + async def snapshot() -> None: + # Windows replacement needs the public handle closed. A private + # copy also leaves the artifact's link count unchanged for ZIP. + self._artifact_handle = None if send_header_only else await asyncio.to_thread(_copy_artifact_ranges, source, ranges) + await asyncio.to_thread(_close_artifact_handle, source) + + await await_drained(snapshot()) + return True + + async def _handle_simple(self, send, send_header_only: bool, send_pathsend: bool) -> None: + assert self.stat_result is not None + await self._snapshot_windows_body([(0, self.stat_result.st_size)], send_header_only) + + await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers}) + if send_header_only: + await send({"type": "http.response.body", "body": b"", "more_body": False}) + return + + handle = self._artifact_handle + assert handle is not None + await await_drained(asyncio.to_thread(handle.seek, 0)) + more_body = True + while more_body: + chunk = await await_drained(asyncio.to_thread(handle.read, self.chunk_size)) + more_body = len(chunk) == self.chunk_size + await send({"type": "http.response.body", "body": chunk, "more_body": more_body}) + + async def _handle_single_range(self, send, start: int, end: int, file_size: int, send_header_only: bool) -> None: + snapshot = await self._snapshot_windows_body([(start, end)], send_header_only) + + headers = MutableHeaders(raw=list(self.raw_headers)) + headers["content-range"] = f"bytes {start}-{end - 1}/{file_size}" + headers["content-length"] = str(end - start) + await send({"type": "http.response.start", "status": 206, "headers": headers.raw}) + if send_header_only: + await send({"type": "http.response.body", "body": b"", "more_body": False}) + return + + handle = self._artifact_handle + assert handle is not None + if not snapshot: + await await_drained(asyncio.to_thread(handle.seek, start)) + more_body = True + while more_body: + chunk = await await_drained(asyncio.to_thread(handle.read, min(self.chunk_size, end - start))) + start += len(chunk) + more_body = len(chunk) == self.chunk_size and start < end + await send({"type": "http.response.body", "body": chunk, "more_body": more_body}) + + async def _handle_multiple_ranges(self, send, ranges: list[tuple[int, int]], file_size: int, send_header_only: bool) -> None: + snapshot = await self._snapshot_windows_body(ranges, send_header_only) + + boundary = secrets.token_hex(13) + content_length, header_generator = self.generate_multipart(ranges, boundary, file_size, self.headers["content-type"]) + headers = MutableHeaders(raw=list(self.raw_headers)) + headers["content-type"] = f"multipart/byteranges; boundary={boundary}" + headers["content-length"] = str(content_length) + await send({"type": "http.response.start", "status": 206, "headers": headers.raw}) + if send_header_only: + await send({"type": "http.response.body", "body": b"", "more_body": False}) + return + + handle = self._artifact_handle + assert handle is not None + for start, end in ranges: + await send({"type": "http.response.body", "body": header_generator(start, end), "more_body": True}) + if not snapshot: + await await_drained(asyncio.to_thread(handle.seek, start)) + while start < end: + chunk = await await_drained(asyncio.to_thread(handle.read, min(self.chunk_size, end - start))) + if not chunk: + raise OSError("Artifact was truncated while streaming a byte range") + start += len(chunk) + await send({"type": "http.response.body", "body": chunk, "more_body": True}) + await send({"type": "http.response.body", "body": b"\r\n", "more_body": True}) + await send({"type": "http.response.body", "body": f"--{boundary}--".encode("latin-1"), "more_body": False}) + def _should_use_range(self, http_if_range: str) -> bool: # Sandbox syncs can preserve mtimes across replacements, so Last-Modified # is not a strong validator. Date-form If-Range must return the full file. diff --git a/backend/tests/test_artifacts_router.py b/backend/tests/test_artifacts_router.py index 0f1890f417..d77655e94e 100644 --- a/backend/tests/test_artifacts_router.py +++ b/backend/tests/test_artifacts_router.py @@ -18,6 +18,7 @@ from starlette.requests import Request from starlette.responses import FileResponse import app.gateway.routers.artifacts as artifacts_router +from app.gateway.artifact_archive import build_artifact_archive from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE from deerflow.config.paths import Paths, make_safe_user_id from deerflow.sandbox.lease import get_sandbox_lease_manager @@ -1041,6 +1042,134 @@ def test_replaced_artifact_rejects_previous_if_range_validator(tmp_path, monkeyp assert response.headers["etag"] == f'"{hashlib.sha256(b"new report").hexdigest()}"' +@pytest.mark.parametrize("ranges", ["bytes=4-9", "bytes=0-1,4-5"], ids=["single", "multipart"]) +@pytest.mark.parametrize( + "copy_before_stream", + [pytest.param(False, marks=pytest.mark.skipif(os.name == "nt", reason="Windows requires a private response copy")), True], + ids=["descriptor", "private-copy"], +) +def test_if_range_stream_keeps_the_validated_artifact_generation(tmp_path, monkeypatch, ranges, copy_before_stream) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + artifact_path = outputs / "note.txt" + original = b"old lesson" + artifact_path.write_bytes(original) + artifacts_router._sha256_of_file_cached.cache_clear() + monkeypatch.setattr(artifacts_router, "_ARTIFACT_COPY_BEFORE_STREAM", copy_before_stream) + monkeypatch.setattr(artifacts_router, "resolve_thread_virtual_path", lambda _thread_id, _path, user_id=None: artifact_path) + app = make_authed_test_app() + app.include_router(artifacts_router.router) + url = "/api/threads/thread-1/artifacts/mnt/user-data/outputs/note.txt" + + replace_after_headers = False + replacement_confirmed = False + + async def replace_between_headers_and_body(scope, receive, send): + async def send_and_replace(message): + nonlocal replacement_confirmed + await send(message) + if replace_after_headers and message["type"] == "http.response.start": + archive = await asyncio.to_thread(build_artifact_archive, outputs, ["mnt/user-data/outputs/note.txt"], user_data_dir=tmp_path) + with archive.file, zipfile.ZipFile(archive.file) as zipped: + assert zipped.read("note.txt") == original + _replace_preserving_artifact_mtime(artifact_path, b"new slides") + replacement_confirmed = True + + await app(scope, receive, send_and_replace) + + with TestClient(replace_between_headers_and_body) as client: + first = client.get(url) + replace_after_headers = True + resumed = client.get(url, headers={"Range": ranges, "If-Range": first.headers["etag"]}) + + assert replacement_confirmed + assert artifact_path.read_bytes() == b"new slides" + assert resumed.status_code == 206 + assert resumed.headers["etag"] == first.headers["etag"] + assert int(resumed.headers["content-length"]) == len(resumed.content) + if ranges == "bytes=4-9": + assert resumed.content == b"lesson" + assert resumed.headers["content-range"] == "bytes 4-9/10" + else: + media_type, _, boundary = resumed.headers["content-type"].partition("; boundary=") + assert media_type == "multipart/byteranges" and boundary + parts = [] + for start, end in [(0, 2), (4, 6)]: + part_header = f"--{boundary}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Range: bytes {start}-{end - 1}/10\r\n\r\n" + parts.append(part_header.encode("ascii") + original[start:end] + b"\r\n") + assert resumed.content == b"".join(parts) + f"--{boundary}--".encode("ascii") + assert list(outputs.iterdir()) == [artifact_path] + + +@pytest.mark.parametrize("stage", ["_prepare_artifact_stream", "_copy_artifact_ranges"]) +def test_cancelled_artifact_response_drains_file_ownership(tmp_path, monkeypatch, stage) -> None: + artifact_path = tmp_path / "note.txt" + artifact_path.write_bytes(b"original") + monkeypatch.setattr(artifacts_router, "_ARTIFACT_COPY_BEFORE_STREAM", True) + worker = getattr(artifacts_router, stage) + release_worker = threading.Event() + handles = [] + + async def exercise(): + loop = asyncio.get_running_loop() + started = asyncio.Event() + finished = asyncio.Event() + + def gated_worker(*args): + result = worker(*args) + handles.append(result[0] if stage == "_prepare_artifact_stream" else result) + loop.call_soon_threadsafe(started.set) + try: + assert release_worker.wait(5) + return result + finally: + loop.call_soon_threadsafe(finished.set) + + async def send(_message): + pytest.fail("Cancellation before preparation completes must send no headers or body") + + monkeypatch.setattr(artifacts_router, stage, gated_worker) + response = artifacts_router._ArtifactFileResponse(artifact_path) + scope = {"type": "http", "method": "GET", "headers": [(b"range", b"bytes=1-2")]} + task = asyncio.create_task(response(scope, AsyncMock(), send)) + try: + await asyncio.wait_for(started.wait(), timeout=5) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + finally: + release_worker.set() + if started.is_set(): + await asyncio.wait_for(finished.wait(), timeout=5) + await asyncio.gather(task, return_exceptions=True) + assert task.cancelled() + assert len(handles) == 1 and handles[0].closed + assert artifact_path.stat().st_nlink == 1 + + asyncio.run(exercise()) + + +def test_pinned_digest_cannot_populate_another_artifacts_cache_key(tmp_path, monkeypatch) -> None: + first = tmp_path / "first.txt" + other = tmp_path / "other.txt" + first.write_bytes(b"first file") + other.write_bytes(b"other file") + cached = artifacts_router._sha256_of_file_cached + cached.cache_clear() + + def nested_lookup(path, *metadata): + if path == str(first): + assert artifacts_router._sha256_of_file(other) == hashlib.sha256(b"other file").hexdigest() + return cached(path, *metadata) + + monkeypatch.setattr(artifacts_router, "_sha256_of_file_cached", nested_lookup) + with first.open("rb") as handle: + handle.seek(2) + assert artifacts_router._artifact_etag_from_handle(first, handle, os.fstat(handle.fileno())) == f'"{hashlib.sha256(b"first file").hexdigest()}"' + assert handle.tell() == 2 + assert artifacts_router._sha256_of_file(other) == hashlib.sha256(b"other file").hexdigest() + + @pytest.mark.parametrize("changed_field", ["st_dev", "st_ino", "st_ctime_ns"]) def test_artifact_digest_invalidates_when_file_identity_changes(tmp_path, monkeypatch, changed_field) -> None: artifact_path = tmp_path / "note.txt"