diff --git a/README.md b/README.md index 31b6b27a52..04fbc3e302 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,7 @@ DeerFlow is model-agnostic — it works with any LLM that implements the OpenAI- - [Install.md](Install.md) - Setup instructions written for coding agents - [Configuration Reference](backend/docs/CONFIGURATION.md) - Every `config.yaml` option - [Backend Architecture](backend/README.md) - Backend architecture and API reference +- [Conversation Sharing](docs/guide/sharing.md) - Expiring read-only snapshots, artifact previews and workspace copies - [Architecture Overview](docs/ARCHITECTURE.md) - How the services fit together - [Contributing Guide](CONTRIBUTING.md) - Development environment, workflow and tests - [Changelog](CHANGELOG.md) - Release notes diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 821665765e..5e8e20b88e 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -22,6 +22,7 @@ replay, archive), [run event stream](../../docs/RUN_EVENT_STREAM.md), ## Trust boundary +- **Share tokens read only frozen snapshots.** See [sharing](../../../docs/guide/sharing.md). - **Server-owned run context.** A client can send run context in `body.context` and in `body.config`. Admit server-owned keys only from internally authenticated callers and strip them from both places for everyone else. When you add a diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index ee1a2f5983..835ebb0c67 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -50,6 +50,7 @@ from app.gateway.routers import ( runs, scheduled_tasks, search, + shares, skill_mutations, skills, subagent_batches, @@ -1382,6 +1383,7 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for # Thread cleanup API is mounted at /api/threads/{thread_id} app.include_router(search.router) app.include_router(threads.router) + app.include_router(shares.router) # Per-user activity feed (own prefix, so /api/threads/{thread_id} never captures it) app.include_router(thread_activity.router) diff --git a/backend/app/gateway/auth_middleware.py b/backend/app/gateway/auth_middleware.py index 2adf8ffc86..f6c6b2495f 100644 --- a/backend/app/gateway/auth_middleware.py +++ b/backend/app/gateway/auth_middleware.py @@ -10,6 +10,7 @@ Fine-grained permission checks remain in authz.py decorators. """ import logging +import re from collections.abc import Callable from fastapi import HTTPException, Request, Response @@ -60,7 +61,11 @@ _PUBLIC_EXACT_PATHS: frozenset[str] = frozenset( ) -def _is_public(path: str) -> bool: +def _is_public(path: str, method: str = "GET") -> bool: + # A share token grants only these reads. Forks and any future write routes + # under /shares still pass through session authentication and permissions. + if method in {"GET", "HEAD"} and re.fullmatch(r"/api/shares/[A-Za-z0-9_-]{43}(?:/artifacts/.+)?", path): + return True stripped = path.rstrip("/") if stripped in _PUBLIC_EXACT_PATHS: return True @@ -91,7 +96,7 @@ class AuthMiddleware(BaseHTTPMiddleware): super().__init__(app) async def dispatch(self, request: Request, call_next: Callable) -> Response: - if _is_public(get_request_route_path(request)): + if _is_public(get_request_route_path(request), request.method): return await call_next(request) internal_user = None diff --git a/backend/app/gateway/extension_artifacts.py b/backend/app/gateway/extension_artifacts.py index 32e2432f8c..ab39834532 100644 --- a/backend/app/gateway/extension_artifacts.py +++ b/backend/app/gateway/extension_artifacts.py @@ -178,7 +178,7 @@ async def import_snapshot(request: Request, user_id: str, thread_id: str, namesp return result -async def read_snapshot(request: Request, user_id: str, thread_id: str, path: str) -> bytes: +async def read_snapshot(request: Request, user_id: str, thread_id: str, path: str, *, max_bytes: int = MAX_BYTES) -> bytes: """Capture bounded output bytes through host storage or an authorized lease. Non-output paths raise HTTP 400; unsafe files raise 422 and host mutation @@ -191,8 +191,8 @@ async def read_snapshot(request: Request, user_id: str, thread_id: str, path: st path = normalize_outputs_virtual_path(path) async with snapshot_sandbox(request, user_id, thread_id) as sandbox: if sandbox is not None: - return await run_file_io(read_remote, sandbox, path.removeprefix("/mnt/user-data/outputs/")) - return await run_file_io(read_output_snapshot, user_id, thread_id, path) + return await run_file_io(read_remote, sandbox, path.removeprefix("/mnt/user-data/outputs/"), max_bytes=max_bytes) + return await run_file_io(read_output_snapshot, user_id, thread_id, path, max_bytes=max_bytes) async def verify_imports(request: Request, user_id: str, thread_id: str, receipts: list[dict[str, Any]], *, lifetime: AsyncExitStack | None = None) -> None: @@ -220,7 +220,7 @@ async def verify_imports(request: Request, user_id: str, thread_id: str, receipt raise AgentRunError(422, "Attached artifact copy is unavailable or changed; inspect the team artifacts and send a new request") from exc -def read_output_snapshot(user_id: str, thread_id: str, path: str) -> bytes: +def read_output_snapshot(user_id: str, thread_id: str, path: str, *, max_bytes: int = MAX_BYTES) -> bytes: """Read bounded host output bytes without following links. Non-output paths raise HTTP 400; unsafe files raise 422 and mutation during @@ -229,10 +229,10 @@ def read_output_snapshot(user_id: str, thread_id: str, path: str) -> bytes: if os.name != "posix" or not isinstance(path, str): raise AgentRunError(422, "Output capture requires POSIX and an output path") path = normalize_outputs_virtual_path(path) - return _read_host_snapshot(user_id, thread_id, path.removeprefix("/mnt/user-data/outputs/"), "outputs") + return _read_host_snapshot(user_id, thread_id, path.removeprefix("/mnt/user-data/outputs/"), "outputs", max_bytes=max_bytes) -def _read_host_snapshot(user_id: str, thread_id: str, path: str, area: str) -> bytes: +def _read_host_snapshot(user_id: str, thread_id: str, path: str, area: str, *, max_bytes: int = MAX_BYTES) -> bytes: paths = get_paths() root = paths.base_dir.resolve() base = paths.sandbox_outputs_dir(thread_id, user_id=user_id) if area == "outputs" else paths.sandbox_uploads_dir(thread_id, user_id=user_id) @@ -247,11 +247,11 @@ def _read_host_snapshot(user_id: str, thread_id: str, path: str, area: str) -> b fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) with os.fdopen(fd, "rb") as source: before = os.fstat(source.fileno()) - if not stat.S_ISREG(before.st_mode) or before.st_size > MAX_BYTES or (area == "uploads" and before.st_nlink != 1): - raise AgentRunError(422, "Artifact must be a regular file of at most 2 MiB") - content = source.read(MAX_BYTES + 1) + if not stat.S_ISREG(before.st_mode) or before.st_size > max_bytes or before.st_nlink != 1: + raise AgentRunError(422, "Artifact must be a regular, unlinked file within the capture limit") + content = source.read(max_bytes + 1) after = os.fstat(source.fileno()) - if len(content) > MAX_BYTES or (before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_size, after.st_mtime_ns, after.st_ctime_ns): + if len(content) > max_bytes or (before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_size, after.st_mtime_ns, after.st_ctime_ns): raise AgentRunError(409, "Artifact changed during capture") return content except OSError as exc: diff --git a/backend/app/gateway/routers/shares.py b/backend/app/gateway/routers/shares.py new file mode 100644 index 0000000000..a611b0c8fc --- /dev/null +++ b/backend/app/gateway/routers/shares.py @@ -0,0 +1,130 @@ +"""Owner-managed snapshots, public read capabilities, and authenticated imports.""" + +import logging +from datetime import UTC, datetime, timedelta +from typing import Annotated +from uuid import UUID + +from deerflow_extension_api import AgentRunError +from fastapi import APIRouter, HTTPException, Path, Request +from fastapi.responses import JSONResponse, Response +from pydantic import BaseModel, Field + +from app.gateway.authz import require_permission +from app.gateway.routers.artifacts import _build_content_disposition, _slice_byte_range +from app.gateway.services import reserve_checkpoint_write +from app.gateway.share_fork import fork_share +from app.gateway.share_snapshot import capture_share +from deerflow.persistence.engine import get_session_factory +from deerflow.persistence.thread_shares.sql import ThreadShareRepository +from deerflow.runtime import ConflictError +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.utils.file_io import await_drained +from deerflow.utils.thread_id import ThreadId + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api", tags=["shares"]) +ShareToken = Annotated[str, Path(pattern=r"^[A-Za-z0-9_-]{43}$")] +PRIVATE_HEADERS = {"Cache-Control": "no-store", "Referrer-Policy": "no-referrer", "X-Robots-Tag": "noindex, nofollow", "X-Content-Type-Options": "nosniff"} + + +class ShareCreateRequest(BaseModel): + checkpoint_id: str | None = Field(default=None, max_length=128) + expires_in_seconds: int = Field(default=7 * 86400, ge=3600, le=30 * 86400) + + +class ShareForkRequest(BaseModel): + request_id: UUID + + +def share_repository() -> ThreadShareRepository: + factory = get_session_factory() + if factory is None: + raise HTTPException(503, "Sharing requires SQL persistence") + return ThreadShareRepository(factory) + + +@router.post("/threads/{thread_id}/shares", status_code=201) +@require_permission("threads", "write", owner_check=True, require_existing=True) +@require_permission("runs", "read") +async def create_share(thread_id: ThreadId, body: ShareCreateRequest, request: Request): + async def create(): + repo = share_repository() + async with reserve_checkpoint_write(request, thread_id, user_id=get_effective_user_id()): + if await repo.count_active(thread_id, get_effective_user_id()) >= 10: + raise HTTPException(409, "Revoke an existing link before creating another") + captured = await capture_share(request, thread_id, body.checkpoint_id) + return await repo.create(**captured, expires_at=datetime.now(UTC) + timedelta(seconds=body.expires_in_seconds)) + + try: + result = await await_drained(create()) + return JSONResponse(result, status_code=201, headers=PRIVATE_HEADERS) + except ConflictError: + raise HTTPException(409, "Thread has work in flight; share it after the work finishes") from None + except AgentRunError as exc: + raise HTTPException(exc.status_code, str(exc)) from None + except HTTPException: + raise + except Exception: + logger.exception("Failed to create conversation share") + raise HTTPException(500, "Unable to create share") from None + + +@router.get("/threads/{thread_id}/shares") +@require_permission("threads", "read", owner_check=True, require_existing=True) +async def list_shares(thread_id: ThreadId, request: Request): + return JSONResponse(await share_repository().list_owned(thread_id, get_effective_user_id()), headers=PRIVATE_HEADERS) + + +@router.delete("/threads/{thread_id}/shares/{share_id}", status_code=204) +@require_permission("threads", "write", owner_check=True, require_existing=True) +async def revoke_share(thread_id: ThreadId, share_id: str, request: Request): + if not await await_drained(share_repository().revoke(thread_id, share_id, get_effective_user_id())): + raise HTTPException(404, "Share not found") + return Response(status_code=204, headers=PRIVATE_HEADERS) + + +@router.get("/shares/{share_token}") +async def read_share(share_token: ShareToken): + share = await share_repository().get_public(share_token) + if share is None: + raise HTTPException(404, "Share unavailable", headers=PRIVATE_HEADERS) + return JSONResponse({**share["snapshot"], "checkpoint_id": share["checkpoint_id"], "seq": share["seq"], "created_at": share["created_at"], "expires_at": share["expires_at"], "artifacts": share["artifacts"]}, headers=PRIVATE_HEADERS) + + +@router.get("/shares/{share_token}/artifacts/{path:path}") +@router.head("/shares/{share_token}/artifacts/{path:path}") +async def read_share_artifact(share_token: ShareToken, path: str, request: Request, download: bool = False): + artifact = await share_repository().get_artifact(share_token, path) + if artifact is None: + raise HTTPException(404, "Shared artifact unavailable", headers=PRIVATE_HEADERS) + content, status, range_headers = _slice_byte_range(artifact["content"], request.headers.get("range")) + # Even a direct tab navigation cannot run active content with our origin. + # HTML previews render in an opaque sandbox and cannot contact the network. + headers = { + **PRIVATE_HEADERS, + **range_headers, + "Content-Disposition": _build_content_disposition("attachment" if download else "inline", path.rsplit("/", 1)[-1]), + "Content-Length": str(len(content)), + } + # Chromium's built-in PDF viewer cannot load in a CSP sandbox. The PDF + # media type and nosniff keep it outside the active HTML document surface. + if artifact["mime_type"] != "application/pdf": + headers["Content-Security-Policy"] = "sandbox; default-src 'none'; style-src 'unsafe-inline'; img-src data:; font-src data:; base-uri 'none'; form-action 'none'" + return Response(content if request.method != "HEAD" else b"", status_code=status, media_type=artifact["mime_type"], headers=headers) + + +@router.post("/shares/{share_token}/fork") +@require_permission("threads", "write") +async def import_share(share_token: ShareToken, body: ShareForkRequest, request: Request): + try: + return JSONResponse(await await_drained(fork_share(request, share_repository(), share_token, str(body.request_id))), headers=PRIVATE_HEADERS) + except ConflictError: + raise HTTPException(409, "Import is already in progress; retry shortly") from None + except AgentRunError as exc: + raise HTTPException(exc.status_code, str(exc)) from None + except HTTPException: + raise + except Exception: + logger.exception("Failed to import conversation share") + raise HTTPException(500, "Unable to import share") from None diff --git a/backend/app/gateway/share_fork.py b/backend/app/gateway/share_fork.py new file mode 100644 index 0000000000..817a6e0846 --- /dev/null +++ b/backend/app/gateway/share_fork.py @@ -0,0 +1,135 @@ +"""Import a share into a fresh owner-scoped default-agent conversation.""" + +import os +import uuid +from contextlib import asynccontextmanager + +from fastapi import HTTPException, Request +from langchain_core.messages import AIMessage, HumanMessage +from langgraph.types import Overwrite + +from app.gateway.deps import get_checkpointer, get_run_event_store, get_run_manager, get_thread_store +from app.gateway.extension_artifacts import snapshot_sandbox +from app.gateway.history_seed import reserve_history_seed, write_history_seed +from app.gateway.persistent_writes import run_drained_write +from app.gateway.services import build_checkpoint_state_mutation_accessor +from app.gateway.share_snapshot import MAX_SHARE_FILE_BYTES, OUTPUT_PREFIX, artifact_path +from deerflow.config.paths import get_paths +from deerflow.persistence.thread_shares.sql import ThreadShareRepository +from deerflow.runtime import ThreadOperationKind +from deerflow.runtime.events.store.db import DbRunEventStore +from deerflow.runtime.goal import goal_thread_lock +from deerflow.runtime.journal import build_branch_history_seed_events +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.sandbox.artifact_snapshot import publish_snapshot +from deerflow.utils.file_io import run_file_io + + +def _write_fork_files(owner: str, thread_id: str, artifacts: list[dict]) -> None: + """A fresh UUID directory, descriptor-confined descendants and exclusive files.""" + root = get_paths().sandbox_outputs_dir(thread_id, user_id=owner) + root.mkdir(mode=0o700, parents=True, exist_ok=False) + directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW + root_fd = os.open(root, directory_flags) + try: + for artifact in artifacts: + parts = artifact_path(artifact["path"]).split("/") + directory = os.dup(root_fd) + try: + for part in parts[:-1]: + try: + os.mkdir(part, 0o700, dir_fd=directory) + except FileExistsError: + pass + following = os.open(part, directory_flags, dir_fd=directory) + os.close(directory) + directory = following + os.fchmod(directory, 0o777) + fd = os.open(parts[-1], os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory) + with os.fdopen(fd, "wb") as target: + target.write(artifact["content"]) + target.flush() + os.fchmod(target.fileno(), 0o666) + os.fsync(target.fileno()) + finally: + os.close(directory) + # The bind mount is writable by the sandbox UID even when Gateway runs + # under a different UID or a restrictive umask. + os.fchmod(root_fd, 0o777) + finally: + os.close(root_fd) + + +@asynccontextmanager +async def _reserve_import(request: Request, thread_id: str, owner: str): + if isinstance(get_run_event_store(request), DbRunEventStore): + async with reserve_history_seed(request, thread_id) as reservation: + yield reservation + else: + async with get_run_manager(request).reserve_thread_operation(thread_id, kind=ThreadOperationKind.checkpoint_write, user_id=owner): + yield None + + +async def fork_share(request: Request, repo: ThreadShareRepository, token: str, request_id: str) -> dict: + share = await repo.get_public(token) + if share is None: + raise HTTPException(404, "Share unavailable") + owner = get_effective_user_id() + thread_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"deerflow-share:{owner}:{share['share_id']}:{request_id}")) + threads = get_thread_store(request) + async with goal_thread_lock(thread_id): + existing = await threads.get(thread_id, user_id=owner) + if existing is not None: + if existing.get("metadata", {}).get("share_import_complete") is True: + return {"thread_id": thread_id} + raise HTTPException(409, "The previous import did not finish; create a new copy") + artifacts = [] + for descriptor in share["artifacts"]: + artifact = await repo.get_artifact(token, descriptor["path"]) + if artifact is None: + raise HTTPException(404, "Share unavailable") + artifacts.append(artifact) + # Tool results become ordinary historical assistant text. The new thread + # inherits no tool execution, system prompt, Agent, project or auth grants. + messages = [] + for item in share["snapshot"]["messages"]: + content = item["content"] + if item["type"] == "tool": + content = f"{item['name']}:\n{content}" + cls = HumanMessage if item["type"] == "human" else AIMessage + messages.append(cls(content=content, id=str(uuid.uuid4()))) + title = str(share["snapshot"]["title"])[:256] + # Recheck after loading all bytes, before creating any new workspace state. + if await repo.get_public(token) is None: + raise HTTPException(404, "Share unavailable") + async with _reserve_import(request, thread_id, owner) as reservation: + existing = await threads.get(thread_id, user_id=owner) + if existing is not None: + if existing.get("metadata", {}).get("share_import_complete") is True: + return {"thread_id": thread_id} + raise HTTPException(409, "The previous import did not finish; create a new copy") + created = False + try: + record = await threads.create(thread_id, user_id=owner, display_name=title) + created = True + if artifacts: + await run_drained_write("Import shared artifacts", _write_fork_files, (), owner, thread_id, artifacts) + async with snapshot_sandbox(request, owner, thread_id) as sandbox: + if sandbox is not None: + for artifact in artifacts: + await run_file_io(publish_snapshot, sandbox, artifact["path"], artifact["content"], area="outputs", max_bytes=MAX_SHARE_FILE_BYTES) + accessor, config = build_checkpoint_state_mutation_accessor(request, thread_id=thread_id, as_node="branch") + await accessor.aupdate(config, {"messages": Overwrite(messages), "title": title, "artifacts": Overwrite([OUTPUT_PREFIX + a["path"] for a in artifacts])}, as_node="branch") + events = build_branch_history_seed_events(messages, thread_id=thread_id, run_id_prefix=f"branch-seed-{thread_id}", parent_thread_id=thread_id) + await write_history_seed(request, events, reservation=reservation, seed_kind="branch", thread=record) + await threads.update_metadata(thread_id, {"share_import_complete": True}, user_id=owner) + except Exception: + if created: + # Keep admission until rollback finishes, so a failed import + # cannot be continued while its files/checkpoints are removed. + await get_checkpointer(request).adelete_thread(thread_id) + await get_run_event_store(request).delete_by_thread(thread_id, user_id=owner) + await run_drained_write("Remove failed share import", get_paths().delete_thread_dir, (), thread_id, user_id=owner) + await threads.delete(thread_id, user_id=owner) + raise + return {"thread_id": thread_id} diff --git a/backend/app/gateway/share_snapshot.py b/backend/app/gateway/share_snapshot.py new file mode 100644 index 0000000000..1e02368fb9 --- /dev/null +++ b/backend/app/gateway/share_snapshot.py @@ -0,0 +1,119 @@ +"""Build a portable transcript under the caller's exclusive thread reservation.""" + +import json +import mimetypes +from pathlib import PurePosixPath + +from fastapi import HTTPException, Request + +from app.gateway.conversation_access import _visible_text +from app.gateway.conversation_reader import read_visible_message_page +from app.gateway.deps import get_run_event_store, get_run_manager, get_thread_store +from app.gateway.extension_artifacts import read_snapshot +from app.gateway.services import build_thread_checkpoint_state_accessor +from deerflow.runtime import serialize_channel_values_for_api +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.utils.file_io import run_file_io + +MAX_SHARE_MESSAGES = 10_000 +MAX_SHARE_JSON_BYTES = 10 * 1024 * 1024 +MAX_SHARE_ARTIFACTS = 64 +MAX_SHARE_FILE_BYTES = 20 * 1024 * 1024 +MAX_SHARE_TOTAL_BYTES = 100 * 1024 * 1024 +OUTPUT_PREFIX = "/mnt/user-data/outputs/" + + +def artifact_path(value: str) -> str: + """Only canonical relative output paths are ever persisted or imported.""" + path = value.removeprefix(OUTPUT_PREFIX) + if not path or len(path) > 1024 or path.startswith("/") or "\\" in path or any(ord(c) < 32 for c in path) or any(p in {"", ".", ".."} for p in path.split("/")) or str(PurePosixPath(path)) != path: + raise HTTPException(422, "Invalid shared artifact path") + return path + + +def project_message(message: dict, index: int) -> dict | None: + """Share visible text, not execution metadata, prompts or private attachments.""" + extra = message.get("additional_kwargs") or {} + if not isinstance(extra, dict) or extra.get("hide_from_ui") or message.get("name") == "summary": + return None + visible = _visible_text({"content": message}) + identity = str(message.get("id") or f"message-{index}") + if visible: + role, content = visible + return {"id": identity, "type": "human" if role == "user" else "ai", "content": content} + if message.get("type") == "tool": + content = message.get("content") + if not isinstance(content, str): + content = json.dumps(content, ensure_ascii=False) + return {"id": identity, "type": "tool", "name": str(message.get("name") or "tool"), "content": content} + return None + + +def project_transcript(messages: list[dict]) -> list[dict]: + projected: list[dict] = [] + positions: dict[str, int] = {} + for index, message in enumerate(messages): + if not isinstance(message, dict): + continue + item = project_message(message, index) + if item is None: + continue + identity = item["id"].removesuffix("__user") if item["type"] == "human" else item["id"] + if identity in positions: + projected[positions[identity]] = item + else: + positions[identity] = len(projected) + projected.append(item) + return projected + + +async def capture_share(request: Request, thread_id: str, expected_checkpoint: str | None) -> dict: + owner = get_effective_user_id() + record = await get_thread_store(request).get(thread_id, user_id=owner) + if record is None: + raise HTTPException(404, "Thread not found") + accessor, config = await build_thread_checkpoint_state_accessor(request, thread_id=thread_id, fail_closed=True) + state = await accessor.aget(config) + checkpoint_id = (state.config or {}).get("configurable", {}).get("checkpoint_id") + if not checkpoint_id or state.next: + raise HTTPException(409, "Finish the current turn before sharing") + if expected_checkpoint is not None and checkpoint_id != expected_checkpoint: + raise HTTPException(409, "Thread changed; refresh before sharing") + rows: list[dict] = [] + cursor = None + while True: + page, more = await read_visible_message_page(event_store=get_run_event_store(request), run_manager=get_run_manager(request), thread_id=thread_id, user_id=owner, limit=200, before_seq=cursor) + rows = page + rows + if len(rows) > MAX_SHARE_MESSAGES: + raise HTTPException(413, "Conversation exceeds the share limit") + if not more: + break + if not page or (cursor is not None and page[0]["seq"] >= cursor): + raise HTTPException(500, "Unable to capture complete conversation history") + cursor = page[0]["seq"] + values = serialize_channel_values_for_api(state.values) + source_messages = [row["content"] for row in rows] if rows else values.get("messages", []) + if len(source_messages) > MAX_SHARE_MESSAGES: + raise HTTPException(413, "Conversation exceeds the share limit") + messages = await run_file_io(project_transcript, source_messages) + if not messages: + raise HTTPException(409, "There are no visible messages to share") + snapshot = {"title": record.get("display_name") or values.get("title") or "Shared conversation", "messages": messages} + if len(json.dumps(snapshot, ensure_ascii=False).encode()) > MAX_SHARE_JSON_BYTES: + raise HTTPException(413, "Conversation exceeds the share limit") + paths = list(dict.fromkeys(values.get("artifacts") or [])) + if len(paths) > MAX_SHARE_ARTIFACTS: + raise HTTPException(413, "Too many artifacts to share") + artifacts: list[dict] = [] + total = 0 + for path in paths: + if not isinstance(path, str) or not path.startswith(OUTPUT_PREFIX): + raise HTTPException(422, "Only output artifacts can be shared") + relative = artifact_path(path) + content = await read_snapshot(request, owner, thread_id, path, max_bytes=MAX_SHARE_FILE_BYTES) + total += len(content) + if total > MAX_SHARE_TOTAL_BYTES: + raise HTTPException(413, "Artifacts exceed the total share limit") + mime = await run_file_io(lambda: mimetypes.guess_type(relative)[0] or "application/octet-stream") + artifacts.append({"path": relative, "content": content, "mime_type": mime}) + return {"thread_id": thread_id, "owner_id": owner, "incarnation": record.get("incarnation"), "checkpoint_id": checkpoint_id, "seq": max((row["seq"] for row in rows), default=0), "snapshot": snapshot, "artifacts": artifacts} diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0055_thread_shares.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0055_thread_shares.py new file mode 100644 index 0000000000..16956d3c4c --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0055_thread_shares.py @@ -0,0 +1,45 @@ +"""Add immutable conversation shares and copied artifact bytes.""" + +import sqlalchemy as sa +from alembic import op + +revision = "0055_thread_shares" +down_revision = "0054_project_memory" +branch_labels = None +depends_on = None + + +def upgrade(): + if not sa.inspect(op.get_bind()).has_table("thread_shares"): + op.create_table( + "thread_shares", + sa.Column("share_id", sa.String(32), primary_key=True), + sa.Column("token_hash", sa.String(64), nullable=False, unique=True), + sa.Column("thread_id", sa.String(64), sa.ForeignKey("threads_meta.thread_id", ondelete="CASCADE"), nullable=False), + sa.Column("owner_id", sa.String(64), nullable=False), + sa.Column("incarnation", sa.String(32)), + sa.Column("checkpoint_id", sa.String(128), nullable=False), + sa.Column("seq", sa.BigInteger(), nullable=False), + sa.Column("snapshot", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("revoked_at", sa.DateTime(timezone=True)), + ) + op.create_index("ix_thread_shares_owner_thread", "thread_shares", ["owner_id", "thread_id"]) + if not sa.inspect(op.get_bind()).has_table("share_artifacts"): + op.create_table( + "share_artifacts", + sa.Column("share_id", sa.String(32), sa.ForeignKey("thread_shares.share_id", ondelete="CASCADE"), primary_key=True), + sa.Column("path", sa.String(1024), primary_key=True), + sa.Column("mime_type", sa.String(128), nullable=False), + sa.Column("sha256", sa.String(64), nullable=False), + sa.Column("size", sa.BigInteger(), nullable=False), + sa.Column("content", sa.LargeBinary(), nullable=False), + ) + + +def downgrade(): + if sa.inspect(op.get_bind()).has_table("share_artifacts"): + op.drop_table("share_artifacts") + if sa.inspect(op.get_bind()).has_table("thread_shares"): + op.drop_table("thread_shares") diff --git a/backend/packages/harness/deerflow/persistence/models/__init__.py b/backend/packages/harness/deerflow/persistence/models/__init__.py index a7f4c0cbc6..cd0cb8a5e8 100644 --- a/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -42,12 +42,15 @@ from deerflow.persistence.skill_mutations.model import SkillAssetRow, SkillOpera from deerflow.persistence.subagent_batches.model import SubagentBatchItemRow, SubagentBatchRow from deerflow.persistence.thread_meta.model import ThreadMetaRow from deerflow.persistence.thread_reads.model import ThreadReadMarkerRow, ThreadReadVersionRow +from deerflow.persistence.thread_shares.model import ShareArtifactRow, ThreadShareRow from deerflow.persistence.tool_approvals.model import ToolApprovalRow from deerflow.persistence.user.im_notifications import ImNotificationPreferenceRow from deerflow.persistence.user.model import UserPreferenceRow, UserRow from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow __all__ = [ + "ShareArtifactRow", + "ThreadShareRow", "ToolApprovalRow", "ExtensionExecutionLeaseRow", "AgentRow", diff --git a/backend/packages/harness/deerflow/persistence/thread_shares/__init__.py b/backend/packages/harness/deerflow/persistence/thread_shares/__init__.py new file mode 100644 index 0000000000..a9b0b8a41e --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/thread_shares/__init__.py @@ -0,0 +1 @@ +"""Immutable, revocable conversation share snapshots.""" diff --git a/backend/packages/harness/deerflow/persistence/thread_shares/model.py b/backend/packages/harness/deerflow/persistence/thread_shares/model.py new file mode 100644 index 0000000000..8f2fb52d9f --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/thread_shares/model.py @@ -0,0 +1,37 @@ +"""Share payloads and their copied artifacts live in the application database.""" + +from datetime import datetime + +from sqlalchemy import JSON, BigInteger, DateTime, ForeignKey, Index, LargeBinary, String +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base + + +class ThreadShareRow(Base): + __tablename__ = "thread_shares" + + share_id: Mapped[str] = mapped_column(String(32), primary_key=True) + token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) + thread_id: Mapped[str] = mapped_column(String(64), ForeignKey("threads_meta.thread_id", ondelete="CASCADE"), nullable=False) + owner_id: Mapped[str] = mapped_column(String(64), nullable=False) + incarnation: Mapped[str | None] = mapped_column(String(32)) + checkpoint_id: Mapped[str] = mapped_column(String(128), nullable=False) + seq: Mapped[int] = mapped_column(BigInteger, nullable=False) + snapshot: Mapped[dict] = mapped_column(JSON, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + __table_args__ = (Index("ix_thread_shares_owner_thread", "owner_id", "thread_id"),) + + +class ShareArtifactRow(Base): + __tablename__ = "share_artifacts" + + share_id: Mapped[str] = mapped_column(String(32), ForeignKey("thread_shares.share_id", ondelete="CASCADE"), primary_key=True) + path: Mapped[str] = mapped_column(String(1024), primary_key=True) + mime_type: Mapped[str] = mapped_column(String(128), nullable=False) + sha256: Mapped[str] = mapped_column(String(64), nullable=False) + size: Mapped[int] = mapped_column(BigInteger, nullable=False) + content: Mapped[bytes] = mapped_column(LargeBinary, nullable=False) diff --git a/backend/packages/harness/deerflow/persistence/thread_shares/sql.py b/backend/packages/harness/deerflow/persistence/thread_shares/sql.py new file mode 100644 index 0000000000..e484e5b6b7 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/thread_shares/sql.py @@ -0,0 +1,120 @@ +"""Only token digests are stored; each public read rechecks expiry and revocation.""" + +import hashlib +import secrets +import uuid +from datetime import UTC, datetime + +from sqlalchemy import case, delete, func, or_, select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.thread_meta.model import ThreadMetaRow +from deerflow.persistence.thread_shares.model import ShareArtifactRow, ThreadShareRow +from deerflow.utils.time import coerce_iso + + +class ThreadShareRepository: + def __init__(self, session_factory: async_sessionmaker[AsyncSession]): + self._sf = session_factory + + @staticmethod + def _summary(row: ThreadShareRow) -> dict: + return { + key: coerce_iso(value) if isinstance(value, datetime) else value + for key, value in { + "share_id": row.share_id, + "checkpoint_id": row.checkpoint_id, + "seq": row.seq, + "created_at": row.created_at, + "expires_at": row.expires_at, + "revoked_at": row.revoked_at, + }.items() + } + + @staticmethod + def _live(token: str): + return ( + select(ThreadShareRow) + .join(ThreadMetaRow, ThreadMetaRow.thread_id == ThreadShareRow.thread_id) + .where( + ThreadShareRow.token_hash == hashlib.sha256(token.encode()).hexdigest(), + ThreadShareRow.revoked_at.is_(None), + ThreadShareRow.expires_at > datetime.now(UTC), + ThreadMetaRow.user_id == ThreadShareRow.owner_id, + or_(ThreadMetaRow.incarnation == ThreadShareRow.incarnation, ThreadMetaRow.incarnation.is_(None) & ThreadShareRow.incarnation.is_(None)), + ) + ) + + async def create(self, *, thread_id: str, owner_id: str, incarnation: str | None, checkpoint_id: str, seq: int, snapshot: dict, artifacts: list[dict], expires_at: datetime) -> dict: + token = secrets.token_urlsafe(32) + row = ThreadShareRow( + share_id=uuid.uuid4().hex, + token_hash=hashlib.sha256(token.encode()).hexdigest(), + thread_id=thread_id, + owner_id=owner_id, + incarnation=incarnation, + checkpoint_id=checkpoint_id, + seq=seq, + snapshot=snapshot, + created_at=datetime.now(UTC), + expires_at=expires_at, + ) + async with self._sf() as session: + session.add(row) + await session.flush() + for artifact in artifacts: + content = artifact["content"] + session.add(ShareArtifactRow(share_id=row.share_id, path=artifact["path"], content=content, mime_type=artifact["mime_type"], size=len(content), sha256=hashlib.sha256(content).hexdigest())) + await session.commit() + return {**self._summary(row), "token": token} + + async def count_active(self, thread_id: str, owner_id: str) -> int: + async with self._sf() as session: + return ( + await session.scalar( + select(func.count()).select_from(ThreadShareRow).where(ThreadShareRow.thread_id == thread_id, ThreadShareRow.owner_id == owner_id, ThreadShareRow.revoked_at.is_(None), ThreadShareRow.expires_at > datetime.now(UTC)) + ) + or 0 + ) + + async def list_owned(self, thread_id: str, owner_id: str) -> list[dict]: + async with self._sf() as session: + active = ThreadShareRow.revoked_at.is_(None) & (ThreadShareRow.expires_at > datetime.now(UTC)) + unrevoked = ThreadShareRow.revoked_at.is_(None) + rows = await session.execute( + select(ThreadShareRow.share_id, ThreadShareRow.checkpoint_id, ThreadShareRow.seq, ThreadShareRow.created_at, ThreadShareRow.expires_at, ThreadShareRow.revoked_at) + .where(ThreadShareRow.thread_id == thread_id, ThreadShareRow.owner_id == owner_id) + .order_by(case((active, 0), (unrevoked, 1), else_=2), ThreadShareRow.created_at.desc()) + .limit(100) + ) + return [self._summary(row) for row in rows] + + async def get_public(self, token: str) -> dict | None: + async with self._sf() as session: + row = await session.scalar(self._live(token)) + if row is None: + return None + artifacts = await session.execute(select(ShareArtifactRow.path, ShareArtifactRow.mime_type, ShareArtifactRow.sha256, ShareArtifactRow.size).where(ShareArtifactRow.share_id == row.share_id).order_by(ShareArtifactRow.path)) + return {**self._summary(row), "snapshot": row.snapshot, "artifacts": [dict(a._mapping) for a in artifacts]} + + async def get_artifact(self, token: str, path: str) -> dict | None: + # One statement checks the grant AND retrieves the bytes. No cached grant + # and no path resolution against the live owner's filesystem. + live_ids = self._live(token).with_only_columns(ThreadShareRow.share_id) + async with self._sf() as session: + row = await session.scalar(select(ShareArtifactRow).where(ShareArtifactRow.share_id.in_(live_ids), ShareArtifactRow.path == path)) + return row.to_dict() if row else None + + async def revoke(self, thread_id: str, share_id: str, owner_id: str) -> bool: + async with self._sf() as session: + result = await session.execute( + update(ThreadShareRow) + .where(ThreadShareRow.share_id == share_id, ThreadShareRow.thread_id == thread_id, ThreadShareRow.owner_id == owner_id) + .values(revoked_at=datetime.now(UTC), snapshot={}) + .returning(ThreadShareRow.share_id) + ) + if result.scalar_one_or_none() is None: + return False + await session.execute(delete(ShareArtifactRow).where(ShareArtifactRow.share_id == share_id)) + await session.commit() + return True diff --git a/backend/packages/harness/deerflow/sandbox/artifact_snapshot.py b/backend/packages/harness/deerflow/sandbox/artifact_snapshot.py index 6cc1763d17..768631d4a2 100644 --- a/backend/packages/harness/deerflow/sandbox/artifact_snapshot.py +++ b/backend/packages/harness/deerflow/sandbox/artifact_snapshot.py @@ -22,21 +22,22 @@ logger = logging.getLogger(__name__) _SCRIPT = r""" import base64, hashlib, json, os, stat, sys p = json.loads(sys.argv[1]) +limit = p.get('max_bytes', 2097152) directory = None def stamp(s): return [s.st_dev, s.st_ino, s.st_size, s.st_mtime_ns, s.st_ctime_ns] def opened(name): fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) s = os.fstat(fd) - if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size > 2097152: + if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size > limit: os.close(fd) raise ValueError('Unsafe or oversized file') return os.fdopen(fd, 'rb'), s def digest(name): source, before = opened(name) with source: - data = source.read(2097153) - if len(data) > 2097152 or stamp(before) != stamp(os.fstat(source.fileno())): + data = source.read(limit + 1) + if len(data) > limit or stamp(before) != stamp(os.fstat(source.fileno())): raise ValueError('File changed during snapshot') return hashlib.sha256(data).hexdigest(), len(data) try: @@ -65,7 +66,7 @@ try: fd = os.open(p['temporary'], flags, 0o600, dir_fd=directory) with os.fdopen(fd, 'ab') as target: s = os.fstat(target.fileno()) - if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size != p['offset'] or s.st_size + len(data) > 2097152: + if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size != p['offset'] or s.st_size + len(data) > limit: raise ValueError('Staging file changed') target.write(data) target.flush() @@ -132,23 +133,23 @@ def validate_snapshot_support(sandbox: Sandbox) -> None: _call(sandbox, "probe") -def publish_snapshot(sandbox: Sandbox, filename: str, content: bytes) -> None: - """Publish at most 2 MiB into uploads, reusing only identical regular files. +def publish_snapshot(sandbox: Sandbox, filename: str, content: bytes, *, area: str = "uploads", max_bytes: int = MAX_BYTES) -> None: + """Publish bounded bytes, reusing only identical regular files. Unsafe paths, changed existing files or rejected remote operations raise AgentRunError(422). Existing edits are retained; staging cleanup is attempted even on failure. Run this synchronous command sequence off the event loop. """ - if not isinstance(content, bytes) or len(content) > MAX_BYTES: - raise AgentRunError(422, "Artifact exceeds 2 MiB") + if area not in {"uploads", "outputs"} or not isinstance(content, bytes) or len(content) > max_bytes: + raise AgentRunError(422, "Artifact exceeds the snapshot size limit") temporary = ".artifact-" + uuid.uuid4().hex + ".part" try: for offset in range(0, max(1, len(content)), CHUNK_BYTES): - _call(sandbox, "append", area="uploads", name=filename, temporary=temporary, offset=offset, data=base64.b64encode(content[offset : offset + CHUNK_BYTES]).decode("ascii")) - _call(sandbox, "publish", area="uploads", name=filename, temporary=temporary, sha256=hashlib.sha256(content).hexdigest(), size=len(content)) + _call(sandbox, "append", area=area, max_bytes=max_bytes, name=filename, temporary=temporary, offset=offset, data=base64.b64encode(content[offset : offset + CHUNK_BYTES]).decode("ascii")) + _call(sandbox, "publish", area=area, max_bytes=max_bytes, name=filename, temporary=temporary, sha256=hashlib.sha256(content).hexdigest(), size=len(content)) finally: try: - _call(sandbox, "cleanup", area="uploads", name=filename, temporary=temporary) + _call(sandbox, "cleanup", area=area, name=filename, temporary=temporary) except Exception: logger.warning("Failed to clean remote artifact staging file", exc_info=True) @@ -158,20 +159,20 @@ def verify_snapshot(sandbox: Sandbox, filename: str, sha256: str, size: int) -> _call(sandbox, "verify", area="uploads", name=filename, sha256=sha256, size=size) -def read_snapshot(sandbox: Sandbox, relative_path: str) -> bytes: - """Read at most 2 MiB from outputs with file identity checked across chunks. +def read_snapshot(sandbox: Sandbox, relative_path: str, *, max_bytes: int = MAX_BYTES) -> bytes: + """Read bounded output bytes with file identity checked across chunks. Raise AgentRunError(422) for unsafe paths, changed files or invalid remote replies. Run this synchronous command sequence off the event loop. """ content, stamp = bytearray(), None while True: - result = _call(sandbox, "read", area="outputs", name=relative_path, offset=len(content), stamp=stamp) + result = _call(sandbox, "read", area="outputs", name=relative_path, offset=len(content), stamp=stamp, max_bytes=max_bytes) try: data = base64.b64decode(result["data"], validate=True) size = result["size"] current_stamp = result["stamp"] - valid = type(size) is int and 0 <= size <= MAX_BYTES and len(data) <= CHUNK_BYTES and len(content) + len(data) <= size + valid = type(size) is int and 0 <= size <= max_bytes and len(data) <= CHUNK_BYTES and len(content) + len(data) <= size if not valid or (stamp is not None and stamp != current_stamp) or (not data and len(content) != size): raise ValueError("Invalid remote snapshot") except (ValueError, KeyError, TypeError) as exc: diff --git a/backend/tests/blocking_io/test_share_snapshots.py b/backend/tests/blocking_io/test_share_snapshots.py new file mode 100644 index 0000000000..7618867a12 --- /dev/null +++ b/backend/tests/blocking_io/test_share_snapshots.py @@ -0,0 +1,27 @@ +"""Snapshot capture and Fork copy real files without blocking the event loop.""" + +import asyncio +import os +from types import SimpleNamespace + +import pytest + +from app.gateway import extension_artifacts, share_fork +from app.gateway.persistent_writes import run_drained_write +from deerflow.config.paths import Paths + + +@pytest.mark.asyncio +@pytest.mark.skipif(os.name != "posix", reason="Descriptor-confined snapshots require POSIX") +async def test_share_capture_and_fork_file_publication(tmp_path, monkeypatch): + paths = await asyncio.to_thread(Paths, base_dir=tmp_path) + monkeypatch.setattr(extension_artifacts, "get_paths", lambda: paths) + monkeypatch.setattr(share_fork, "get_paths", lambda: paths) + monkeypatch.setattr(extension_artifacts, "get_sandbox_provider", lambda: SimpleNamespace(uses_thread_data_mounts=True)) + content = b"share data\x00\xff" * 1024 + artifact = {"path": "nested/report.txt", "content": content} + await run_drained_write("Write source outputs", share_fork._write_fork_files, (), "alice", "source", [artifact]) + captured = await extension_artifacts.read_snapshot(None, "alice", "source", "/mnt/user-data/outputs/nested/report.txt", max_bytes=20 * 1024 * 1024) + await run_drained_write("Import shared outputs", share_fork._write_fork_files, (), "bob", "copy", [{**artifact, "content": captured}]) + actual = await asyncio.to_thread((paths.sandbox_outputs_dir("copy", user_id="bob") / artifact["path"]).read_bytes) + assert actual == content diff --git a/backend/tests/test_migration_0047_mcp_oauth_tokens.py b/backend/tests/test_migration_0047_mcp_oauth_tokens.py index c35ffc59e1..551239a414 100644 --- a/backend/tests/test_migration_0047_mcp_oauth_tokens.py +++ b/backend/tests/test_migration_0047_mcp_oauth_tokens.py @@ -44,9 +44,10 @@ pytestmark = pytest.mark.asyncio async def test_0047_remains_in_the_single_migration_chain(): - assert bootstrap._get_head_revision() == "0054_project_memory" + assert bootstrap._get_head_revision() == "0055_thread_shares" script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR)) - assert script.get_heads() == ["0054_project_memory"] + assert script.get_heads() == ["0055_thread_shares"] + assert script.get_revision("0055_thread_shares").down_revision == "0054_project_memory" assert script.get_revision("0054_project_memory").down_revision == "0053_tool_approvals" assert script.get_revision("0053_tool_approvals").down_revision == "0052_batch_continuation" assert script.get_revision("0052_batch_continuation").down_revision == "0051_merge_run_inputs_fencing" diff --git a/backend/tests/test_migration_0051_run_inputs_fencing.py b/backend/tests/test_migration_0051_run_inputs_fencing.py index e98e90e318..ba3d80c0e6 100644 --- a/backend/tests/test_migration_0051_run_inputs_fencing.py +++ b/backend/tests/test_migration_0051_run_inputs_fencing.py @@ -52,7 +52,7 @@ async def test_both_migration_branches_upgrade_to_one_head(tmp_path, backend, pr ) await asyncio.to_thread(command.upgrade, cfg, "head") async with engine.connect() as conn: - assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == ["0054_project_memory"] + assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == [bootstrap._get_head_revision()] columns = await conn.run_sync(lambda sync: {c["name"] for c in inspect(sync).get_columns("runs")}) assert {"run_inputs_json", "run_incarnation", "producer_epoch", "producer_state"} <= columns assert (await conn.execute(text("SELECT status,run_inputs_json,run_incarnation,producer_epoch,producer_state FROM runs WHERE run_id='old'"))).one() == ("success", None, None, None, None) diff --git a/backend/tests/test_migration_0054_project_memory.py b/backend/tests/test_migration_0054_project_memory.py index 8f4852173f..248ac1096e 100644 --- a/backend/tests/test_migration_0054_project_memory.py +++ b/backend/tests/test_migration_0054_project_memory.py @@ -45,7 +45,7 @@ async def test_bootstrap_upgrades_tool_approvals_database_without_losing_data(tm await conn.execute(text("INSERT INTO tool_approvals (request_id,user_id,thread_id,incarnation,run_id,fingerprint,kind,status,payload,created_at) VALUES ('r1','u','t','i','run','fingerprint','audit','consumed','{}',1)")) await bootstrap.bootstrap_schema(engine, backend="sqlite") async with engine.connect() as conn: - assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == ["0054_project_memory"] + assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == [bootstrap._get_head_revision()] assert (await conn.execute(text("SELECT instructions, memory FROM projects WHERE id='p1'"))).one() == ("Keep this instruction", None) assert (await conn.execute(text("SELECT status FROM tool_approvals WHERE request_id='r1'"))).scalar_one() == "consumed" finally: diff --git a/backend/tests/test_migration_0055_thread_shares.py b/backend/tests/test_migration_0055_thread_shares.py new file mode 100644 index 0000000000..db9293677f --- /dev/null +++ b/backend/tests/test_migration_0055_thread_shares.py @@ -0,0 +1,59 @@ +"""Migration and ORM shapes agree; upgrades preserve existing conversations.""" + +import asyncio +import os +import uuid +from datetime import UTC, datetime + +import pytest +from alembic import command +from alembic.autogenerate import compare_metadata +from alembic.migration import MigrationContext +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine +from support.postgres import asyncpg_test_url + +from deerflow.persistence import bootstrap +from deerflow.persistence.base import Base +from deerflow.persistence.postgres_schema import build_asyncpg_connect_args +from deerflow.persistence.thread_meta.model import ThreadMetaRow + + +@pytest.mark.asyncio +@pytest.mark.parametrize("backend", ["sqlite", "postgres"]) +async def test_share_upgrade_downgrade_and_schema(tmp_path, backend): + schema = None + if backend == "sqlite": + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'migration.db'}") + else: + uri = os.environ.get("TEST_POSTGRES_URI") + if not uri: + pytest.skip("requires TEST_POSTGRES_URI") + schema = "share_migration_" + uuid.uuid4().hex + engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema)) + config = bootstrap._get_alembic_config(engine, postgres_schema=schema or "") + try: + if schema: + async with engine.begin() as conn: + await conn.execute(text(f'CREATE SCHEMA "{schema}"')) + await asyncio.to_thread(command.upgrade, config, "0054_project_memory") + async with engine.begin() as conn: + await conn.execute( + ThreadMetaRow.__table__.insert().values(thread_id="existing", user_id="alice", display_name="Existing conversation", status="idle", metadata_json={}, created_at=datetime.now(UTC), updated_at=datetime.now(UTC)) + ) + await asyncio.to_thread(command.upgrade, config, "0055_thread_shares") + async with engine.connect() as conn: + assert (await conn.execute(text("SELECT display_name FROM threads_meta WHERE thread_id='existing'"))).scalar_one() == "Existing conversation" + + def schema_diff(sync): + context = MigrationContext.configure(sync, opts={"compare_type": True, "include_object": lambda obj, name, type_, reflected, compare_to: type_ != "table" or name in {"thread_shares", "share_artifacts"}}) + return compare_metadata(context, Base.metadata) + + assert await conn.run_sync(schema_diff) == [] + await asyncio.to_thread(command.downgrade, config, "0054_project_memory") + await asyncio.to_thread(command.upgrade, config, "0055_thread_shares") + finally: + if schema: + async with engine.begin() as conn: + await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await engine.dispose() diff --git a/backend/tests/test_remote_artifact_snapshots.py b/backend/tests/test_remote_artifact_snapshots.py index c46c25382f..0953be3d2f 100644 --- a/backend/tests/test_remote_artifact_snapshots.py +++ b/backend/tests/test_remote_artifact_snapshots.py @@ -109,3 +109,13 @@ def test_remote_capture_rejects_replacement_between_chunks(remote, tmp_path): remote.execute_command = command with pytest.raises(AgentRunError): read_snapshot(remote, "report.txt") + + +def test_shared_outputs_keep_larger_bytes_and_nested_paths(remote): + from deerflow.sandbox.artifact_snapshot import publish_snapshot, read_snapshot + + content = b"x" * (MAX_BYTES + 1) + publish_snapshot(remote, "charts/report.bin", content, area="outputs", max_bytes=3 * 1024 * 1024) + assert read_snapshot(remote, "charts/report.bin", max_bytes=3 * 1024 * 1024) == content + with pytest.raises(AgentRunError): + read_snapshot(remote, "charts/report.bin") diff --git a/backend/tests/test_share_snapshots.py b/backend/tests/test_share_snapshots.py new file mode 100644 index 0000000000..c5d93dd44e --- /dev/null +++ b/backend/tests/test_share_snapshots.py @@ -0,0 +1,141 @@ +"""Share capabilities persist copies, never grant access to the source thread.""" + +import os +import uuid +from datetime import UTC, datetime, timedelta + +import pytest +import pytest_asyncio +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from support.postgres import asyncpg_test_url + +from deerflow.persistence.base import Base +from deerflow.persistence.postgres_schema import build_asyncpg_connect_args +from deerflow.persistence.thread_meta.model import ThreadMetaRow +from deerflow.persistence.thread_shares.model import ShareArtifactRow, ThreadShareRow +from deerflow.persistence.thread_shares.sql import ThreadShareRepository + + +@pytest_asyncio.fixture(params=["sqlite", "postgres"]) +async def repository(tmp_path, request): + schema = None + if request.param == "sqlite": + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'shares.db'}") + else: + uri = os.environ.get("TEST_POSTGRES_URI") + if not uri: + pytest.skip("requires TEST_POSTGRES_URI") + schema = "share_repository_" + uuid.uuid4().hex + engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema)) + async with engine.begin() as connection: + if schema: + await connection.execute(text(f'CREATE SCHEMA "{schema}"')) + await connection.run_sync(Base.metadata.create_all, tables=[ThreadMetaRow.__table__, ThreadShareRow.__table__, ShareArtifactRow.__table__]) + factory = async_sessionmaker(engine, expire_on_commit=False) + async with factory() as session: + session.add(ThreadMetaRow(thread_id="source", user_id="alice", incarnation="original")) + await session.commit() + try: + yield ThreadShareRepository(factory), factory + finally: + if schema: + async with engine.begin() as connection: + await connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await engine.dispose() + + +async def create_share(repo, *, expires_at=None): + return await repo.create( + thread_id="source", + owner_id="alice", + incarnation="original", + checkpoint_id="checkpoint-1", + seq=42, + snapshot={"title": "Report", "messages": [{"type": "human", "content": "Question"}]}, + artifacts=[{"path": "report.html", "content": b"

