feat: 更新管理员登录为简易版jwt

apps/admin/dependencies.py

@@ -2,22 +2,88 @@
 # @Author  : Lan
 # @File    : depends.py
 # @Software: PyCharm
-from fastapi import Header, HTTPException
+from fastapi import Header, HTTPException, Depends
 from fastapi.requests import Request
+import base64
+import hmac
+import json
+import time
 from core.settings import settings
 from apps.admin.services import FileService, ConfigService, LocalFileService
 
+def create_token(data: dict, expires_in: int = 3600 * 24) -> str:
+    """
+    创建JWT token
+    :param data: 数据负载
+    :param expires_in: 过期时间(秒)
+    """
+    header = base64.b64encode(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()).decode()
+    payload = base64.b64encode(json.dumps({
+        **data,
+        "exp": int(time.time()) + expires_in
+    }).encode()).decode()
+    
+    signature = hmac.new(
+        settings.jwt_secret_key.encode(),
+        f"{header}.{payload}".encode(),
+        'sha256'
+    ).digest()
+    signature = base64.b64encode(signature).decode()
+    
+    return f"{header}.{payload}.{signature}"
+
+def verify_token(token: str) -> dict:
+    """
+    验证JWT token
+    :param token: JWT token
+    :return: 解码后的数据
+    """
+    try:
+        header_b64, payload_b64, signature_b64 = token.split('.')
+        
+        # 验证签名
+        expected_signature = hmac.new(
+            settings.jwt_secret_key.encode(),
+            f"{header_b64}.{payload_b64}".encode(),
+            'sha256'
+        ).digest()
+        expected_signature_b64 = base64.b64encode(expected_signature).decode()
+        
+        if signature_b64 != expected_signature_b64:
+            raise ValueError("无效的签名")
+            
+        # 解码payload
+        payload = json.loads(base64.b64decode(payload_b64))
+        
+        # 检查是否过期
+        if payload.get("exp", 0) < time.time():
+            raise ValueError("token已过期")
+            
+        return payload
+    except Exception as e:
+        raise ValueError(f"token验证失败: {str(e)}")
 
 async def admin_required(authorization: str = Header(default=None), request: Request = None):
-    is_admin = (authorization.split(' ')[-1] if authorization else '') == settings.admin_token
-    if request.url.path.startswith('/share/'):
-        if not settings.openUpload and not is_admin:
-            raise HTTPException(status_code=403, detail='本站未开启游客上传，如需上传请先登录后台')
-    else:
-        if not is_admin:
-            raise HTTPException(status_code=401, detail='未授权或授权校验失败')
-    return is_admin
-
+    """
+    验证管理员权限
+    """
+    try:
+        if not authorization or not authorization.startswith('Bearer '):
+            raise ValueError("缺少Bearer token")
+            
+        token = authorization.split(' ')[1]
+        payload = verify_token(token)
+        is_admin = payload.get("is_admin", False)
+        
+        if request.url.path.startswith('/share/'):
+            if not settings.openUpload and not is_admin:
+                raise HTTPException(status_code=403, detail='本站未开启游客上传，如需上传请先登录后台')
+        else:
+            if not is_admin:
+                raise HTTPException(status_code=401, detail='未授权或授权校验失败')
+        return is_admin
+    except ValueError as e:
+        raise HTTPException(status_code=401, detail=str(e))
 
 async def get_file_service():
     return FileService()

apps/admin/schemas.py

@@ -13,3 +13,7 @@ class ShareItem(BaseModel):
 
 class DeleteItem(BaseModel):
     filename: str
+
+
+class LoginData(BaseModel):
+    password: str

apps/admin/views.py

@@ -4,19 +4,30 @@
 # @Software: PyCharm
 import datetime
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from apps.admin.services import FileService, ConfigService, LocalFileService
 from apps.admin.dependencies import admin_required, get_file_service, get_config_service, get_local_file_service
-from apps.admin.schemas import IDData, ShareItem, DeleteItem
+from apps.admin.schemas import IDData, ShareItem, DeleteItem, LoginData
 from core.response import APIResponse
 from apps.base.models import FileCodes, KeyValue
+from apps.admin.dependencies import create_token
+from core.settings import settings
 
 admin_api = APIRouter(prefix='/admin', tags=['管理'])
 
 
 @admin_api.post('/login')
-async def login(admin: bool = Depends(admin_required)):
-    return APIResponse()
+async def login(data: LoginData):
+    # 验证管理员密码
+    if data.password != settings.admin_password:
+        raise HTTPException(status_code=401, detail="密码错误")
+        
+    # 生成包含管理员身份的token
+    token = create_token({"is_admin": True})
+    return APIResponse(detail={
+        "token": token,
+        "token_type": "Bearer"
+    })
 
 
 @admin_api.get('/dashboard')