""" 文件系统相关路由 """ import os import shutil import subprocess from typing import List from fastapi import APIRouter, HTTPException, Query from app.core.config import settings from app.schemas.fs import ( FileContentResponse, FileContentUpdateRequest, FileContentUpdateResponse, FileCreateRequest, FileOperationResponse, FileTreeItem, ) from app.schemas.project import ProjectCreateRequest, ProjectCreateResponse router = APIRouter() def _is_git_ignored(directory: str, absolute_path: str) -> bool: """ 判断指定路径是否被 git 忽略。 依赖目标目录是否为 git 仓库： - 若是 git 仓库，使用 `git check-ignore` 判定； - 若不是 git 仓库或 git 不可用，默认返回 False。 """ try: result = subprocess.run( ["git", "-C", directory, "check-ignore", absolute_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=5, ) # git check-ignore 退出码：0 表示被忽略，1 表示未被忽略 return result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired): # git 未安装或超时，按非忽略处理 return False def _assert_path_allowed(real_directory: str, real_target: str) -> None: """ 确保解析后的真实路径在允许范围内。 允许范围： - 项目目录 real_directory 内（常规文件访问）； - BASE_WORKSPACE 工作空间内（允许 tables_meta 这类指向工作空间根的共享软链接）。 这样既阻止通过 `..` 或外部软链接逃逸出工作空间，又允许项目内合法的共享软链接。 """ real_workspace = os.path.realpath(settings.BASE_WORKSPACE) within_dir = real_target == real_directory or real_target.startswith(real_directory + os.sep) within_workspace = real_target == real_workspace or real_target.startswith(real_workspace + os.sep) if not (within_dir or within_workspace): raise HTTPException( status_code=400, detail="非法的查询路径，检测到路径逃逸尝试" ) @router.get("/file", response_model=List[FileTreeItem]) async def list_file_tree( directory: str = Query(..., description="项目顶层绝对目录（URL 编码传输）"), path: str = Query(..., description="相对于 directory 的查询路径，. 表示根目录"), ): """ 查询指定目录下的文件树列表 - directory：项目顶层绝对目录，固定值（如 /root/workspace/test） - path：相对查询路径 - 传 `.`：读取 directory 根目录下所有文件 / 文件夹 - 传 `文件夹名/`（如 `ningxia/`）：读取 directory/文件夹名 子目录内文件树 """ # 规范化 directory：去除尾部斜杠 directory = os.path.normpath(directory) # 安全校验：directory 必须存在且为目录 if not os.path.isdir(directory): raise HTTPException( status_code=400, detail=f"目录不存在或不是文件夹：{directory}" ) # 规范化 path：仅允许 `.` 或以 `/` 结尾的相对子路径 # 处理 path 为 `.` 的根目录情形 if path == "." or path == "": target_dir = directory else: # 去除首尾空白，确保以 / 结尾（约定：文件夹末尾带 /） rel_path = path.strip() if not rel_path.endswith("/"): rel_path = rel_path + "/" # 去除结尾的 / 用于拼接 rel_path_clean = rel_path.rstrip("/") # 禁止路径逃逸 if ".." in rel_path_clean.split("/"): raise HTTPException( status_code=400, detail="非法的相对路径，禁止路径逃逸 (..)" ) # 禁止绝对路径 if os.path.isabs(rel_path_clean): raise HTTPException( status_code=400, detail="相对路径 path 不能为绝对路径" ) target_dir = os.path.join(directory, rel_path_clean) # 解析真实物理路径，确保最终路径在允许范围内（防逃逸，允许工作空间内软链接） real_directory = os.path.realpath(directory) real_target = os.path.realpath(target_dir) _assert_path_allowed(real_directory, real_target) if not os.path.isdir(real_target): raise HTTPException( status_code=404, detail=f"查询的子目录不存在：{path}" ) # 计算 path 前缀（基于 directory 的相对路径，用于拼接每个条目的 path） # path 为 `.` 时前缀为空；否则前缀为 `文件夹名/` if path == "." or path == "": rel_prefix = "" else: rel_prefix = path.strip() if not rel_prefix.endswith("/"): rel_prefix = rel_prefix + "/" items: List[FileTreeItem] = [] try: entries = sorted(os.listdir(real_target)) except OSError as e: raise HTTPException( status_code=500, detail=f"读取目录失败：{str(e)}" ) for entry in entries: abs_path = os.path.join(real_target, entry) is_dir = os.path.isdir(abs_path) # 基于 directory 的相对路径：文件夹末尾带 /，文件不带 rel_path = rel_prefix + entry + ("/" if is_dir else "") items.append( FileTreeItem( name=entry, path=rel_path, absolute=abs_path, type="directory" if is_dir else "file", ignored=_is_git_ignored(directory, abs_path), ) ) return items def _resolve_file_path(directory: str, path: str): """ 解析并校验「相对文件路径」对应的真实物理路径。 校验规则： - directory 必须存在且为目录 - path 必须指向具体文件，不能为根目录（. / 空） - 禁止路径逃逸（..）与绝对路径 - realpath 解析后必须仍位于 directory 内（防符号链接逃逸） 返回：(real_directory, real_target, rel_path) """ # 规范化 directory directory = os.path.normpath(directory) if not os.path.isdir(directory): raise HTTPException( status_code=400, detail=f"目录不存在或不是文件夹：{directory}" ) rel_path = path.strip().rstrip("/") # 仅允许相对路径，禁止路径逃逸 if rel_path in ("", "."): raise HTTPException( status_code=400, detail="path 必须指向具体文件，不能为根目录" ) if ".." in rel_path.split("/"): raise HTTPException( status_code=400, detail="非法的相对路径，禁止路径逃逸 (..)" ) if os.path.isabs(rel_path): raise HTTPException( status_code=400, detail="相对路径 path 不能为绝对路径" ) target = os.path.join(directory, rel_path) # 解析真实物理路径，确保最终路径在允许范围内（防逃逸，允许工作空间内软链接） real_directory = os.path.realpath(directory) real_target = os.path.realpath(target) _assert_path_allowed(real_directory, real_target) return real_directory, real_target, rel_path @router.get("/file/content", response_model=FileContentResponse) async def get_file_content( directory: str = Query(..., description="项目顶层绝对目录（URL 编码传输）"), path: str = Query(..., description="相对于 directory 的文件路径，如 appfile/main.py"), ): """ 预览文件内容 - directory：项目顶层绝对目录 - path：相对于 directory 的文件路径（不带尾部 /，指向具体文件） - 返回文本文件内容；二进制文件返回 type=binary 且 content 为空 """ _real_directory, real_target, _rel_path = _resolve_file_path(directory, path) if not os.path.exists(real_target): raise HTTPException( status_code=404, detail=f"文件不存在：{path}" ) if os.path.isdir(real_target): raise HTTPException( status_code=400, detail=f"目标路径是文件夹，不是文件：{path}" ) # 限制预览文件大小，避免读取超大文件 max_size = 5 * 1024 * 1024 # 5MB try: file_size = os.path.getsize(real_target) if file_size > max_size: raise HTTPException( status_code=413, detail=f"文件过大，无法预览（{file_size} 字节，上限 {max_size} 字节）" ) except OSError as e: raise HTTPException( status_code=500, detail=f"读取文件信息失败：{str(e)}" ) # 先按二进制读取前若干字节判断是否为文本 try: with open(real_target, "rb") as f: chunk = f.read(1024) # 含 NUL 字节通常视为二进制 if b"\x00" in chunk: return FileContentResponse(type="binary", content="") # 尝试以 UTF-8 解码全文 with open(real_target, "r", encoding="utf-8") as f: content = f.read() return FileContentResponse(type="text", content=content) except UnicodeDecodeError: # 非 UTF-8 文本，按二进制处理 return FileContentResponse(type="binary", content="") except OSError as e: raise HTTPException( status_code=500, detail=f"读取文件失败：{str(e)}" ) @router.put("/file/content", response_model=FileContentUpdateResponse) async def update_file_content( request: FileContentUpdateRequest, directory: str = Query(..., description="项目顶层绝对目录（URL 编码传输）"), path: str = Query(..., description="相对于 directory 的文件路径，如 appfile/main.py"), ): """ 修改（覆盖写入）文件内容 - directory：项目顶层绝对目录 - path：相对于 directory 的文件路径（不带尾部 /，指向具体文件） - 请求体 content：文件新内容（UTF-8 文本，整文件覆盖写入） """ _real_directory, real_target, rel_path = _resolve_file_path(directory, path) if not os.path.exists(real_target): raise HTTPException( status_code=404, detail=f"文件不存在：{path}" ) if os.path.isdir(real_target): raise HTTPException( status_code=400, detail=f"目标路径是文件夹，不是文件：{path}" ) try: # 整文件覆盖写入（UTF-8） with open(real_target, "w", encoding="utf-8") as f: f.write(request.content) except OSError as e: raise HTTPException( status_code=500, detail=f"写入文件失败：{str(e)}" ) return FileContentUpdateResponse( success=True, message="文件已保存", path=rel_path, absolute=real_target, ) @router.post("/file", response_model=FileOperationResponse) async def create_file_entry( request: FileCreateRequest, directory: str = Query(..., description="项目顶层绝对目录（URL 编码传输）"), path: str = Query(..., description="相对于 directory 的新建项路径，如 appfile/new.py"), ): """ 新建文件 / 文件夹 - directory：项目顶层绝对目录 - path：相对于 directory 的新建项路径（不带尾部 /） - 请求体 type：file / directory；type=file 时可用 content 指定初始内容 - 目标已存在返回 409；新建文件时会自动创建所需父目录 """ _real_directory, real_target, rel_path = _resolve_file_path(directory, path) if os.path.exists(real_target): raise HTTPException( status_code=409, detail=f"目标已存在：{path}" ) item_type = request.type try: if item_type == "directory": os.makedirs(real_target, exist_ok=False) else: # 文件：确保父目录存在后写入初始内容 parent_dir = os.path.dirname(real_target) if parent_dir: os.makedirs(parent_dir, exist_ok=True) with open(real_target, "w", encoding="utf-8") as f: f.write(request.content or "") except OSError as e: raise HTTPException( status_code=500, detail=f"新建失败：{str(e)}" ) return FileOperationResponse( success=True, message=f"{'文件夹' if item_type == 'directory' else '文件'}已创建", path=rel_path, absolute=real_target, type=item_type, ) @router.delete("/file", response_model=FileOperationResponse) async def delete_file_entry( directory: str = Query(..., description="项目顶层绝对目录（URL 编码传输）"), path: str = Query(..., description="相对于 directory 的待删除项路径"), ): """ 删除文件 / 文件夹 - directory：项目顶层绝对目录 - path：相对于 directory 的待删除项路径（文件夹可带尾部 /） - 文件夹递归删除；不存在返回 404 - path 不能为根目录（.），不会删除 directory 本身 """ _real_directory, real_target, rel_path = _resolve_file_path(directory, path) if not os.path.exists(real_target): raise HTTPException( status_code=404, detail=f"目标不存在：{path}" ) is_dir = os.path.isdir(real_target) try: if is_dir: shutil.rmtree(real_target) else: os.remove(real_target) except OSError as e: raise HTTPException( status_code=500, detail=f"删除失败：{str(e)}" ) return FileOperationResponse( success=True, message=f"{'文件夹' if is_dir else '文件'}已删除", path=rel_path, absolute=real_target, type="directory" if is_dir else "file", ) @router.post("/project/create", response_model=ProjectCreateResponse) async def create_project(request: ProjectCreateRequest): """ 创建新项目 - 校验项目名格式 - 拼接完整路径 - 确保路径在安全目录内 - 创建项目目录 - 软链接 BASE_WORKSPACE/tables_meta 到新项目下（共享表元数据） """ project_name = request.project_name # 二次校验：禁止路径逃逸字符（防御性编程） if ".." in project_name or "/" in project_name or "\\" in project_name: raise HTTPException( status_code=400, detail="项目名称不能包含路径逃逸字符 (../ 或 / 或 \\)" ) # 拼接完整路径 project_path = os.path.join(settings.BASE_WORKSPACE, project_name) # 安全校验：确保最终路径在 base_workspace 内 real_workspace = os.path.realpath(settings.BASE_WORKSPACE) real_project_path = os.path.realpath(os.path.dirname(project_path)) # 检查路径是否逃逸 if not real_project_path.startswith(real_workspace): raise HTTPException( status_code=400, detail="非法的项目路径，检测到路径逃逸尝试" ) try: # 创建项目目录 os.makedirs(project_path, exist_ok=True) except OSError as e: raise HTTPException( status_code=500, detail=f"创建项目目录失败：{str(e)}" ) # 软链接 BASE_WORKSPACE/tables_meta 到新项目下（共享表元数据） # 使用相对路径，便于工作空间整体迁移；已存在则不覆盖 tables_meta_src = os.path.join(settings.BASE_WORKSPACE, "tables_meta") tables_meta_link = os.path.join(project_path, "tables_meta") symlink_note = "" try: if os.path.lexists(tables_meta_link): # 已存在（可能是历史软链接或实际文件），不覆盖 pass elif not os.path.exists(tables_meta_src): symlink_note = "（未找到 tables_meta 源，已跳过软链接）" else: rel_src = os.path.relpath(tables_meta_src, project_path) os.symlink(rel_src, tables_meta_link) except OSError as e: symlink_note = f"（软链接 tables_meta 失败：{str(e)}）" return ProjectCreateResponse( success=True, message=f"项目 '{project_name}' 创建成功{symlink_note}", project_path=project_path )