From 6269f2868068b63e3583fbe093bd75bcf346e6dc Mon Sep 17 00:00:00 2001 From: lux-liang Date: Tue, 21 Jul 2026 11:52:21 +0800 Subject: [PATCH] fix: prevent task file path traversal --- backend/app/routers/files_router.py | 45 +++++++++++--- backend/app/routers/modeling_router.py | 42 ++++++++----- backend/app/tests/test_path_utils.py | 83 ++++++++++++++++++++++++++ backend/app/utils/common_utils.py | 27 ++------- backend/app/utils/path_utils.py | 59 ++++++++++++++++++ 5 files changed, 212 insertions(+), 44 deletions(-) create mode 100644 backend/app/tests/test_path_utils.py create mode 100644 backend/app/utils/path_utils.py diff --git a/backend/app/routers/files_router.py b/backend/app/routers/files_router.py index b7d7e76c..06d049a4 100644 --- a/backend/app/routers/files_router.py +++ b/backend/app/routers/files_router.py @@ -1,28 +1,59 @@ """文件管理路由模块，提供文件下载、列表和目录打开等接口。""" -from fastapi import APIRouter -from app.utils.common_utils import get_current_files, get_work_dir import os import subprocess +from urllib.parse import quote + +from fastapi import APIRouter, HTTPException + +from app.utils.common_utils import get_current_files, get_work_dir +from app.utils.path_utils import ( + ensure_safe_task_id, + resolve_path_within, +) from icecream import ic # type: ignore[import-unresolved] -from fastapi import HTTPException router = APIRouter() +def _require_work_dir(task_id: str) -> tuple[str, str]: + """Return a validated task ID and its existing work directory.""" + try: + safe_task_id = ensure_safe_task_id(task_id) + return safe_task_id, get_work_dir(safe_task_id) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="任务不存在") from exc + + @router.get("/download_url") async def get_download_url(task_id: str, filename: str): - return {"download_url": f"http://localhost:8000/static/{task_id}/{filename}"} + safe_task_id, work_dir = _require_work_dir(task_id) + try: + safe_filename = resolve_path_within(work_dir, filename).name + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + return { + "download_url": ( + "http://localhost:8000/static/" + f"{quote(safe_task_id, safe='')}/{quote(safe_filename, safe='')}" + ) + } @router.get("/download_all_url") async def get_download_all_url(task_id: str): - return {"download_url": f"http://localhost:8000/static/{task_id}/all.zip"} + safe_task_id, _ = _require_work_dir(task_id) + return { + "download_url": f"http://localhost:8000/static/{quote(safe_task_id, safe='')}/all.zip" + } @router.get("/files") async def get_files(task_id: str): - work_dir = get_work_dir(task_id) + _, work_dir = _require_work_dir(task_id) files = get_current_files(work_dir, "all") file_all = [] @@ -37,7 +68,7 @@ async def get_files(task_id: str): async def open_folder(task_id: str): ic(task_id) # 打开工作目录 - work_dir = get_work_dir(task_id) + _, work_dir = _require_work_dir(task_id) # 打开工作目录 if os.name == "nt": diff --git a/backend/app/routers/modeling_router.py b/backend/app/routers/modeling_router.py index fdedd510..c48534b5 100644 --- a/backend/app/routers/modeling_router.py +++ b/backend/app/routers/modeling_router.py @@ -13,6 +13,7 @@ get_current_files, md_2_docx, ) +from app.utils.path_utils import resolve_path_within import os import asyncio from typing import Dict, Tuple @@ -193,15 +194,26 @@ async def exampleModeling( ): task_id = create_task_id() work_dir = create_work_dir(task_id) - example_dir = os.path.join("app", "example", "example", example_request.source) + try: + example_dir = resolve_path_within( + os.path.join("app", "example", "example"), + example_request.source, + "示例来源", + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + if not example_dir.is_dir(): + raise HTTPException(status_code=404, detail="示例不存在") + ic(example_dir) - with open(os.path.join(example_dir, "questions.txt"), "r", encoding="utf-8") as f: + with open(example_dir / "questions.txt", "r", encoding="utf-8") as f: ques_all = f.read() - current_files = get_current_files(example_dir, "data") + current_files = get_current_files(str(example_dir), "data") for file in current_files: - src_file = os.path.join(example_dir, file) - dst_file = os.path.join(work_dir, file) + src_file = resolve_path_within(example_dir, file) + dst_file = resolve_path_within(work_dir, file) with open(src_file, "rb") as src, open(dst_file, "wb") as dst: dst.write(src.read()) # 存储任务ID @@ -234,19 +246,19 @@ async def modeling( if files: logger.info(f"开始处理上传的文件，工作目录: {work_dir}") for file in files: + filename = file.filename or "" try: - assert file.filename is not None - data_file_path = os.path.join(work_dir, file.filename) - logger.info(f"保存文件: {file.filename} -> {data_file_path}") + data_file_path = resolve_path_within(work_dir, filename) + except ValueError as exc: + logger.warning(f"拒绝非法上传文件名: {filename!r}") + raise HTTPException(status_code=400, detail=str(exc)) from exc - # 确保文件名不为空 - if not file.filename: - logger.warning("跳过空文件名") - continue + try: + logger.info(f"保存文件: {filename} -> {data_file_path}") content = await file.read() if not content: - logger.warning(f"文件 {file.filename} 内容为空") + logger.warning(f"文件 {filename} 内容为空") continue with open(data_file_path, "wb") as f: @@ -254,9 +266,9 @@ async def modeling( logger.info(f"成功保存文件: {data_file_path}") except Exception as e: - logger.error(f"保存文件 {file.filename} 失败: {str(e)}") + logger.error(f"保存文件 {filename} 失败: {str(e)}") raise HTTPException( - status_code=500, detail=f"保存文件 {file.filename} 失败: {str(e)}" + status_code=500, detail=f"保存文件 {filename} 失败: {str(e)}" ) else: logger.warning("没有上传文件") diff --git a/backend/app/tests/test_path_utils.py b/backend/app/tests/test_path_utils.py new file mode 100644 index 00000000..2f646f4d --- /dev/null +++ b/backend/app/tests/test_path_utils.py @@ -0,0 +1,83 @@ +"""Regression tests for task and file path validation.""" + +import tempfile +import unittest +from pathlib import Path + +from app.utils.path_utils import ( + ensure_safe_path_component, + ensure_safe_task_id, + resolve_path_within, +) + + +class TestPathUtils(unittest.TestCase): + """Verify that user-controlled path components cannot escape their base.""" + + def test_safe_task_id_is_preserved(self): + self.assertEqual( + ensure_safe_task_id("20260721-120000-deadbeef"), + "20260721-120000-deadbeef", + ) + + def test_task_id_path_traversal_is_rejected(self): + invalid_task_ids = [ + "../task", + "../../..", + "/tmp/task", + r"..\task", + "", + ] + + for task_id in invalid_task_ids: + with self.subTest(task_id=task_id): + with self.assertRaises(ValueError): + ensure_safe_task_id(task_id) + + def test_cross_platform_unsafe_components_are_rejected(self): + invalid_components = [ + "../outside.csv", + "folder/data.csv", + r"..\outside.csv", + r"C:\temp\data.csv", + "C:data.csv", + "/tmp/data.csv", + ".", + "..", + "data\n.csv", + "", + ] + + for component in invalid_components: + with self.subTest(component=component): + with self.assertRaises(ValueError): + ensure_safe_path_component(component, "文件名") + + def test_valid_filename_resolves_inside_base_directory(self): + with tempfile.TemporaryDirectory() as temp_dir: + base_dir = Path(temp_dir).resolve() + resolved_path = resolve_path_within(base_dir, "数据 01.csv") + + self.assertEqual(resolved_path, base_dir / "数据 01.csv") + self.assertEqual(resolved_path.parent, base_dir) + + def test_symlink_escape_is_rejected(self): + with ( + tempfile.TemporaryDirectory() as temp_dir, + tempfile.TemporaryDirectory() as outside_dir, + ): + outside_file = Path(outside_dir) / "outside.csv" + outside_file.write_text("sensitive", encoding="utf-8") + symlink_path = Path(temp_dir) / "data.csv" + + try: + symlink_path.symlink_to(outside_file) + except OSError as exc: + self.skipTest(f"symlink creation is unavailable: {exc}") + + with self.assertRaises(ValueError): + resolve_path_within(temp_dir, "data.csv") + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/app/utils/common_utils.py b/backend/app/utils/common_utils.py index aabad39c..0465f3d0 100644 --- a/backend/app/utils/common_utils.py +++ b/backend/app/utils/common_utils.py @@ -10,8 +10,7 @@ import re import pypandoc # type: ignore[import-unresolved] from app.config.setting import settings - -TASK_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") +from app.utils.path_utils import ensure_safe_task_id def create_task_id() -> str: @@ -22,24 +21,6 @@ def create_task_id() -> str: return f"{timestamp}-{random_hash}" -def ensure_safe_task_id(task_id: str) -> str: - """验证任务 ID 的合法性，防止路径遍历攻击。 - - Args: - task_id: 待验证的任务 ID。 - - Returns: - 验证通过的任务 ID。 - - Raises: - ValueError: 任务 ID 不合法时抛出。 - """ - normalized = (task_id or "").strip() - if not normalized or not TASK_ID_PATTERN.fullmatch(normalized): - raise ValueError("非法 task_id") - return normalized - - def create_work_dir(task_id: str) -> str: """为指定任务创建工作目录，并复制字体文件到工作目录。 @@ -50,7 +31,8 @@ def create_work_dir(task_id: str) -> str: 工作目录路径。 """ # 设置主工作目录和子目录 - work_dir = os.path.join("project", "work_dir", task_id) + safe_task_id = ensure_safe_task_id(task_id) + work_dir = os.path.join("project", "work_dir", safe_task_id) try: # 创建目录，如果目录已存在也不会报错 @@ -103,7 +85,8 @@ def get_work_dir(task_id: str) -> str: Raises: FileNotFoundError: 工作目录不存在时抛出。 """ - work_dir = os.path.join("project", "work_dir", task_id) + safe_task_id = ensure_safe_task_id(task_id) + work_dir = os.path.join("project", "work_dir", safe_task_id) if os.path.exists(work_dir): return work_dir else: diff --git a/backend/app/utils/path_utils.py b/backend/app/utils/path_utils.py new file mode 100644 index 00000000..ed5dd36c --- /dev/null +++ b/backend/app/utils/path_utils.py @@ -0,0 +1,59 @@ +"""Path validation helpers used by task and file operations.""" + +import ntpath +import os +import re +from pathlib import Path + + +TASK_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") + + +def ensure_safe_task_id(task_id: str) -> str: + """Validate a task ID before using it as a directory name.""" + normalized = (task_id or "").strip() + if not normalized or not TASK_ID_PATTERN.fullmatch(normalized): + raise ValueError("非法 task_id") + return normalized + + +def ensure_safe_path_component(value: str, field_name: str = "路径组件") -> str: + """Validate a single cross-platform path component. + + Absolute paths, drive-qualified paths, separators, traversal markers, and + control characters are rejected instead of silently normalizing them. + """ + normalized = (value or "").strip() + drive, _ = ntpath.splitdrive(normalized) + has_control_character = any(ord(character) < 32 for character in normalized) + + if ( + not normalized + or normalized in {".", ".."} + or drive + or os.path.isabs(normalized) + or "/" in normalized + or "\\" in normalized + or has_control_character + ): + raise ValueError(f"非法{field_name}") + + return normalized + + +def resolve_path_within( + base_dir: str | os.PathLike[str], + component: str, + field_name: str = "文件名", +) -> Path: + """Resolve one path component and guarantee it stays inside ``base_dir``.""" + safe_component = ensure_safe_path_component(component, field_name) + resolved_base = Path(base_dir).resolve() + resolved_path = (resolved_base / safe_component).resolve() + + try: + resolved_path.relative_to(resolved_base) + except ValueError as exc: + raise ValueError(f"非法{field_name}") from exc + + return resolved_path