Original

", "mime_type": "text/html"}], + expires_at=expires_at or datetime.now(UTC) + timedelta(days=1), + ) + + +@pytest.mark.asyncio +async def test_snapshot_is_durable_and_token_is_not_stored(repository): + repo, factory = repository + created = await create_share(repo) + restarted = ThreadShareRepository(factory) + snapshot = await restarted.get_public(created["token"]) + assert snapshot["snapshot"]["messages"][0]["content"] == "Question" + assert snapshot["checkpoint_id"] == "checkpoint-1" + assert snapshot["seq"] == 42 + assert (await restarted.get_artifact(created["token"], "report.html"))["content"] == b"

Original

" + async with factory() as session: + row = await session.get(ThreadShareRow, created["share_id"]) + assert created["token"] not in str(row.to_dict()) + assert "token" not in (await restarted.list_owned("source", "alice"))[0] + assert await restarted.list_owned("source", "bob") == [] + assert await restarted.get_public("invalid") is None + assert await restarted.get_artifact(created["token"], "../secret") is None + + +@pytest.mark.asyncio +async def test_revocation_checks_owner_and_closes_every_surface(repository): + repo, _ = repository + share = await create_share(repo) + assert not await repo.revoke("source", share["share_id"], "bob") + assert await repo.get_public(share["token"]) is not None + assert await repo.revoke("source", share["share_id"], "alice") + assert await repo.get_public(share["token"]) is None + assert await repo.get_artifact(share["token"], "report.html") is None + + +@pytest.mark.asyncio +async def test_expired_and_recreated_sources_cannot_authorize_old_links(repository): + repo, factory = repository + expired = await create_share(repo, expires_at=datetime.now(UTC) - timedelta(seconds=1)) + assert await repo.get_public(expired["token"]) is None + assert await repo.get_artifact(expired["token"], "report.html") is None + current = await create_share(repo) + async with factory() as session: + source = await session.get(ThreadMetaRow, "source") + source.incarnation = "replacement" + await session.commit() + assert await repo.get_public(current["token"]) is None + + +@pytest.mark.asyncio +async def test_active_link_remains_revocable_after_many_inactive_links(repository): + repo, factory = repository + active = await create_share(repo) + now = datetime.now(UTC) + async with factory() as session: + session.add_all( + [ + ThreadShareRow( + share_id=f"revoked-{i}", token_hash=f"digest-{i}", thread_id="source", owner_id="alice", incarnation="original", checkpoint_id="cp", seq=0, snapshot={}, created_at=now, expires_at=now + timedelta(days=1), revoked_at=now + ) + for i in range(101) + ] + ) + await session.commit() + assert await repo.count_active("source", "alice") == 1 + assert await repo.count_active("source", "bob") == 0 + assert (await repo.list_owned("source", "alice"))[0]["share_id"] == active["share_id"] + await repo.revoke("source", active["share_id"], "alice") + assert await repo.count_active("source", "alice") == 0 + + +@pytest.mark.asyncio +async def test_expired_snapshot_remains_listed_and_revocation_purges_its_copy(repository): + repo, factory = repository + expired = await create_share(repo, expires_at=datetime.now(UTC) - timedelta(seconds=1)) + for _ in range(101): + current = await create_share(repo) + assert await repo.revoke("source", current["share_id"], "alice") + assert (await repo.list_owned("source", "alice"))[0]["share_id"] == expired["share_id"] + assert await repo.revoke("source", expired["share_id"], "alice") + async with factory() as session: + row = await session.get(ThreadShareRow, expired["share_id"]) + assert row.revoked_at is not None + assert row.snapshot == {} + assert await session.scalar(select(ShareArtifactRow).where(ShareArtifactRow.share_id == expired["share_id"])) is None + assert (await session.get(ThreadMetaRow, "source")).incarnation == "original" diff --git a/backend/tests/test_shares_router.py b/backend/tests/test_shares_router.py new file mode 100644 index 0000000000..ebff1b9835 --- /dev/null +++ b/backend/tests/test_shares_router.py @@ -0,0 +1,535 @@ +"""Real HTTP/auth, SQL shares, checkpoints, history and host file copies.""" + +import asyncio +import json +import os +import stat +from contextlib import asynccontextmanager +from types import SimpleNamespace +from uuid import UUID, uuid4 + +import httpx +import pytest +import pytest_asyncio +from fastapi import FastAPI, HTTPException, Request +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage +from langgraph.checkpoint.memory import InMemorySaver +from sqlalchemy import text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from support.postgres import asyncpg_test_url + +import deerflow.persistence.models # noqa: F401 +from app.gateway import extension_artifacts, history_seed, share_fork, share_snapshot +from app.gateway.auth.models import User +from app.gateway.auth_middleware import AuthMiddleware, _is_public +from app.gateway.csrf_middleware import CSRFMiddleware +from app.gateway.routers import shares, threads +from app.gateway.services import build_checkpoint_state_mutation_accessor +from deerflow.config.authorization_config import AuthorizationConfig +from deerflow.config.paths import Paths +from deerflow.persistence.base import Base +from deerflow.persistence.postgres_schema import build_asyncpg_connect_args +from deerflow.persistence.run.sql import RunRepository +from deerflow.persistence.thread_meta import ThreadMetaRepository +from deerflow.runtime import ConflictError, RunManager, ThreadOperationKind +from deerflow.runtime.events.store.db import DbRunEventStore +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.journal import build_branch_history_seed_events +from deerflow.runtime.user_context import reset_current_user, set_current_user + +ALICE = User(id=UUID(int=1), email="alice@example.com", password_hash="unused", system_role="admin") +BOB = User(id=UUID(int=2), email="bob@example.com", password_hash="unused", system_role="user") +pytestmark = [pytest.mark.asyncio, pytest.mark.skipif(os.name != "posix", reason="descriptor-confined artifact capture requires POSIX")] + + +@pytest_asyncio.fixture(params=[("full", "memory", "sqlite"), ("delta", "memory", "sqlite"), ("full", "db", "sqlite"), ("delta", "db", "sqlite"), ("full", "db", "postgres")]) +async def share_app(tmp_path, monkeypatch, request): + schema = None + if request.param[2] == "postgres": + uri = os.environ.get("TEST_POSTGRES_URI") + if not uri: + pytest.skip("requires TEST_POSTGRES_URI") + schema = "share_http_" + uuid4().hex + engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema)) + else: + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'http.db'}") + async with engine.begin() as connection: + if schema: + await connection.execute(text(f'CREATE SCHEMA "{schema}"')) + await connection.run_sync(Base.metadata.create_all) + factory = async_sessionmaker(engine, expire_on_commit=False) + monkeypatch.setattr(shares, "get_session_factory", lambda: factory) + monkeypatch.setattr("app.gateway.auth_middleware.is_auth_disabled", lambda: False) + monkeypatch.setattr("app.gateway.csrf_middleware.is_auth_disabled", lambda: False) + monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: AuthorizationConfig()) + + async def session_user(req): + user = {"alice": ALICE, "bob": BOB}.get(req.cookies.get("access_token")) + if not user: + raise HTTPException(401, "Invalid session") + return user + + monkeypatch.setattr("app.gateway.deps.get_current_user_from_request", session_user) + paths = Paths(base_dir=tmp_path / "home") + monkeypatch.setattr(extension_artifacts, "get_paths", lambda: paths) + monkeypatch.setattr(share_fork, "get_paths", lambda: paths) + monkeypatch.setattr(extension_artifacts, "get_sandbox_provider", lambda: SimpleNamespace(uses_thread_data_mounts=True)) + app = FastAPI() + app.add_middleware(AuthMiddleware) + app.add_middleware(CSRFMiddleware) + app.state.thread_store = ThreadMetaRepository(factory) + app.state.checkpointer = InMemorySaver() + app.state.run_event_store = DbRunEventStore(factory) if request.param[1] == "db" else MemoryRunEventStore() + app.state.run_manager = RunManager(store=RunRepository(factory), event_store=app.state.run_event_store) + app.state.checkpoint_channel_mode = request.param[0] + app.state.store = None + app.include_router(shares.router) + app.include_router(threads.router) + req = Request({"type": "http", "app": app}) + + async def accessor(req, *, thread_id, **kwargs): + # Replace agent/model assembly only. Materialization and writes use + # the actual full/delta accessor over the real checkpoint store. + return build_checkpoint_state_mutation_accessor(req, thread_id=thread_id, as_node="branch") + + monkeypatch.setattr(share_snapshot, "build_thread_checkpoint_state_accessor", accessor) + await app.state.thread_store.create("source", user_id=str(ALICE.id), display_name="Shared report") + paths.sandbox_outputs_dir("source", user_id=str(ALICE.id)).mkdir(parents=True) + output = paths.sandbox_outputs_dir("source", user_id=str(ALICE.id)) / "report.html" + output.write_text("

Original report

", encoding="utf-8") + messages = [HumanMessage(content="Visible question", id="h1", additional_kwargs={"files": [{"path": "private-upload"}]})] + messages += [AIMessage(content=f"Step {i}", id=f"ai-{i}") for i in range(205)] + messages += [ToolMessage(content="tool result", name="search", tool_call_id="c1", id="tool"), AIMessage(content="Final report", id="answer")] + hidden = AIMessage(content="private control", id="hidden", additional_kwargs={"hide_from_ui": True}) + events = build_branch_history_seed_events([*messages, hidden, SystemMessage(content="secret prompt")], thread_id="source", run_id_prefix="source-run", parent_thread_id="source") + user_token = set_current_user(ALICE) + try: + await app.state.run_event_store.put_batch(events) + finally: + reset_current_user(user_token) + if request.param[1] == "db": + app.state.run_event_store._producer_fencing = "strict" + state, config = await accessor(req, thread_id="source") + config = await state.aupdate(config, {"messages": messages[-1:], "artifacts": ["/mnt/user-data/outputs/report.html"]}, as_node="branch") + try: + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client: + yield SimpleNamespace(app=app, client=client, paths=paths, output=output, state=state, config=config, request=req, factory=factory) + finally: + if schema: + async with engine.begin() as connection: + await connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await engine.dispose() + + +def headers(user="alice", csrf=True): + return {"Cookie": f"access_token={user}; csrf_token=test-csrf", **({"X-CSRF-Token": "test-csrf"} if csrf else {})} + + +async def make_share(ctx): + response = await ctx.client.post("/api/threads/source/shares", json={}, headers=headers()) + assert response.status_code == 201, response.text + return response.json() + + +async def test_anonymous_snapshot_and_files_are_immutable_and_scoped(share_app): + ctx = share_app + share = await make_share(ctx) + url = f"/api/shares/{share['token']}" + response = await ctx.client.get(url) + assert response.status_code == 200 + body = response.json() + assert len(body["messages"]) == 208 # whole paginated history, not the compacted head + assert body["messages"][0]["content"] == "Visible question" + assert body["messages"][-1]["content"] == "Final report" + assert "private-upload" not in response.text and "secret prompt" not in response.text and "private control" not in response.text + assert response.headers["cache-control"] == "no-store" + ctx.output.write_text("changed file", encoding="utf-8") + await ctx.state.aupdate(ctx.config, {"messages": [AIMessage(content="Later private answer", id="later")]}, as_node="branch") + assert (await ctx.client.get(url)).json() == body + artifact = await ctx.client.get(url + "/artifacts/report.html") + assert "Original report" in artifact.text + assert "sandbox" in artifact.headers["content-security-policy"] + assert "allow-scripts" not in artifact.headers["content-security-policy"] + assert artifact.headers["referrer-policy"] == "no-referrer" + ranged = await ctx.client.get(url + "/artifacts/report.html", headers={"Range": "bytes=0-3"}) + assert ranged.status_code == 206 and ranged.content == b"

" + assert (await ctx.client.get(url + "/artifacts/private-upload")).status_code == 404 + assert (await ctx.client.get("/api/threads/source/state")).status_code == 401 + assert (await ctx.client.get("/api/threads/source/state", headers=headers("bob"))).status_code == 404 + + +async def test_share_mutations_require_owner_and_csrf(share_app): + ctx = share_app + assert (await ctx.client.post("/api/threads/source/shares", json={}, headers=headers("bob"))).status_code == 404 + assert (await ctx.client.post("/api/threads/source/shares", json={}, headers=headers(csrf=False))).status_code == 403 + assert (await ctx.client.post("/api/threads/source/shares", json={"expires_in_seconds": 0}, headers=headers())).status_code == 422 + assert (await ctx.client.post("/api/threads/source/shares", json={"checkpoint_id": "old"}, headers=headers())).status_code == 409 + async with ctx.app.state.run_manager.reserve_thread_operation("source", kind=ThreadOperationKind.artifact_write, user_id=str(ALICE.id)): + assert (await ctx.client.post("/api/threads/source/shares", json={}, headers=headers())).status_code == 409 + share = await make_share(ctx) + path = f"/api/threads/source/shares/{share['share_id']}" + assert (await ctx.client.delete(path, headers=headers("bob"))).status_code == 404 + assert (await ctx.client.delete(path, headers=headers())).status_code == 204 + public = f"/api/shares/{share['token']}" + assert (await ctx.client.get(public)).status_code == 404 + assert (await ctx.client.get(public + "/artifacts/report.html")).status_code == 404 + assert (await ctx.client.post(public + "/fork", json={"request_id": str(uuid4())}, headers=headers("bob"))).status_code == 404 + + +async def test_pdf_preview_uses_browser_viewer_without_sandbox(share_app): + ctx = share_app + pdf = b"%PDF-1.4\n% Frozen PDF bytes\n%%EOF" + ctx.output.with_name("report.pdf").write_bytes(pdf) + await ctx.state.aupdate(ctx.config, {"artifacts": ["/mnt/user-data/outputs/report.pdf"]}, as_node="branch") + share = await make_share(ctx) + response = await ctx.client.get(f"/api/shares/{share['token']}/artifacts/report.pdf") + assert response.content == pdf + assert response.headers["content-type"] == "application/pdf" + assert response.headers["x-content-type-options"] == "nosniff" + assert response.headers["cache-control"] == "no-store" + assert "sandbox" not in response.headers.get("content-security-policy", "") + + +async def test_fork_is_owned_by_visitor_and_survives_revocation(share_app): + ctx = share_app + share = await make_share(ctx) + url = f"/api/shares/{share['token']}/fork" + body = {"request_id": str(uuid4())} + assert (await ctx.client.post(url, json=body)).status_code in {401, 403} + response = await ctx.client.post(url, json=body, headers=headers("bob")) + assert response.status_code == 200, response.text + new_id = response.json()["thread_id"] + assert (await ctx.client.post(url, json=body, headers=headers("bob"))).json() == {"thread_id": new_id} + row = await ctx.app.state.thread_store.get(new_id, user_id=str(BOB.id)) + assert row and row["assistant_id"] is None + assert await ctx.app.state.thread_store.get(new_id, user_id=str(ALICE.id)) is None + accessor, config = build_checkpoint_state_mutation_accessor(ctx.request, thread_id=new_id, as_node="branch") + state = await accessor.aget(config) + assert len(state.values["messages"]) == 208 + assert all(m.type in {"human", "ai"} for m in state.values["messages"]) + assert "sandbox" not in state.values + assert len(await ctx.app.state.run_event_store.list_messages(new_id, limit=1000, user_id=str(BOB.id))) == 208 + copied = ctx.paths.sandbox_outputs_dir(new_id, user_id=str(BOB.id)) / "report.html" + assert copied.read_bytes() == ctx.output.read_bytes() + await ctx.client.delete(f"/api/threads/source/shares/{share['share_id']}", headers=headers()) + assert copied.exists() + # Continue with a real checkpoint write in the new owner's conversation. + await accessor.aupdate(config, {"messages": [HumanMessage(content="Follow up", id="followup")]}, as_node="branch") + assert (await accessor.aget(config)).values["messages"][-1].content == "Follow up" + + +async def test_fork_rechecks_completion_after_durable_admission(share_app, monkeypatch): + ctx = share_app + if not isinstance(ctx.app.state.run_event_store, DbRunEventStore): + pytest.skip("requires database-backed history and admission") + share = await make_share(ctx) + url = f"/api/shares/{share['token']}/fork" + body = {"request_id": str(uuid4())} + delayed = asyncio.Event() + resume = asyncio.Event() + reserves = 0 + original_reserve = share_fork._reserve_import + first_manager = ctx.app.state.run_manager + second_manager = RunManager(store=RunRepository(ctx.factory), event_store=ctx.app.state.run_event_store) + + @asynccontextmanager + async def separate_worker_lock(_thread_id): + yield + + @asynccontextmanager + async def park_before_reservation(*args): + nonlocal reserves + reserves += 1 + if reserves == 1: + delayed.set() + await resume.wait() + async with original_reserve(*args) as reservation: + yield reservation + + monkeypatch.setattr(share_fork, "goal_thread_lock", separate_worker_lock) + monkeypatch.setattr(share_fork, "_reserve_import", park_before_reservation) + monkeypatch.setattr(history_seed, "get_run_manager", lambda request: second_manager if request.headers.get("X-Test-Worker") == "second" else first_manager) + first = asyncio.create_task(ctx.client.post(url, json=body, headers=headers("bob"))) + async with asyncio.timeout(10): + try: + await delayed.wait() + second = await ctx.client.post(url, json=body, headers={**headers("bob"), "X-Test-Worker": "second"}) + assert second.status_code == 200, second.text + thread_id = second.json()["thread_id"] + copied = ctx.paths.sandbox_outputs_dir(thread_id, user_id=str(BOB.id)) / "report.html" + copied.write_bytes(b"Edited by the recipient") + finally: + resume.set() + duplicate = await first + assert duplicate.status_code == 200, duplicate.text + assert duplicate.json() == {"thread_id": thread_id} + assert [row["thread_id"] for row in await ctx.app.state.thread_store.search(user_id=str(BOB.id))] == [thread_id] + assert copied.read_bytes() == b"Edited by the recipient" + assert len(await ctx.app.state.run_event_store.list_messages(thread_id, limit=1000, user_id=str(BOB.id))) == 208 + + +async def test_fork_continues_under_recipient_run_authority(share_app, monkeypatch, tmp_path): + from _agent_e2e_helpers import FakeToolCallingModel + + from app.gateway.routers import thread_runs + from deerflow.agents.lead_agent import agent + from deerflow.config import app_config + from deerflow.config.extensions_config import ExtensionsConfig + from deerflow.mcp import cache, tools + from deerflow.persistence.projects import ProjectRepository + from deerflow.persistence.thread_meta.base import THREAD_PROJECT_METADATA_KEY + from deerflow.persistence.thread_meta.model import ThreadMetaRow + from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge + + ctx = share_app + if not isinstance(ctx.app.state.run_event_store, DbRunEventStore) or ctx.app.state.checkpoint_channel_mode != "full": + pytest.skip("exercises the gateway runtime with durable full checkpoints") + publisher, recipient = str(ALICE.id), str(BOB.id) + monkeypatch.setenv("DEER_FLOW_HOME", str(ctx.paths.base_dir)) + monkeypatch.setattr("deerflow.config.paths._paths", ctx.paths) + skill_dir = ctx.paths.user_custom_skills_dir(publisher) / "publisher-skill" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text("---\nname: publisher-skill\ndescription: Publisher-only skill\n---\nPublisher-only skill body.", encoding="utf-8") + custom_agent = ctx.paths.user_agent_dir(publisher, "publisher-agent") + custom_agent.mkdir(parents=True) + (custom_agent / "config.yaml").write_text("name: publisher-agent\nskills: [publisher-skill]\n", encoding="utf-8") + (custom_agent / "SOUL.md").write_text("Publisher-only agent instructions.", encoding="utf-8") + from deerflow.mcp.user_config import load_user_mcp_config, user_mcp_config_path + + personal_config = user_mcp_config_path(publisher) + personal_config.parent.mkdir(parents=True, exist_ok=True) + personal_config.write_text(json.dumps({"mcpServers": {"publisher": {"type": "http", "url": "https://fixture.example/personal", "headers": {"Authorization": "Bearer personal-publisher"}}}}), encoding="utf-8") + assert load_user_mcp_config(publisher).mcp_servers + assert not load_user_mcp_config(recipient).mcp_servers + async with ctx.factory() as session: + source = await session.get(ThreadMetaRow, "source") + source.assistant_id = "publisher-agent" + await session.commit() + project = await ProjectRepository(ctx.factory).create(name="Publisher project", instructions="Publisher-only instructions", user_id=publisher) + await ctx.app.state.thread_store.set_project("source", project["id"], user_id=publisher) + await ctx.state.aupdate(ctx.config, {"messages": [HumanMessage(content="Publish this", additional_kwargs={"slash_skill_activation": {"skill_name": "publisher-skill"}})]}, as_node="branch") + share = await make_share(ctx) + fork = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": str(uuid4())}, headers=headers("bob")) + assert fork.status_code == 200, fork.text + destination = fork.json()["thread_id"] + copied = await ctx.app.state.thread_store.get(destination, user_id=recipient) + assert THREAD_PROJECT_METADATA_KEY not in copied["metadata"] and copied["assistant_id"] is None + accessor, checkpoint = build_checkpoint_state_mutation_accessor(ctx.request, thread_id=destination, as_node="branch") + copied_messages = (await accessor.aget(checkpoint)).values["messages"] + assert all("slash_skill_activation" not in message.additional_kwargs for message in copied_messages) + + config = app_config.AppConfig.model_validate( + { + "models": [{"name": "share-test", "display_name": "Share test", "use": "langchain_openai:ChatOpenAI", "model": "unused"}], + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + "skills": {"path": str(tmp_path / "skills")}, + "tool_search": {"enabled": False}, + "memory": {"enabled": False}, + "title": {"enabled": False}, + "authorization": { + "enabled": True, + "provider": { + "use": "deerflow.authz.rbac:RbacAuthorizationProvider", + "config": {"roles": {"admin": {}, "user": {"tools": {"allow": "*", "deny": ["denied_tool"]}, "skills": {"allow": "*", "deny": ["restricted-skill"]}}}}, + }, + }, + } + ) + restricted = tmp_path / "skills" / "public" / "restricted-skill" + restricted.mkdir(parents=True) + (restricted / "SKILL.md").write_text("---\nname: restricted-skill\ndescription: Role-restricted skill\n---\nRole-restricted skill body.", encoding="utf-8") + monkeypatch.setattr(app_config, "_app_config", config) + monkeypatch.setattr(app_config, "_app_config_is_custom", True) + monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config.authorization) + ctx.app.state.project_repo = ProjectRepository(ctx.factory) + ctx.app.state.stream_bridge = MemoryStreamBridge() + ctx.app.include_router(thread_runs.router) + + requests = [] + visited_paths = [] + + def mcp_http(request): + visited_paths.append(request.url.path) + if request.method != "POST": + return httpx.Response(405) + payload = json.loads(request.content) + method = payload["method"] + if "id" not in payload: + return httpx.Response(202) + if method == "initialize": + result = {"protocolVersion": payload["params"]["protocolVersion"], "capabilities": {"tools": {}}, "serverInfo": {"name": "fixture", "version": "1"}} + elif method == "tools/list": + names = {"scoped": ["scoped_probe"], "deployment": ["deployment_probe", "denied_tool"], "personal": ["publisher_only"]}[request.url.path.lstrip("/")] + result = {"tools": [{"name": name, "inputSchema": {"type": "object"}, "description": name} for name in names]} + else: + assert method == "tools/call" + requests.append((payload["params"]["name"], request.headers.get("authorization"))) + result = {"content": [{"type": "text", "text": "fixture success"}]} + return httpx.Response(200, json={"jsonrpc": "2.0", "id": payload["id"], "result": result}) + + def client_factory(headers=None, timeout=None, auth=None): + return httpx.AsyncClient(headers=headers, timeout=timeout, auth=auth, transport=httpx.MockTransport(mcp_http)) + + original_build = tools.build_servers_config + + def build_servers(extensions): + return {name: {**params, "httpx_client_factory": client_factory} for name, params in original_build(extensions).items()} + + monkeypatch.setattr(tools, "build_servers_config", build_servers) + calls = [] + model_tool_names = [] + + class RecordingModel(FakeToolCallingModel): + def bind_tools(self, available, **kwargs): + model_tool_names.append({tool.name for tool in available}) + return self + + def _generate(self, messages, *args, **kwargs): + prompt = "\n".join(str(message.content) for message in messages) + assert "Publisher-only instructions" not in prompt + assert "Publisher-only agent instructions" not in prompt + assert "Publisher-only skill" not in prompt + assert "Role-restricted skill" not in prompt + return super()._generate(messages, *args, **kwargs) + + def create_model(**kwargs): + return RecordingModel( + responses=[ + AIMessage( + content="", + tool_calls=[{"name": name, "args": {}, "id": name} for name in ("scoped_probe", "deployment_probe", "denied_tool")], + ), + AIMessage(content="Recipient continuation finished."), + ] + ) + + async def run_tool_observer(request, handler): + context = request.runtime.context + calls.append((context["user_id"], context["user_role"], context.get("agent_name"), context.get("project_id"))) + return await handler(request) + + from deerflow.mcp import interceptors + + original_interceptors = tools.build_mcp_tool_interceptors + + def build_interceptors(*args, **kwargs): + return [run_tool_observer, *original_interceptors(*args, **kwargs)] + + monkeypatch.setattr(tools, "build_mcp_tool_interceptors", build_interceptors) + monkeypatch.setattr(interceptors, "build_mcp_tool_interceptors", build_interceptors) + monkeypatch.setattr(agent, "create_chat_model", create_model) + for recipient_credential in ("Bearer recipient", None): + credentials = {publisher: "Bearer publisher"} + if recipient_credential: + credentials[recipient] = recipient_credential + extensions = ExtensionsConfig.model_validate( + { + "mcpServers": { + "scoped": {"type": "http", "url": "https://fixture.example/scoped", "tool_name_prefix": False, "headers": {"Authorization": "Bearer discovery"}, "user_auth": {"users": credentials}}, + "deployment": {"type": "http", "url": "https://fixture.example/deployment", "tool_name_prefix": False, "headers": {"Authorization": "Bearer deployment"}}, + } + } + ) + monkeypatch.setattr(ExtensionsConfig, "from_file", classmethod(lambda cls, *args, **kwargs: extensions)) + loaded = await tools.get_mcp_tools(extensions) + assert {tool.name for tool in loaded} == {"scoped_probe", "deployment_probe", "denied_tool"} + monkeypatch.setattr(cache, "get_cached_mcp_tools", lambda: loaded) + requests.clear() + result = await ctx.client.post(f"/api/threads/{destination}/runs/wait", json={"input": {"messages": [{"type": "human", "content": "Continue the report"}]}}, headers=headers("bob")) + assert result.status_code == 200, result.text + assert "Recipient continuation finished." in result.text + assert set(requests) == {("deployment_probe", "Bearer deployment")} | ({("scoped_probe", recipient_credential)} if recipient_credential else set()) + assert calls and all(call == (recipient, "user", None, None) for call in calls) + assert model_tool_names and all("denied_tool" not in names for names in model_tool_names) + assert "/personal" not in visited_paths + + +async def test_imported_nested_files_have_sandbox_write_permissions(share_app): + ctx = share_app + nested = ctx.output.parent / "reports" / "week" + nested.mkdir(parents=True) + (nested / "summary.txt").write_bytes(b"Report") + await ctx.state.aupdate(ctx.config, {"artifacts": ["/mnt/user-data/outputs/reports/week/summary.txt"]}, as_node="branch") + share = await make_share(ctx) + previous_umask = os.umask(0o077) + try: + result = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": str(uuid4())}, headers=headers("bob")) + finally: + os.umask(previous_umask) + assert result.status_code == 200, result.text + root = ctx.paths.sandbox_outputs_dir(result.json()["thread_id"], user_id=str(BOB.id)) + output = root / "reports" / "week" / "summary.txt" + assert output.read_bytes() == b"Report" + assert stat.S_IMODE(output.stat().st_mode) == 0o666 + for directory in (root, root / "reports", output.parent): + assert stat.S_IMODE(directory.stat().st_mode) == 0o777 + + +async def test_unsafe_artifact_fails_without_publishing_partial_share(share_app): + ctx = share_app + ctx.output.unlink() + ctx.output.symlink_to(ctx.paths.base_dir / "secret.txt") + (ctx.paths.base_dir / "secret.txt").write_text("Do not share", encoding="utf-8") + response = await ctx.client.post("/api/threads/source/shares", json={}, headers=headers()) + assert response.status_code == 422, response.text + assert (await ctx.client.get("/api/threads/source/shares", headers=headers())).json() == [] + + +async def test_public_allowlist_never_grants_mutation_authority(): + path = "/api/shares/" + "x" * 43 + for method in ("POST", "PATCH", "PUT", "DELETE"): + assert not _is_public(path, method) + assert not _is_public(path + "/artifacts/report.html", method) + assert not _is_public(path + "/fork", "GET") + assert _is_public(path, "GET") + assert _is_public(path + "/artifacts/report.html", "HEAD") + + +async def test_failed_fork_rolls_back_metadata_checkpoint_history_and_files(share_app, monkeypatch): + ctx = share_app + share = await make_share(ctx) + reached = [] + original = share_fork.write_history_seed + request_id = str(uuid4()) + + async def fail_seed(*args, **kwargs): + await original(*args, **kwargs) + reached.append(True) + raise RuntimeError("private filesystem diagnostic") + + monkeypatch.setattr(share_fork, "write_history_seed", fail_seed) + response = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": request_id}, headers=headers("bob")) + assert reached == [True] + assert response.status_code == 500 and "private filesystem" not in response.text + assert await ctx.app.state.thread_store.search(user_id=str(BOB.id)) == [] + assert list(ctx.paths.base_dir.rglob("report.html")) == [ctx.output] + monkeypatch.setattr(share_fork, "write_history_seed", original) + retry = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": request_id}, headers=headers("bob")) + assert retry.status_code == 200, retry.text + assert len(await ctx.app.state.run_event_store.list_messages(retry.json()["thread_id"], limit=1000, user_id=str(BOB.id))) == 208 + + +async def test_disconnect_drains_capture_before_releasing_source(share_app, monkeypatch): + ctx = share_app + entered, release = asyncio.Event(), asyncio.Event() + original = shares.capture_share + + async def slow_capture(*args, **kwargs): + entered.set() + await release.wait() + return await original(*args, **kwargs) + + monkeypatch.setattr(shares, "capture_share", slow_capture) + task = asyncio.create_task(ctx.client.post("/api/threads/source/shares", json={}, headers=headers())) + try: + await asyncio.wait_for(entered.wait(), 5) + task.cancel() + # A competing writer must not enter while the cancelled request drains. + with pytest.raises(ConflictError): + async with ctx.app.state.run_manager.reserve_thread_operation("source", kind=ThreadOperationKind.artifact_write, user_id=str(ALICE.id)): + pytest.fail("capture released its reservation before finishing") + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + links = await ctx.client.get("/api/threads/source/shares", headers=headers()) + assert len(links.json()) == 1 diff --git a/backend/tests/test_thread_id_route_contract.py b/backend/tests/test_thread_id_route_contract.py index 4dd716135c..0bd37b350b 100644 --- a/backend/tests/test_thread_id_route_contract.py +++ b/backend/tests/test_thread_id_route_contract.py @@ -78,6 +78,7 @@ def _collect_thread_id_routes(): runs, scheduled_tasks, search, + shares, skills, subagent_batches, suggestions, @@ -95,6 +96,7 @@ def _collect_thread_id_routes(): runs, scheduled_tasks, search, + shares, skills, subagent_batches, suggestions, diff --git a/docs/guide/assets/sharing/share-entry.png b/docs/guide/assets/sharing/share-entry.png new file mode 100644 index 0000000000..15c9892071 Binary files /dev/null and b/docs/guide/assets/sharing/share-entry.png differ diff --git a/docs/guide/assets/sharing/share-menu.png b/docs/guide/assets/sharing/share-menu.png new file mode 100644 index 0000000000..acf6caeaa6 Binary files /dev/null and b/docs/guide/assets/sharing/share-menu.png differ diff --git a/docs/guide/assets/sharing/share-preview.png b/docs/guide/assets/sharing/share-preview.png new file mode 100644 index 0000000000..047450ad04 Binary files /dev/null and b/docs/guide/assets/sharing/share-preview.png differ diff --git a/docs/guide/sharing.md b/docs/guide/sharing.md new file mode 100644 index 0000000000..c248f0c754 --- /dev/null +++ b/docs/guide/sharing.md @@ -0,0 +1,77 @@ +# Share a conversation snapshot + +Open a conversation's sidebar menu, choose **Share**, select an expiry, then +choose **Create snapshot link**. Copy the generated `/share/` URL. Anyone +with that link can read the snapshot without signing in. The dialog lists your +links and lets you revoke them. Revocation and expiry stop subsequent reads, +file downloads and imports; recipients keep copies they already downloaded or +imported. + +![The conversation menu's Share action](assets/sharing/share-menu.png) + +![Create, copy and revoke a snapshot link](assets/sharing/share-entry.png) + +The snapshot contains the visible conversation history and the output files +presented as artifacts at creation time. Later messages and file edits do not +change it. Uploads, hidden messages, system prompts and execution metadata are +excluded. Tool result text is included, so review the conversation before +sharing it. A source conversation must have finished its current turn. Deleting +the source conversation also makes its links unavailable. + +The shared page has no composer or source workspace controls. It previews +Markdown, text, CSV/TSV, images, PDF and static HTML; other files are available +for download. Text previews show at most 512 KiB. HTML scripts, forms and network +requests are blocked. The page rechecks availability on return and every +30 seconds while visible. + +![A public read-only conversation snapshot](assets/sharing/share-preview.png) + +Choose **Continue in my workspace** to create an independent conversation with +copied history and output files. Visitors sign in first and return to the shared +page. The copy uses their default agent and permissions. It does not inherit the +source's agent, project, sandbox, uploads, tool calls or credentials. Tool results +are imported as historical assistant text. The copied conversation remains +usable after the source link is revoked. + +## Deployment and limits + +Sharing requires SQLite or PostgreSQL persistence. The Gateway's startup +migration creates the share tables. Both the transcript and file bytes are +stored in the database, so backups include shared content and all Gateway workers +read the same frozen bytes. No separate public object store is required. + +Links expire after 7 days by default; the supported range is 1 hour to 30 days. +A conversation can have 10 active links. Each snapshot allows 10,000 visible +history rows, 10 MiB of transcript JSON, and 64 artifacts, with limits of 20 MiB +per file and 100 MiB total. Exceeding a limit rejects the whole snapshot. Revoking +a link deletes its stored transcript and file bytes. You can revoke an expired +link from the same dialog to remove its stored copy. + +Treat share URLs as access credentials, including in proxy access logs. The +server stores only a hash of the random token; a token is returned once on +creation. Public responses disable caching, referrers and search indexing. + +## API contract + +All paths below are relative to `/api`. + +| Method and path | Access and behavior | +| --- | --- | +| `POST /threads/{id}/shares` | Owner with `threads:write` and `runs:read`; creates the snapshot. Body: optional `expires_in_seconds` and `checkpoint_id`. Returns `201`, token and snapshot coordinates. | +| `GET /threads/{id}/shares` | Owner with `threads:read`; returns up to 100 link summaries, with active links first and expired links awaiting cleanup before revoked links. Tokens are never listed. | +| `DELETE /threads/{id}/shares/{share_id}` | Owner with `threads:write`; revokes access and removes stored snapshot bytes. | +| `GET /shares/{token}` | Anonymous capability read; returns visible messages, artifact manifest, checkpoint and sequence. | +| `GET`, `HEAD /shares/{token}/artifacts/{path}` | Reads only a copied manifest entry. Supports byte ranges and `?download=true`. | +| `POST /shares/{token}/fork` | Signed-in visitor with `threads:write`; body contains a UUID `request_id`. Returns the new `thread_id`. Retrying the same request reuses a completed import. | + +Creation captures the latest idle root checkpoint and its persisted visible +history under exclusive thread admission. `checkpoint_id`, when supplied, is +an expected-head guard: a stale value returns `409`. Older checkpoints cannot +be selected because the source filesystem does not retain file versions for +historical checkpoints. The response binds the snapshot to the captured +checkpoint and highest visible history sequence. + +Expired, revoked and unknown tokens all return `404`. A running conversation or +stale checkpoint returns `409`; size limits return `413`; SQL-disabled servers +return `503`. Share tokens only authorize the two public read routes. They never +authorize thread mutation, execution, or access to the original file tree. diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index a94c252293..3440811ab0 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -46,6 +46,10 @@ through `scripts/pnpm.py`. ## Conventions the tools do not enforce +- **Public shares stay outside the workspace auth layout.** Read snapshots + without session credentials, rewrite file links through their manifest, and + authenticate only when importing into a workspace. See + [sharing](../docs/guide/sharing.md). - **`src/components/ui/` and `src/components/ai-elements/` are generated** from component registries. Change them by regenerating, because a manual edit is lost on the next update. diff --git a/frontend/src/app/share/[share_token]/page.tsx b/frontend/src/app/share/[share_token]/page.tsx new file mode 100644 index 0000000000..041dcbafc8 --- /dev/null +++ b/frontend/src/app/share/[share_token]/page.tsx @@ -0,0 +1,10 @@ +import { SharePage } from "@/components/shares/share-page"; + +export default async function Page({ + params, +}: { + params: Promise<{ share_token: string }>; +}) { + const { share_token } = await params; + return ; +} diff --git a/frontend/src/app/share/layout.tsx b/frontend/src/app/share/layout.tsx new file mode 100644 index 0000000000..0041996082 --- /dev/null +++ b/frontend/src/app/share/layout.tsx @@ -0,0 +1,25 @@ +import "katex/dist/katex.min.css"; +import "streamdown/styles.css"; + +import type { Metadata } from "next"; + +import { I18nProvider } from "@/core/i18n/context"; +import { detectLocaleServer } from "@/core/i18n/server"; + +export const metadata: Metadata = { + title: "Shared conversation · DeerFlow", + robots: { index: false, follow: false }, + referrer: "no-referrer", +}; + +export default async function ShareLayout({ + children, +}: { + children: React.ReactNode; +}) { + return ( + + {children} + + ); +} diff --git a/frontend/src/components/shares/share-artifact.tsx b/frontend/src/components/shares/share-artifact.tsx new file mode 100644 index 0000000000..829c4d2da0 --- /dev/null +++ b/frontend/src/components/shares/share-artifact.tsx @@ -0,0 +1,115 @@ +"use client"; + +import { useEffect, useState } from "react"; + +import { ArtifactTablePreview } from "@/components/workspace/artifacts/artifact-table-preview"; +import { useI18n } from "@/core/i18n/hooks"; +import { shareArtifactURL, type SharedArtifact } from "@/core/shares/api"; + +import { ShareMarkdown } from "./share-markdown"; + +export function ShareArtifact({ + token, + artifact, + artifacts, +}: { + token: string; + artifact: SharedArtifact; + artifacts: SharedArtifact[]; +}) { + const { t } = useI18n(); + const url = shareArtifactURL(token, artifact.path); + const extension = artifact.path.split(".").at(-1)?.toLowerCase(); + const isHTML = extension === "html" || extension === "htm"; + const isImage = ["png", "jpg", "jpeg", "gif", "webp", "svg"].includes( + extension ?? "", + ); + const isText = + !isHTML && + (artifact.mime_type.startsWith("text/") || + ["json", "md", "csv", "tsv"].includes(extension ?? "")); + const [text, setText] = useState(null); + const [error, setError] = useState(false); + useEffect(() => { + if (!isText) return; + const controller = new AbortController(); + void globalThis + .fetch(url, { + credentials: "omit", + cache: "no-store", + referrerPolicy: "no-referrer", + headers: artifact.size > 0 ? { Range: "bytes=0-524287" } : undefined, + signal: controller.signal, + }) + .then(async (response) => { + if (!response.ok) throw new Error(); + return response.text(); + }) + .then((value) => { + if (!controller.signal.aborted) setText(value); + }) + .catch(() => { + if (!controller.signal.aborted) setError(true); + }); + return () => controller.abort(); + }, [url, isText, artifact.size]); + + if (isHTML || extension === "pdf") + return ( +
+

{t.sharing.previewNote}

+