fix: patch CVE dependencies and path traversal risks

Upgrade aiohttp and related packages, harden local file and storage path checks, tighten password/JWT comparisons, and default Docker forwarded IP trust to empty.

Author: vastsa

Dockerfile

@@ -57,16 +57,24 @@ RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \
 COPY --from=frontend-builder /build/fronted-2024/dist ./themes/2024
 COPY --from=frontend-builder /build/fronted-2023/dist ./themes/2023
 
-# 安装 Python 依赖
-RUN pip install --no-cache-dir -r requirements.txt
+# 安装系统安全更新 + Python 依赖
+# 清理 apt 缓存，降低镜像噪音与扫描面
+RUN apt-get update \
+ && apt-get upgrade -y --no-install-recommends \
+ && rm -rf /var/lib/apt/lists/* \
+ && pip install --no-cache-dir -r requirements.txt \
+ && pip cache purge || true
 
 # 环境变量配置
 ENV HOST="0.0.0.0" \
     PORT=12345 \
     WORKERS=1 \
-    LOG_LEVEL="info"
+    LOG_LEVEL="info" \
+    FORWARDED_ALLOW_IPS=""
 
 EXPOSE 12345
 
 # 生产环境启动命令
-CMD ["sh", "-c", "exec uvicorn main:app --host \"$HOST\" --port \"$PORT\" --workers \"$WORKERS\" --log-level \"$LOG_LEVEL\" --proxy-headers --forwarded-allow-ips '*'"]
+# FORWARDED_ALLOW_IPS 默认为空：仅信任直连 IP，避免任意客户端伪造 X-Forwarded-*。
+# 若前面有反向代理，请显式设置为代理网段，例如 "10.0.0.0/8,172.16.0.0/12"。
+CMD ["sh", "-c", "exec uvicorn main:app --host \"$HOST\" --port \"$PORT\" --workers \"$WORKERS\" --log-level \"$LOG_LEVEL\" --proxy-headers --forwarded-allow-ips \"${FORWARDED_ALLOW_IPS:-}\""]

apps/admin/dependencies.py

@@ -48,15 +48,20 @@ def create_token(data: dict, expires_in: int | None = None) -> str:
     token_lifetime = (
         get_admin_session_expire_seconds() if expires_in is None else expires_in
     )
-    header = base64.b64encode(
-        json.dumps({"alg": "HS256", "typ": "JWT"}).encode()
-    ).decode()
-    payload = base64.b64encode(
-        json.dumps({**data, "exp": int(time.time()) + token_lifetime}).encode()
-    ).decode()
-
-    signature = hmac.new(_get_jwt_secret(), f"{header}.{payload}".encode(), "sha256").digest()
-    signature = base64.b64encode(signature).decode()
+    header = base64.urlsafe_b64encode(
+        json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode()
+    ).decode().rstrip("=")
+    payload = base64.urlsafe_b64encode(
+        json.dumps(
+            {**data, "exp": int(time.time()) + token_lifetime},
+            separators=(",", ":"),
+        ).encode()
+    ).decode().rstrip("=")
+
+    signature = hmac.new(
+        _get_jwt_secret(), f"{header}.{payload}".encode(), "sha256"
+    ).digest()
+    signature = base64.urlsafe_b64encode(signature).decode().rstrip("=")
 
     return f"{header}.{payload}.{signature}"
 
@@ -76,13 +81,20 @@ def verify_token(token: str) -> dict:
             f"{header_b64}.{payload_b64}".encode(),
             "sha256",
         ).digest()
-        expected_signature_b64 = base64.b64encode(expected_signature).decode()
+        expected_signature_b64 = (
+            base64.urlsafe_b64encode(expected_signature).decode().rstrip("=")
+        )
 
         if not hmac.compare_digest(signature_b64, expected_signature_b64):
             raise ValueError("无效的签名")
 
-        # 解码payload
-        payload = json.loads(base64.b64decode(payload_b64))
+        # 解码payload（兼容历史标准 base64 与 urlsafe base64）
+        padded = payload_b64 + "=" * (-len(payload_b64) % 4)
+        try:
+            payload_bytes = base64.urlsafe_b64decode(padded)
+        except Exception:
+            payload_bytes = base64.b64decode(padded)
+        payload = json.loads(payload_bytes)
 
         # 检查是否过期
         if payload.get("exp", 0) < time.time():

apps/admin/services.py

@@ -1,4 +1,5 @@
 import hashlib
+from pathlib import Path
 import os
 import time
 import uuid
@@ -1627,9 +1628,35 @@ async def delete_file(self, filename: str):
 
 class LocalFileClass:
     def __init__(self, file):
-        self.file = file
-        self.path = data_root / "local" / file
-        if os.path.exists(self.path):
+        # 仅允许 data/local 目录下的单层文件名，阻断路径穿越与绝对路径访问。
+        raw_name = str(file or "")
+        normalized = Path(raw_name).as_posix()
+        # 输入本身不得包含路径分隔符或绝对路径形态。
+        if (
+            not raw_name
+            or raw_name in {".", ".."}
+            or normalized in {".", ".."}
+            or "/" in normalized
+            or normalized.startswith("~")
+            or Path(raw_name).is_absolute()
+            or Path(raw_name).name != raw_name
+        ):
+            raise HTTPException(status_code=400, detail="非法文件名")
+
+        safe_name = Path(raw_name).name
+        if not safe_name or safe_name in {".", ".."}:
+            raise HTTPException(status_code=400, detail="非法文件名")
+
+        local_root = (data_root / "local").resolve()
+        candidate = (local_root / safe_name).resolve()
+        try:
+            candidate.relative_to(local_root)
+        except ValueError:
+            raise HTTPException(status_code=400, detail="非法文件路径")
+
+        self.file = safe_name
+        self.path = candidate
+        if self.path.is_file():
             self.ctime = time.strftime(
                 "%Y-%m-%d %H:%M:%S", time.localtime(os.path.getctime(self.path))
             )
@@ -1649,4 +1676,4 @@ async def delete(self):
         os.remove(self.path)
 
     async def exists(self):
-        return os.path.exists(self.path)
+        return self.path.is_file()

core/storage.py

@@ -111,6 +111,17 @@ def __init__(self):
         self.chunk_size = 256 * 1024
         self.root_path = data_root
 
+    def _resolve_safe_path(self, relative_path: str) -> Path:
+        """将相对路径解析到数据根目录内，阻止路径穿越。"""
+        root = self.root_path.resolve()
+        raw = str(relative_path or "").replace("\\", "/").lstrip("/")
+        candidate = (root / raw).resolve()
+        try:
+            candidate.relative_to(root)
+        except ValueError as exc:
+            raise ValueError("非法文件路径") from exc
+        return candidate
+
     def _save(self, file, save_path):
         with open(save_path, "wb") as f:
             chunk = file.read(self.chunk_size)
@@ -119,27 +130,27 @@ def _save(self, file, save_path):
                 chunk = file.read(self.chunk_size)
 
     async def save_file(self, file: UploadFile, save_path: str):
-        path_obj = Path(save_path)
-        directory = str(path_obj.parent)
+        path_obj = Path(str(save_path).replace("\\", "/"))
+        directory = str(path_obj.parent).replace("\\", "/").lstrip("/")
         # 提取原始文件名并进行清理
         filename = await sanitize_filename(path_obj.name)
         # 构建安全的完整保存路径
-        safe_save_path = self.root_path / directory / filename
+        safe_save_path = self._resolve_safe_path(f"{directory}/{filename}" if directory not in {"", "."} else filename)
         # 确保目录存在
         if not safe_save_path.parent.exists():
             safe_save_path.parent.mkdir(parents=True)
         await asyncio.to_thread(self._save, file.file, safe_save_path)
 
     async def delete_file(self, file_code: FileCodes):
-        save_path = self.root_path / await file_code.get_file_path()
+        save_path = self._resolve_safe_path(await file_code.get_file_path())
         if save_path.exists():
             save_path.unlink()
 
     async def get_file_url(self, file_code: FileCodes):
         return await get_file_url(file_code.code)
 
     async def get_file_response(self, file_code: FileCodes):
-        file_path = self.root_path / await file_code.get_file_path()
+        file_path = self._resolve_safe_path(await file_code.get_file_path())
         if not file_path.exists():
             return APIResponse(code=404, detail="文件已过期删除")
         filename = f"{file_code.prefix}{file_code.suffix}"
@@ -171,8 +182,11 @@ async def save_chunk(self, upload_id: str, chunk_index: int, chunk_data: bytes,
         :param chunk_hash: 分片哈希值
         :param save_path: 文件保存路径
         """
-        chunk_dir = self.root_path / save_path
-        chunk_path = chunk_dir.parent / 'chunks' / upload_id / f"{chunk_index}.part"
+        # 先校验目标文件路径合法，再将分片落到同级 chunks 目录。
+        self._resolve_safe_path(save_path)
+        chunk_path = self._resolve_safe_path(
+            str(Path(save_path).parent / "chunks" / upload_id / f"{chunk_index}.part")
+        )
         if not chunk_path.parent.exists():
             chunk_path.parent.mkdir(parents=True, exist_ok=True)
         # 使用临时文件写入，确保原子性
@@ -195,9 +209,11 @@ async def merge_chunks(self, upload_id: str, chunk_info: UploadChunk, save_path:
         :param save_path: 文件保存路径
         :return: (文件路径, 文件哈希值)
         """
-        output_path = self.root_path / save_path
+        output_path = self._resolve_safe_path(save_path)
         output_path.parent.mkdir(parents=True, exist_ok=True)
-        chunk_base_dir = output_path.parent / 'chunks' / upload_id
+        chunk_base_dir = self._resolve_safe_path(
+            str(Path(save_path).parent / "chunks" / upload_id)
+        )
         file_sha256 = hashlib.sha256()
 
         # 使用临时文件写入，确保原子性
@@ -233,7 +249,9 @@ async def clean_chunks(self, upload_id: str, save_path: str):
         :param upload_id: 上传会话ID
         :param save_path: 文件保存路径
         """
-        chunk_dir = (self.root_path / save_path).parent / 'chunks' / upload_id
+        chunk_dir = self._resolve_safe_path(
+            str(Path(save_path).parent / "chunks" / upload_id)
+        )
         if chunk_dir.exists():
             try:
                 shutil.rmtree(chunk_dir)
@@ -253,7 +271,10 @@ async def file_exists(self, save_path: str) -> bool:
         :param save_path: 文件路径
         :return: 文件是否存在
         """
-        file_path = self.root_path / save_path
+        try:
+            file_path = self._resolve_safe_path(save_path)
+        except ValueError:
+            return False
         return file_path.exists()
 
 

core/utils.py

@@ -4,6 +4,7 @@
 # @Software: PyCharm
 import datetime
 import hashlib
+import hmac
 import os
 import re
 import secrets
@@ -124,10 +125,10 @@ def verify_password(password: str, hashed: str) -> bool:
             return False
         _, salt, stored_hash = parts
         password_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest()
-        return password_hash == stored_hash
+        return hmac.compare_digest(password_hash, stored_hash)
 
     # 旧格式: 明文比较 (兼容迁移前的数据)
-    return password == hashed
+    return hmac.compare_digest(str(password), str(hashed))
 
 
 def is_password_hashed(password: str) -> bool:

requirements.txt

@@ -1,8 +1,9 @@
 aioboto3==15.5.0
-aiohttp==3.13.3
+aiohttp==3.14.2
 aiofiles==25.1.0
-fastapi==0.128.0
+fastapi==0.139.2
+starlette==1.3.1
 pydantic==2.12.5
-uvicorn==0.40.0
+uvicorn==0.51.0
 tortoise-orm==0.25.3
-python-multipart==0.0.21
+python-multipart==0.0.32

tests/test_security_hardening.py

@@ -0,0 +1,93 @@
+import asyncio
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from fastapi import HTTPException
+
+from apps.admin import dependencies as admin_dependencies
+from apps.admin.dependencies import create_token, verify_token
+from apps.admin.services import LocalFileClass
+from core.settings import data_root, settings
+from core.storage import SystemFileStorage
+from core.utils import hash_password, verify_password
+
+
+class PasswordCompareTests(unittest.TestCase):
+    def test_verify_password_accepts_valid_hash(self):
+        hashed = hash_password("s3cret-pass")
+        self.assertTrue(verify_password("s3cret-pass", hashed))
+        self.assertFalse(verify_password("wrong-pass", hashed))
+
+    def test_verify_password_supports_legacy_plaintext(self):
+        self.assertTrue(verify_password("legacy", "legacy"))
+        self.assertFalse(verify_password("legacy", "other"))
+
+
+class LocalFilePathTraversalTests(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.local_root = Path(self._tmpdir.name) / "local"
+        self.local_root.mkdir(parents=True, exist_ok=True)
+        (self.local_root / "safe.txt").write_text("ok", encoding="utf-8")
+        self._data_root_patch = patch(
+            "apps.admin.services.data_root", Path(self._tmpdir.name)
+        )
+        self._data_root_patch.start()
+
+    def tearDown(self):
+        self._data_root_patch.stop()
+        self._tmpdir.cleanup()
+
+    def test_rejects_dotdot_filename(self):
+        with self.assertRaises(HTTPException) as ctx:
+            LocalFileClass("../etc/passwd")
+        self.assertEqual(ctx.exception.status_code, 400)
+
+    def test_rejects_absolute_filename(self):
+        with self.assertRaises(HTTPException) as ctx:
+            LocalFileClass("/etc/passwd")
+        self.assertEqual(ctx.exception.status_code, 400)
+
+    def test_allows_basename_inside_local(self):
+        local_file = LocalFileClass("safe.txt")
+        self.assertTrue(asyncio.run(local_file.exists()))
+        self.assertEqual(local_file.file, "safe.txt")
+        self.assertEqual(local_file.path, (self.local_root / "safe.txt").resolve())
+
+
+class SystemStoragePathTests(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.storage = SystemFileStorage()
+        self.storage.root_path = Path(self._tmpdir.name)
+
+    def tearDown(self):
+        self._tmpdir.cleanup()
+
+    def test_resolve_safe_path_blocks_escape(self):
+        with self.assertRaises(ValueError):
+            self.storage._resolve_safe_path("../etc/passwd")
+
+    def test_resolve_safe_path_allows_nested(self):
+        target = self.storage._resolve_safe_path("share/data/a/b.txt")
+        self.assertTrue(str(target).startswith(str(Path(self._tmpdir.name).resolve())))
+
+
+class AdminJwtUrlSafeTests(unittest.TestCase):
+    def setUp(self):
+        settings.jwt_secret = "j" * 48
+
+    def test_create_and_verify_roundtrip(self):
+        token = create_token({"is_admin": True}, expires_in=60)
+        # urlsafe token 不应依赖标准 base64 填充字符
+        self.assertNotIn("+", token)
+        self.assertNotIn("/", token)
+        payload = verify_token(token)
+        self.assertTrue(payload["is_admin"])
+
+
+if __name__ == "__main__":
+    unittest.main()