From 03afd0591774d55caa2ad1998b7afc82ae871ca7 Mon Sep 17 00:00:00 2001 From: Willem Jiang Date: Tue, 27 Oct 2026 16:59:19 +0800 Subject: [PATCH] fix(artifacts): authorize shared image reads and offload MIME lookup (#7604) * fix(artifacts): authorize shared image reads and offload MIME lookup * fix(artifacts): skip disabled blob lookups and warn on failures --- README.md | 7 + backend/app/gateway/routers/artifacts.py | 17 +- backend/docs/blob-storage.md | 10 +- .../harness/deerflow/storage/AGENTS.md | 8 + .../blocking_io/test_artifacts_router.py | 45 ++++ backend/tests/test_artifacts_router.py | 206 ++++++++++++++---- 6 files changed, 248 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index fc088eeb30..5c810882c4 100644 --- a/README.md +++ b/README.md @@ -497,6 +497,13 @@ Custom blob storage backends accept `blob_storage.backend: module.Class` or The selected class must implement `BlobStore`; backend options go under `blob_storage.backend_config`. See [blob storage](backend/docs/blob-storage.md). +Viewed-image artifact URLs can use shared blobs when the serving Gateway lacks +the local file. This fallback requires an existing thread record accessible to +the caller; deleting that record revokes access even if checkpoint cleanup fails. +When blob storage is disabled, a missing local file returns 404 without loading +checkpoint state. Shared-store failures also return 404 and emit a warning with +the exception type, without backend error details. + ### Running the Application #### Deployment Sizing diff --git a/backend/app/gateway/routers/artifacts.py b/backend/app/gateway/routers/artifacts.py index e3ca9b2bd7..2887ff59a8 100644 --- a/backend/app/gateway/routers/artifacts.py +++ b/backend/app/gateway/routers/artifacts.py @@ -531,6 +531,9 @@ async def get_artifact(thread_id: ThreadId, path: str, request: Request, downloa raise AssertionError(f"Unhandled artifact response kind: {kind!r}") +# Legacy local-file reads allow untracked threads. Shared checkpoint blobs must +# recheck a surviving authorization record, including trusted internal owners. +@require_permission("threads", "read", owner_check=True, require_existing=True) async def _get_viewed_image_blob_response( thread_id: str, path: str, @@ -548,6 +551,10 @@ async def _get_viewed_image_blob_response( virtual_path = f"/{path.lstrip('/')}" try: + store = await asyncio.to_thread(get_blob_store_if_enabled) + if store is None: + return None + from app.gateway.services import build_thread_checkpoint_state_accessor accessor, config = await build_thread_checkpoint_state_accessor(request, thread_id=thread_id) @@ -572,7 +579,8 @@ async def _get_viewed_image_blob_response( or not mime_type.startswith("image/") ): return None - path_mime_type, _ = mimetypes.guess_type(virtual_path) + # A cold mimetypes database stats/reads system files during its first lookup. + path_mime_type, _ = await asyncio.to_thread(mimetypes.guess_type, virtual_path) if path_mime_type != mime_type: return None blob_ref_data = metadata.get("blob_ref") @@ -581,14 +589,11 @@ async def _get_viewed_image_blob_response( blob_ref = BlobRef.model_validate(blob_ref_data) if blob_ref.kind != "viewed-image" or blob_ref.sha256 != expected_sha256 or blob_ref.size != expected_size or blob_ref.content_type != mime_type: return None - store = await asyncio.to_thread(get_blob_store_if_enabled) - if store is None: - return None content = await asyncio.to_thread(store.get_bytes, blob_ref) if len(content) != expected_size or hashlib.sha256(content).hexdigest() != expected_sha256: return None - except Exception: - logger.debug("Could not resolve viewed-image blob for thread artifact", exc_info=True) + except Exception as exc: + logger.warning("Could not resolve viewed-image blob for thread artifact (%s)", type(exc).__name__) return None etag = f'"{expected_sha256}"' diff --git a/backend/docs/blob-storage.md b/backend/docs/blob-storage.md index 0d265e3c99..c825c210cf 100644 --- a/backend/docs/blob-storage.md +++ b/backend/docs/blob-storage.md @@ -65,8 +65,14 @@ The producer migrations are independently revertible: digest-bound blob before the sandbox/local compatibility paths. Thread artifact GET also serves a checkpoint-bound image blob when the reader Gateway lacks the local file; the reference is checked against the thread's - virtual path, digest, size, kind and MIME type. IM channels and - `present_file_tool` still use local-path reads. Store initialization or write + virtual path, digest, size, kind and MIME type. Blob fallback requires an + existing authorized thread record, including for trusted internal owner + requests. A deleted or untracked thread cannot expose retained checkpoint + blobs; legacy local-file artifact reads keep their existing access policy. + After authorization, a disabled-store check skips checkpoint access. Store + initialization/read failures retain the 404 response and log a warning with + the exception type only, without raw backend messages or tracebacks. + IM channels and `present_file_tool` still use local-path reads. Store initialization or write failures return a generic tool error; model-read failures fall through to the validated sandbox/host compatibility paths. 2. **Externalized tool results — migrated.** Host externalization stores the diff --git a/backend/packages/harness/deerflow/storage/AGENTS.md b/backend/packages/harness/deerflow/storage/AGENTS.md index d5fe2d2943..38df81117a 100644 --- a/backend/packages/harness/deerflow/storage/AGENTS.md +++ b/backend/packages/harness/deerflow/storage/AGENTS.md @@ -31,6 +31,14 @@ sandbox/host compatibility path as fallback after any store failure. The thread artifact GET route also resolves the checkpoint-bound image ref when the reader Gateway cannot find the local file; it confines this fallback to the authorized thread and exact virtual path and verifies the bytes before serving. +The blob fallback reuses `require_permission` with `require_existing=True` +before the off-loop store probe, including trusted internal owners. A disabled +store skips accessor construction/checkpoint reads; reuse the returned store +for the blob read. Unexpected failures warn with only the exception type and +still fall back to 404. Local-file reads keep legacy missing-row access. +MIME detection runs off-loop even on a cold +reader. Regression coverage uses the decorated artifact route with a real +thread store and the strict blocking-I/O suite with an uninitialized MIME DB. Other artifact and IM-channel reads still use their existing local-path flow. `ToolOutputBudgetMiddleware` is a migrated producer for host-externalized diff --git a/backend/tests/blocking_io/test_artifacts_router.py b/backend/tests/blocking_io/test_artifacts_router.py index ba146a6f09..c4d43e865a 100644 --- a/backend/tests/blocking_io/test_artifacts_router.py +++ b/backend/tests/blocking_io/test_artifacts_router.py @@ -27,16 +27,23 @@ from __future__ import annotations import asyncio import hashlib +import mimetypes import zipfile from contextlib import asynccontextmanager from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest +from starlette.requests import Request from starlette.responses import FileResponse import app.gateway.routers.artifacts as artifacts_router +from app.gateway import services +from app.gateway.authz import AuthContext from app.gateway.path_utils import resolve_thread_virtual_path from app.gateway.routers.artifacts import ArtifactUpdateRequest, get_artifact, update_artifact +from deerflow.storage import BlobRef pytestmark = pytest.mark.asyncio @@ -149,3 +156,41 @@ async def test_large_artifact_validator_does_not_block_event_loop(tmp_path: Path assert isinstance(response, FileResponse) assert response.headers["etag"].startswith('"stat-') + + +async def test_shared_image_blob_cold_mime_lookup_does_not_block(tmp_path, monkeypatch): + payload = b"\x89PNG\r\n\x1a\nimage" + virtual_path = "/mnt/user-data/outputs/chart.png" + ref = BlobRef(sha256=hashlib.sha256(payload).hexdigest(), size=len(payload), kind="viewed-image", content_type="image/png") + mime_file = tmp_path / "mime.types" + await asyncio.to_thread(mime_file.write_text, "image/png png\n", encoding="utf-8") + # Restore all globals replaced by init(), not just its lazy-init flag. + for name in ("suffix_map", "encodings_map", "types_map", "common_types"): + monkeypatch.setattr(mimetypes, name, getattr(mimetypes, name)) + monkeypatch.setattr(mimetypes, "inited", False) + monkeypatch.setattr(mimetypes, "_db", None) + monkeypatch.setattr(mimetypes, "knownfiles", [str(mime_file)]) + + class Accessor: + async def aget(self, _config): + return SimpleNamespace(values={"viewed_images": {virtual_path: {"mime_type": "image/png", "size": len(payload), "sha256": ref.sha256, "blob_ref": ref.model_dump(mode="json")}}}) + + async def build_accessor(_request, *, thread_id): + return Accessor(), {"configurable": {"thread_id": thread_id}} + + class Store: + def get_bytes(self, requested_ref): + assert requested_ref == ref + return payload + + monkeypatch.setattr(services, "build_thread_checkpoint_state_accessor", build_accessor) + monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", lambda: Store()) + monkeypatch.setattr(artifacts_router, "resolve_thread_virtual_path", lambda *_args, **_kwargs: tmp_path / "missing.png") + request = Request({"type": "http", "method": "GET", "path": "/", "headers": [], "query_string": b"", "app": SimpleNamespace(state=SimpleNamespace(thread_store=SimpleNamespace(check_access=AsyncMock(return_value=True))))}) + user = SimpleNamespace(id="owner", system_role="user") + request.state.user = user + request.state.auth = AuthContext(user=user, permissions=["threads:read"]) + + response = await _get_artifact("thread-1", virtual_path.lstrip("/"), request=request) + assert response.status_code == 200 + assert response.body == payload diff --git a/backend/tests/test_artifacts_router.py b/backend/tests/test_artifacts_router.py index dcb902a3f9..0f1890f417 100644 --- a/backend/tests/test_artifacts_router.py +++ b/backend/tests/test_artifacts_router.py @@ -1,5 +1,6 @@ import asyncio import hashlib +import logging import os import stat import threading @@ -7,6 +8,7 @@ import zipfile from contextlib import asynccontextmanager from pathlib import Path from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock import pytest from _router_auth_helpers import call_unwrapped, make_authed_test_app @@ -109,39 +111,21 @@ def test_get_artifact_reads_shared_viewed_image_blob_when_local_path_is_missing( monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", lambda: SharedStore()) - response = asyncio.run( - call_unwrapped( - artifacts_router.get_artifact, - "thread-1", - virtual_path.lstrip("/"), - _make_request(), - ) - ) + app = make_authed_test_app() + app.include_router(artifacts_router.router) + url = f"/api/threads/thread-1/artifacts/{virtual_path.lstrip('/')}" + with TestClient(app) as client: + response = client.get(url) assert response.status_code == 200 - assert response.body == image_bytes - assert response.media_type == "image/png" + assert response.content == image_bytes + assert response.headers["content-type"] == "image/png" assert response.headers["ETag"] == f'"{digest}"' - range_request = Request( - { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"range", b"bytes=1-3"), (b"if-range", f'"{digest}"'.encode())], - "query_string": b"", - } - ) - ranged = asyncio.run( - call_unwrapped( - artifacts_router.get_artifact, - "thread-1", - virtual_path.lstrip("/"), - range_request, - ) - ) + with TestClient(app) as client: + ranged = client.get(url, headers={"Range": "bytes=1-3", "If-Range": f'"{digest}"'}) assert ranged.status_code == 206 - assert ranged.body == image_bytes[1:4] + assert ranged.content == image_bytes[1:4] assert ranged.headers["Content-Range"] == f"bytes 1-3/{len(image_bytes)}" @@ -188,15 +172,163 @@ def test_get_artifact_rejects_checkpoint_image_blob_metadata_mismatch(tmp_path, monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", lambda: MustNotReadStore()) - with pytest.raises(HTTPException, match="Artifact not found"): - asyncio.run( - call_unwrapped( - artifacts_router.get_artifact, - "thread-1", - virtual_path.lstrip("/"), - _make_request(), - ) - ) + app = make_authed_test_app() + app.include_router(artifacts_router.router) + with TestClient(app) as client: + response = client.get(f"/api/threads/thread-1/artifacts/{virtual_path.lstrip('/')}") + assert response.status_code == 404 + assert "Artifact not found" in response.json()["detail"] + + +@pytest.mark.parametrize( + ("record_state", "internal", "owner_header", "expected_status"), + [ + ("owned", False, None, 200), + ("foreign", False, None, 404), + ("deleted", False, None, 404), + ("missing", False, None, 404), + ("deleted", True, "owner", 404), + ("foreign", True, "owner", 200), + ("foreign", False, "owner", 404), + ("shared", False, None, 200), + ], +) +def test_shared_image_blob_requires_existing_thread_access(tmp_path, monkeypatch, record_state, internal, owner_header, expected_status): + from langgraph.store.memory import InMemoryStore + + from app.gateway import services + from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore + from deerflow.storage import BlobRef + + image_bytes = b"\x89PNG\r\n\x1a\nprivate image" + virtual_path = "/mnt/user-data/outputs/private.png" + ref = BlobRef(sha256=hashlib.sha256(image_bytes).hexdigest(), size=len(image_bytes), kind="viewed-image", content_type="image/png") + thread_store = MemoryThreadMetaStore(InMemoryStore()) + if record_state != "missing": + owner = "reader" if record_state == "owned" else None if record_state == "shared" else "owner" + asyncio.run(thread_store.create("thread-1", user_id=owner)) + if record_state == "deleted": + # Checkpoint cleanup can fail independently of metadata deletion. + asyncio.run(thread_store.delete("thread-1", user_id=owner)) + + metadata_reads, blob_reads, store_probes = [], [], [] + + class Accessor: + async def aget(self, _config): + metadata_reads.append(True) + return SimpleNamespace(values={"viewed_images": {virtual_path: {"mime_type": "image/png", "size": len(image_bytes), "sha256": ref.sha256, "blob_ref": ref.model_dump(mode="json")}}}) + + async def build_accessor(_request, *, thread_id): + assert thread_id == "thread-1" + return Accessor(), {"configurable": {"thread_id": thread_id}} + + class SharedStore: + def get_bytes(self, requested_ref): + blob_reads.append(requested_ref) + return image_bytes + + def get_store(): + store_probes.append(True) + return SharedStore() + + monkeypatch.setattr(services, "build_thread_checkpoint_state_accessor", build_accessor) + monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", get_store) + monkeypatch.setattr(artifacts_router, "resolve_thread_virtual_path", lambda *_args, **_kwargs: tmp_path / "absent.png") + user = SimpleNamespace(id="reader", system_role=INTERNAL_SYSTEM_ROLE if internal else "user") + app = make_authed_test_app(user_factory=lambda: user) + app.state.thread_store = thread_store + app.include_router(artifacts_router.router) + headers = {INTERNAL_OWNER_USER_ID_HEADER_NAME: owner_header} if owner_header else {} + with TestClient(app) as client: + response = client.get(f"/api/threads/thread-1/artifacts/{virtual_path.lstrip('/')}", headers=headers) + + assert response.status_code == expected_status + if expected_status == 200: + assert response.content == image_bytes + assert blob_reads == [ref] + assert metadata_reads == [True] + assert store_probes == [True] + else: + assert not metadata_reads + assert not blob_reads + assert not store_probes + + +@pytest.fixture +def shared_image_blob_request(tmp_path, monkeypatch): + from app.gateway import services + from deerflow.storage import BlobRef + + payload = b"\x89PNG\r\n\x1a\nimage" + virtual_path = "/mnt/user-data/outputs/chart.png" + ref = BlobRef(sha256=hashlib.sha256(payload).hexdigest(), size=len(payload), kind="viewed-image", content_type="image/png") + metadata = {"mime_type": "image/png", "size": len(payload), "sha256": ref.sha256, "blob_ref": ref.model_dump(mode="json")} + accessor = SimpleNamespace(aget=AsyncMock(return_value=SimpleNamespace(values={"viewed_images": {virtual_path: metadata}}))) + build_accessor = AsyncMock(return_value=(accessor, {"configurable": {"thread_id": "thread-1"}})) + monkeypatch.setattr(services, "build_thread_checkpoint_state_accessor", build_accessor) + monkeypatch.setattr(artifacts_router, "resolve_thread_virtual_path", lambda *_args, **_kwargs: tmp_path / "absent.png") + app = make_authed_test_app() + app.include_router(artifacts_router.router) + with TestClient(app) as client: + yield SimpleNamespace(client=client, url=f"/api/threads/thread-1/artifacts/{virtual_path.lstrip('/')}", build_accessor=build_accessor, accessor=accessor, ref=ref, payload=payload) + + +def test_disabled_blob_storage_skips_checkpoint_reads(shared_image_blob_request, monkeypatch): + scenario = shared_image_blob_request + get_store = Mock(return_value=None) + monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", get_store) + + response = scenario.client.get(scenario.url) + + assert response.status_code == 404 + assert "Artifact not found" in response.json()["detail"] + get_store.assert_called_once_with() + scenario.build_accessor.assert_not_awaited() + scenario.accessor.aget.assert_not_awaited() + + +@pytest.mark.parametrize("failure_stage", ["initialize", "read"]) +def test_shared_image_blob_store_failures_warn_without_details(shared_image_blob_request, monkeypatch, caplog, failure_stage): + from deerflow.storage import BlobReadError + + scenario = shared_image_blob_request + private_detail = "sensitive-backend-detail-not-for-logs" + error = RuntimeError(private_detail) if failure_stage == "initialize" else BlobReadError(private_detail) + read_blob = Mock(side_effect=error) + get_store = Mock(side_effect=error) if failure_stage == "initialize" else Mock(return_value=SimpleNamespace(get_bytes=read_blob)) + monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", get_store) + + with caplog.at_level(logging.WARNING, logger=artifacts_router.__name__): + response = scenario.client.get(scenario.url) + + assert response.status_code == 404 + records = [record for record in caplog.records if record.name == artifacts_router.__name__ and record.levelno == logging.WARNING] + assert len(records) == 1 + assert records[0].getMessage() == f"Could not resolve viewed-image blob for thread artifact ({type(error).__name__})" + assert private_detail not in caplog.text + assert private_detail not in response.text + get_store.assert_called_once_with() + if failure_stage == "initialize": + scenario.build_accessor.assert_not_awaited() + else: + read_blob.assert_called_once_with(scenario.ref) + + +def test_local_artifact_keeps_legacy_missing_thread_access(tmp_path, monkeypatch): + from langgraph.store.memory import InMemoryStore + + from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore + + local_file = tmp_path / "legacy.txt" + local_file.write_text("legacy local artifact", encoding="utf-8") + monkeypatch.setattr(artifacts_router, "resolve_thread_virtual_path", lambda *_args, **_kwargs: local_file) + app = make_authed_test_app() + app.state.thread_store = MemoryThreadMetaStore(InMemoryStore()) + app.include_router(artifacts_router.router) + with TestClient(app) as client: + response = client.get("/api/threads/legacy-thread/artifacts/mnt/user-data/outputs/legacy.txt") + assert response.status_code == 200 + assert response.text == "legacy local artifact" @asynccontextmanager