mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-28 06:00:38 +00:00
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
This commit is contained in:
parent
24f15cc479
commit
03afd05917
@ -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
|
||||
|
||||
@ -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}"'
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user