"""沙箱文件预览 API。 通过此端点,前端可以读取沙箱中由工具写入的文件内容,实现源码/渲染双模式预览。 使用场景: - write_file 写入 HTML 文件 → 前端 iframe 渲染预览 - write_file 写入 SVG → 前端 img 展示 - shell 生成图片/文档 → 前端按 MIME 类型渲染 - 任何工具的 stdout 生成的文件 安全约束: - 用户必须经过 JWT 鉴权 - 路径由沙箱端校验(防逃逸) - 可选 run_id/character_id 绑定授权上下文 """ from __future__ import annotations import base64 import mimetypes from fastapi import APIRouter, Depends, HTTPException, Query from starlette.responses import Response from app.api.v1.deps import get_current_user, get_tool_runtime from app.application.tool_runtime.runtime import ToolRuntime from app.integrations.sandbox.exceptions import SandboxError, SandboxNotFound from app.schemas.auth import CurrentUser router = APIRouter(prefix="/sandbox", tags=["sandbox"]) @router.get("/file") async def read_sandbox_file( path: str = Query( ..., description="文件路径。isolated=True 时视为工作区相对路径" "(如 write_file 传入的 ``path`` 参数);" "isolated=False 时作为沙箱 base 下的完整相对路径。", ), run_id: str | None = Query( None, description="关联的 run_id(可选,写入授权上下文)。仅用于审计链路,不影响文件可读性。", ), character_id: str | None = Query( None, description="人物 ID。提供时自动构造隔离工作区路径" "(``workspace//``)," "与 write_file/shell 等工具的 isolated_workspace 路径一致。", ), isolated: bool = Query( True, description="是否自动按 isolated_workspace 规则补全路径前缀。" "write_file 写入的文件在 workspace// 下," "传递 write_file 的 path 时保持 True 即可。", ), runtime: ToolRuntime = Depends(get_tool_runtime), _current_user: CurrentUser = Depends(get_current_user), ) -> Response: """读取沙箱内文件,以原始字节返回(MIME 类型由路径扩展名推导)。 前端根据 MIME 类型自行决定展示方式: - ``text/html`` → iframe sandbox 渲染预览 - ``image/*`` → img 标签展示 - ``text/*`` → 源码高亮 - ``application/pdf`` → PDF 内嵌展示 - 其他 → 下载 路径安全:沙箱端会对路径做 canonicalize + starts_with(base) 校验, 防止目录穿越到沙箱 base 之外。 """ # 1) 推导 MIME 类型 mime_type, _ = mimetypes.guess_type(path) if mime_type is None: mime_type = "application/octet-stream" # 2) 从沙箱读取 try: result = await runtime.read_sandbox_file( path=path, run_id=run_id, character_id=character_id, isolated=isolated, ) except SandboxNotFound: raise HTTPException( status_code=404, detail={"error": {"code": "file_not_found", "message": f"文件不存在: {path}"}}, ) from None except SandboxError as exc: raise HTTPException( status_code=502, detail={ "error": { "code": "sandbox_error", "message": f"沙箱文件读取失败: {exc}", } }, ) from None # 3) base64url-no-pad → raw bytes pad_len = -len(result.content_b64) % 4 content = base64.urlsafe_b64decode(result.content_b64 + "=" * pad_len) return Response(content=content, media_type=mime_type) @router.get("/files") async def list_sandbox_files( path: str = Query( ".", description="目录路径。默认工作区根。", ), run_id: str | None = Query(None, description="关联的 run_id(可选)"), character_id: str | None = Query( None, description="人物 ID。提供时自动应用 isolated_workspace 隔离前缀。", ), isolated: bool = Query(True, description="是否自动补全隔离前缀"), runtime: ToolRuntime = Depends(get_tool_runtime), _current_user: CurrentUser = Depends(get_current_user), ) -> list[dict[str, object]]: """列出沙箱工作区目录内容。 返回有序列表,每个条目含 ``name`` / ``kind``(``"file"`` | ``"dir"``)/ ``size``。 适用于前端文件浏览器导航。 """ try: entries = await runtime.list_sandbox_dir( path=path, run_id=run_id, character_id=character_id, isolated=isolated, ) except SandboxNotFound: raise HTTPException( status_code=404, detail={"error": {"code": "dir_not_found", "message": f"目录不存在: {path}"}}, ) from None except SandboxError as exc: raise HTTPException( status_code=502, detail={ "error": { "code": "sandbox_error", "message": f"沙箱目录读取失败: {exc}", } }, ) from None return [{"name": e.name, "kind": e.kind, "size": e.size} for e in entries]