feat: https://github.com/vastsa/FileCodeBox/issues/442 密码进行hash存储

apps/admin/services.py

@@ -8,6 +8,7 @@ from apps.base.models import FileCodes, KeyValue
 from apps.base.utils import get_expire_info, get_file_path_name
 from fastapi import HTTPException
 from core.settings import data_root
+from core.utils import hash_password, is_password_hashed
 
 
 class FileService:
@@ -76,6 +77,9 @@ class ConfigService:
         if admin_token is None or admin_token == "":
             raise HTTPException(status_code=400, detail="管理员密码不能为空")
 
+        if not is_password_hashed(admin_token):
+            data["admin_token"] = hash_password(admin_token)
+
         for key, value in data.items():
             if key not in settings.default_config:
                 continue

apps/admin/views.py

@@ -17,18 +17,16 @@ from core.response import APIResponse
 from apps.base.models import FileCodes, KeyValue
 from apps.admin.dependencies import create_token
 from core.settings import settings
-from core.utils import get_now
+from core.utils import get_now, verify_password
 
 admin_api = APIRouter(prefix="/admin", tags=["管理"])
 
 
 @admin_api.post("/login")
 async def login(data: LoginData):
-    # 验证管理员密码
+    if not verify_password(data.password, settings.admin_token):
-    if data.password != settings.admin_token:
         raise HTTPException(status_code=401, detail="密码错误")
 
-    # 生成包含管理员身份的token
     token = create_token({"is_admin": True})
     return APIResponse(detail={"token": token, "token_type": "Bearer"})
 

core/utils.py

@@ -47,7 +47,9 @@ async def get_select_token(code: str):
     :return:
     """
     token = settings.admin_token
-    return hashlib.sha256(f"{code}{int(time.time() / 1000)}000{token}".encode()).hexdigest()
+    return hashlib.sha256(
+        f"{code}{int(time.time() / 1000)}000{token}".encode()
+    ).hexdigest()
 
 
 async def get_file_url(code: str):
@@ -95,6 +97,44 @@ async def max_save_times_desc(max_save_seconds: int):
     return desc_zh, desc_en
 
 
+def hash_password(password: str) -> str:
+    """
+    使用 SHA256 + salt 哈希密码
+    返回格式: sha256$<salt>$<hash>
+    """
+    salt = os.urandom(16).hex()
+    password_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest()
+    return f"sha256${salt}${password_hash}"
+
+
+def verify_password(password: str, hashed: str) -> bool:
+    """
+    验证密码是否匹配
+    支持新格式 (sha256$salt$hash) 和旧格式 (明文)
+    """
+    if not hashed:
+        return False
+
+    # 新格式: sha256$salt$hash
+    if hashed.startswith("sha256$"):
+        parts = hashed.split("$")
+        if len(parts) != 3:
+            return False
+        _, salt, stored_hash = parts
+        password_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest()
+        return password_hash == stored_hash
+
+    # 旧格式: 明文比较 (兼容迁移前的数据)
+    return password == hashed
+
+
+def is_password_hashed(password: str) -> bool:
+    """
+    检查密码是否已经是哈希格式
+    """
+    return password.startswith("sha256$") and len(password.split("$")) == 3
+
+
 async def sanitize_filename(filename: str) -> str:
     """
     安全处理文件名：
@@ -105,15 +145,15 @@ async def sanitize_filename(filename: str) -> str:
     filename = os.path.basename(filename)
     illegal_chars = r'[\\/*?:"<>|\x00-\x1F]'  # 包含控制字符
     # 替换非法字符为下划线
-    cleaned = re.sub(illegal_chars, '_', filename)
+    cleaned = re.sub(illegal_chars, "_", filename)
     # 处理空格（可选替换为_）
-    cleaned = cleaned.replace(' ', '_')
+    cleaned = cleaned.replace(" ", "_")
     # 处理连续下划线
-    cleaned = re.sub(r'_+', '_', cleaned)
+    cleaned = re.sub(r"_+", "_", cleaned)
     # 处理首尾特殊字符
-    cleaned = cleaned.strip('._')
+    cleaned = cleaned.strip("._")
     # 处理空文件名情况
     if not cleaned:
-        cleaned = 'unnamed_file'
+        cleaned = "unnamed_file"
     # 长度限制（按需调整）
     return cleaned[:255]

main.py

@@ -22,6 +22,7 @@ from core.logger import logger
 from core.response import APIResponse
 from core.settings import data_root, settings, BASE_DIR, DEFAULT_CONFIG
 from core.tasks import delete_expire_files, clean_incomplete_uploads
+from core.utils import hash_password, is_password_hashed
 
 
 @asynccontextmanager
@@ -63,13 +64,26 @@ async def load_config():
         key="sys_start", defaults={"value": int(time.time() * 1000)}
     )
     settings.user_config = user_config.value
-    # 更新 ip_limit 配置
+
+    await migrate_password_to_hash()
+
     ip_limit["error"].minutes = settings.errorMinute
     ip_limit["error"].count = settings.errorCount
     ip_limit["upload"].minutes = settings.uploadMinute
     ip_limit["upload"].count = settings.uploadCount
 
 
+async def migrate_password_to_hash():
+    if not is_password_hashed(settings.admin_token):
+        hashed = hash_password(settings.admin_token)
+        settings.admin_token = hashed
+        config_record = await KeyValue.filter(key="settings").first()
+        if config_record and config_record.value:
+            config_record.value["admin_token"] = hashed
+            await config_record.save()
+            logger.info("已将管理员密码迁移为哈希存储")
+
+
 app = FastAPI(lifespan=lifespan)
 
 app.add_middleware(
@@ -149,5 +163,9 @@ if __name__ == "__main__":
     import uvicorn
 
     uvicorn.run(
-        app="main:app", host=settings.serverHost, port=settings.serverPort, reload=False, workers=settings.serverWorkers
+        app="main:app",
+        host=settings.serverHost,
+        port=settings.serverPort,
+        reload=False,
+        workers=settings.serverWorkers,
     )