plugins/sandbox/api.py

"""
沙盒MCP插件 API路由

支持文件列表、读写、上传、下载和命令执行。
"""

from fastapi import APIRouter, HTTPException, UploadFile, File, Query
from fastapi.responses import FileResponse
from typing import Optional
from pathlib import Path

router = APIRouter()

# 在运行时获取插件实例
plugin = None


def set_plugin_instance(plugin_instance):
    """由系统调用，注入插件实例"""
    global plugin
    plugin = plugin_instance


@router.get("/")
async def api_root():
    """API根路径"""
    return {"message": "沙盒MCP插件API", "status": "运行中"}


@router.get("/status")
async def get_status():
    """获取插件状态"""
    if plugin is None:
        return {
            "name": "sandbox",
            "enabled": False,
            "sandbox_root": "",
            "file_count": 0,
            "total_size": 0,
            "tools": [],
            "message": "插件未加载",
        }

    return plugin.get_status()


@router.get("/files")
async def list_files(path: str = ""):
    """列出文件和目录

    参数:
    - path: 目录相对路径（空表示沙盒根目录）
    """
    if plugin is None or not plugin.enabled:
        raise HTTPException(status_code=400, detail="插件未启用")

    result = plugin.file_ops.list_files(path)
    return result


@router.get("/read")
async def read_file(path: str, encoding: str = "utf-8"):
    """读取文件内容

    参数:
    - path: 文件相对路径
    - encoding: 文件编码（默认utf-8）
    """
    if plugin is None or not plugin.enabled:
        raise HTTPException(status_code=400, detail="插件未启用")

    result = plugin.file_ops.read_file(path, encoding)
    return result


@router.post("/write")
async def write_file(path: str, content: str, mode: str = "write"):
    """写入文件内容

    参数:
    - path: 文件相对路径
    - content: 文件内容
    - mode: 写入模式（write 或 append）
    """
    if plugin is None or not plugin.enabled:
        raise HTTPException(status_code=400, detail="插件未启用")

    result = plugin.file_ops.write_file(path, content, mode)
    return result


@router.post("/upload")
async def upload_file(file: UploadFile = File(...), path: str = Query("")):
    """上传文件到沙盒

    参数:
    - file: 上传的文件
    - path: 目标目录相对路径（空表示沙盒根目录）

    返回:
    {
        "success": true/false,
        "path": "上传后的相对路径",
        "size": 文件大小（字节）,
        "error": "错误信息"
    }
    """
    if plugin is None or not plugin.enabled:
        raise HTTPException(status_code=400, detail="插件未启用")

    try:
        # 读取文件内容
        content = await file.read()
        file_size = len(content)

        # 验证文件大小
        size_validation = plugin.security.validate_file_size(file_size)
        if not size_validation["valid"]:
            return {
                "success": False,
                "error": size_validation["error"],
            }

        # 构建目标路径
        filename = file.filename or "uploaded_file"
        if path:
            target_path = f"{path}/{filename}"
        else:
            target_path = filename

        # 验证路径安全
        path_validation = plugin.security.validate_path(target_path)
        if not path_validation["safe"]:
            return {
                "success": False,
                "error": path_validation["error"],
            }

        # 写入文件
        target = Path(path_validation["path"])
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(content)

        return {
            "success": True,
            "path": target_path,
            "size": file_size,
        }

    except Exception as e:
        return {
            "success": False,
            "error": f"上传失败: {str(e)}",
        }


@router.get("/download")
async def download_file(path: str):
    """下载沙盒内的文件

    参数:
    - path: 文件相对路径

    返回:
    文件响应（如果成功）或错误 JSON
    """
    if plugin is None or not plugin.enabled:
        raise HTTPException(status_code=400, detail="插件未启用")

    try:
        # 验证路径安全
        path_validation = plugin.security.validate_path(path)
        if not path_validation["safe"]:
            raise HTTPException(status_code=400, detail=path_validation["error"])

        file_path = Path(path_validation["path"])

        # 检查文件是否存在
        if not file_path.exists():
            raise HTTPException(status_code=404, detail="文件不存在")

        if not file_path.is_file():
            raise HTTPException(status_code=400, detail="不能下载目录")

        # 返回文件响应
        return FileResponse(
            path=str(file_path),
            filename=file_path.name,
            media_type="application/octet-stream",
        )

    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"下载失败: {str(e)}")


@router.post("/execute")
async def execute_command(command: str, timeout: int = 30, cwd: str = ""):
    """执行命令

    参数:
    - command: 要执行的命令
    - timeout: 超时时间（秒）
    - cwd: 工作目录（相对路径）
    """
    if plugin is None or not plugin.enabled:
        raise HTTPException(status_code=400, detail="插件未启用")

    result = await plugin.command_exec.execute(command, timeout, cwd)
    return result


@router.post("/enable")
async def enable_plugin():
    """启用插件"""
    if plugin is None:
        raise HTTPException(status_code=400, detail="插件未加载")

    plugin.on_enable()
    return {"message": "插件已启用"}


@router.post("/disable")
async def disable_plugin():
    """禁用插件"""
    if plugin is None:
        raise HTTPException(status_code=400, detail="插件未加载")

    plugin.on_disable()
    return {"message": "插件已禁用"}