fix(backend): reject path traversal in upload folder

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/public.py

@@ -7,6 +7,7 @@ from sqlalchemy.orm import Session
 from app.core import storage
 from app.core.database import get_db
 from app.core.response import AppHTTPError, success_response
+from app.core.storage import StorageError
 from app.crud import attachment as attachment_crud
 
 router = APIRouter(prefix="/public", tags=["public"])
@@ -21,7 +22,10 @@ async def upload_attachment(
     if not file or not file.filename:
         raise AppHTTPError(400, "请选择文件")
     content = await file.read()
-    stored = storage.upload_file(content, file.filename, folder or "")
+    try:
+        stored = storage.upload_file(content, file.filename, folder or "")
+    except StorageError as exc:
+        raise AppHTTPError(400, exc.message) from exc
     row = attachment_crud.create_attachment(
         db,
         file_name=file.filename,

backend/app/core/storage.py

@@ -11,26 +11,62 @@ from typing import Any, Dict
 from app.core.config import get_settings
 
 
+class StorageError(Exception):
+    """存储参数/写入错误（如路径穿越）。"""
+
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(message)
+
+
 def _unique_name(filename: str) -> str:
     ext = os.path.splitext(filename)[1] or ""
     return f"{uuid.uuid4().hex}{ext}"
 
 
+def _normalize_folder(folder: str) -> str:
+    """规范化 folder：拒绝空段以外的 '..' / 绝对路径痕迹。"""
+    raw = (folder or "").strip().replace("\\", "/")
+    if not raw:
+        return ""
+    # 拒绝绝对路径或盘符式前缀
+    if raw.startswith("/") or (len(raw) >= 2 and raw[1] == ":"):
+        raise StorageError("非法的文件夹路径")
+    parts: list[str] = []
+    for part in raw.split("/"):
+        if part in ("", "."):
+            continue
+        if part == "..":
+            raise StorageError("非法的文件夹路径")
+        parts.append(part)
+    return "/".join(parts)
+
+
 def _object_key(folder: str, filename: str) -> str:
     unique = _unique_name(filename)
     date_part = datetime.now(timezone.utc).strftime("%Y%m/%d")
-    folder = (folder or "").strip().strip("/")
+    folder = _normalize_folder(folder)
     if folder:
         return f"{folder}/{date_part}/{unique}"
     return f"{date_part}/{unique}"
 
 
+def _ensure_within_base(base_dir: Path, dest: Path) -> None:
+    base_resolved = base_dir.resolve()
+    dest_resolved = dest.resolve()
+    try:
+        dest_resolved.relative_to(base_resolved)
+    except ValueError as exc:
+        raise StorageError("非法的文件夹路径") from exc
+
+
 class LocalStorage:
     def upload_file(self, content: bytes, filename: str, folder: str = "") -> Dict[str, str]:
         settings = get_settings()
         base_dir = Path(settings.LOCAL_STORAGE_DIR)
         object_key = _object_key(folder, filename)
         dest = base_dir / object_key
+        _ensure_within_base(base_dir, dest)
         dest.parent.mkdir(parents=True, exist_ok=True)
         dest.write_bytes(content)
         url = f"/uploads/{object_key}"

backend/tests/test_upload.py

@@ -1,12 +1,19 @@
 """Public attachment upload (local storage)."""
+import pytest
+
 from app.core.config import get_settings
 
 
-def test_public_upload(client, tmp_path, monkeypatch):
+@pytest.fixture
+def local_storage_env(tmp_path, monkeypatch):
     monkeypatch.setenv("STORAGE_TYPE", "local")
     monkeypatch.setenv("LOCAL_STORAGE_DIR", str(tmp_path))
     get_settings.cache_clear()
+    yield tmp_path
+    get_settings.cache_clear()
+
 
+def test_public_upload(client, local_storage_env):
     files = {"file": ("a.png", b"fake-png-bytes", "image/png")}
     r = client.post("/api/public/attachment/upload", files=files)
     assert r.status_code == 200
@@ -16,3 +23,23 @@ def test_public_upload(client, tmp_path, monkeypatch):
     assert data["fileName"] == "a.png"
     assert data["size"] == len(b"fake-png-bytes")
     assert data["url"].startswith("/uploads/")
+
+
+def test_public_upload_rejects_path_traversal(client, local_storage_env):
+    base = local_storage_env
+    outside = (base / ".." / "outside").resolve()
+    assert not outside.exists()
+
+    files = {"file": ("evil.png", b"evil-bytes", "image/png")}
+    r = client.post(
+        "/api/public/attachment/upload",
+        files=files,
+        data={"folder": "../outside"},
+    )
+    assert r.status_code == 400
+    body = r.json()
+    assert body["code"] == 400
+    assert "message" in body
+    assert "detail" not in body
+    assert not outside.exists()
+    assert not any(base.rglob("*"))