fix: prevent task file path traversal

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":

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,29 +246,29 @@ 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:
                     f.write(content)
                 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("没有上传文件")

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()

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:

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