fix: #349

Author: vastsa

1

@@ -9,25 +9,25 @@
 from apps.base.dependencies import IPRateLimit
 from apps.base.models import FileCodes
 from core.settings import settings
-from core.utils import get_random_num, get_random_string, max_save_times_desc
+from core.utils import get_random_num, get_random_string, max_save_times_desc, sanitize_filename
 
 
 async def get_file_path_name(file: UploadFile) -> Tuple[str, str, str, str, str]:
     """获取文件路径和文件名"""
     today = datetime.datetime.now()
     storage_path = settings.storage_path.strip("/")  # 移除开头和结尾的斜杠
     file_uuid = uuid.uuid4().hex
-
+    filename = await sanitize_filename(file.filename)
     # 使用 UUID 作为子目录名
     base_path = f"share/data/{today.strftime('%Y/%m/%d')}/{file_uuid}"
 
     # 如果设置了存储路径，将其添加到基础路径中
     path = f"{storage_path}/{base_path}" if storage_path else base_path
 
-    prefix, suffix = os.path.splitext(file.filename)
+    prefix, suffix = os.path.splitext(filename)
     # 保持原始文件名
-    save_path = f"{path}/{file.filename}"
-    return path, suffix, prefix, file.filename, save_path
+    save_path = f"{path}/{filename}"
+    return path, suffix, prefix, filename, save_path
 
 
 async def get_chunk_file_path_name(file_name: str, upload_id: str) -> Tuple[str, str, str, str, str]:

apps/base/utils.py

@@ -22,7 +22,7 @@
 from core.response import APIResponse
 from core.settings import data_root, settings
 from apps.base.models import FileCodes, UploadChunk
-from core.utils import get_file_url
+from core.utils import get_file_url, sanitize_filename
 from fastapi.responses import FileResponse
 
 
@@ -109,10 +109,16 @@ def _save(self, file, save_path):
                 chunk = file.read(self.chunk_size)
 
     async def save_file(self, file: UploadFile, save_path: str):
-        save_path = self.root_path / save_path
-        if not save_path.parent.exists():
-            save_path.parent.mkdir(parents=True)
-        await asyncio.to_thread(self._save, file.file, save_path)
+        path_obj = Path(save_path)
+        directory = str(path_obj.parent)
+        # 提取原始文件名并进行清理
+        filename = await sanitize_filename(path_obj.name)
+        # 构建安全的完整保存路径
+        safe_save_path = self.root_path / directory / 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()
@@ -447,7 +453,7 @@ def _get_path_str(self, path):
 
     def _save(self, file, save_path):
         content = file.file.read()
-        name = file.filename
+        name = save_path(file.filename)
         path = self._get_path_str(save_path)
         self.root_path.get_by_path(path).upload(name, content).execute_query()
 
@@ -627,16 +633,20 @@ async def _delete_empty_dirs(self, file_path: str, session: aiohttp.ClientSessio
 
     async def save_file(self, file: UploadFile, save_path: str):
         """保存文件（自动创建目录）"""
-        # 分离文件名和目录路径
         path_obj = Path(save_path)
         directory_path = str(path_obj.parent)
+        # 提取原始文件名并进行清理
+        filename = await sanitize_filename(path_obj.name)
+        # 构建安全的保存路径
+        safe_save_path = str(Path(directory_path) / filename)
+
         try:
             # 先创建目录结构
             await self._mkdir_p(directory_path)
             # 上传文件
-            url = self._build_url(save_path)
+            url = self._build_url(safe_save_path)
             async with aiohttp.ClientSession(auth=self.auth) as session:
-                content = await file.read()  # 注意：大文件需要分块读取
+                content = await file.read()
                 async with session.put(
                         url, data=content, headers={
                             "Content-Type": file.content_type}

core/storage.py

@@ -4,7 +4,9 @@
 # @Software: PyCharm
 import datetime
 import hashlib
+import os
 import random
+import re
 import string
 import time
 from core.settings import settings
@@ -92,3 +94,27 @@ def gen_desc_en(value: int, desc: str):
     desc_zh += gen_desc_zh(max_timedelta.seconds % 60, "秒")
     desc_en += gen_desc_en(max_timedelta.seconds % 60, "second")
     return desc_zh, desc_en
+
+
+async def sanitize_filename(filename: str) -> str:
+    """
+    安全处理文件名：
+    1. 剥离路径只保留文件名
+    2. 替换非法字符
+    3. 处理空文件名情况
+    """
+    filename = os.path.basename(filename)
+    illegal_chars = r'[\\/*?:"<>|\x00-\x1F]'  # 包含控制字符
+    # 替换非法字符为下划线
+    cleaned = re.sub(illegal_chars, '_', filename)
+    # 处理空格（可选替换为_）
+    cleaned = cleaned.replace(' ', '_')
+    # 处理连续下划线
+    cleaned = re.sub(r'_+', '_', cleaned)
+    # 处理首尾特殊字符
+    cleaned = cleaned.strip('._')
+    # 处理空文件名情况
+    if not cleaned:
+        cleaned = 'unnamed_file'
+    # 长度限制（按需调整）
+    return cleaned[:255]