Swagger API 交互文档无法登录认证问题解决

kinit-api/application/settings.py

@@ -11,7 +11,7 @@ from fastapi.security import OAuth2PasswordBearer
 """
 系统版本
 """
-VERSION = "1.10.2"
+VERSION = "1.10.3"
 
 """安全警告: 不要在生产中打开调试运行!"""
 DEBUG = True
@@ -45,7 +45,7 @@ BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 """
 OAUTH_ENABLE = True
 """登录认证视图"""
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login/", auto_error=False) if OAUTH_ENABLE else lambda: ""
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/api/login", auto_error=True) if OAUTH_ENABLE else lambda: ""
 """安全的随机密钥，该密钥将用于对 JWT 令牌进行签名"""
 SECRET_KEY = 'vgb0tnl9d58+6n-6h-ea&u^1#s0ccp!794=kbvqacjq75vzps$'
 """用于设定 JWT 令牌签名算法"""

kinit-api/apps/vadmin/auth/utils/login.py

@@ -22,8 +22,10 @@ PassLib 是一个用于处理哈希密码的很棒的 Python 包。它支持许
 from datetime import timedelta
 from aioredis import Redis
 from fastapi import APIRouter, Depends, Request, Body
+from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
 from core.database import db_getter, redis_getter
+from core.exception import CustomException
 from utils import status
 from utils.response import SuccessResponse, ErrorResponse
 from application import settings
@@ -31,14 +33,39 @@ from .login_manage import LoginManage
 from .validation import LoginForm, WXLoginForm
 from apps.vadmin.record.models import VadminLoginRecord
 from apps.vadmin.auth.crud import MenuDal, UserDal
+from apps.vadmin.auth.models import VadminUser
 from .current import FullAdminAuth
 from .validation.auth import Auth
 from utils.wx.oauth import WXOAuth
 import jwt
 
+
 app = APIRouter()
 
 
+@app.post("/api/login", summary="API 手机号密码登录", description="Swagger API 文档登录认证")
+async def api_login_for_access_token(
+        request: Request,
+        data: OAuth2PasswordRequestForm = Depends(),
+        db: AsyncSession = Depends(db_getter)
+):
+    user = await UserDal(db).get_data(telephone=data.username, v_return_none=True)
+    if not user:
+        raise CustomException(status_code=401, code=401, msg="该手机号不存在")
+    result = VadminUser.verify_password(data.password, user.password)
+    if not result:
+        raise CustomException(status_code=401, code=401, msg="手机号或密码错误")
+    if not user.is_active:
+        raise CustomException(status_code=401, code=401, msg="此手机号已被冻结")
+    elif not user.is_staff:
+        raise CustomException(status_code=401, code=401, msg="此手机号无权限")
+    access_token = LoginManage.create_token({"sub": user.telephone})
+    record = LoginForm(platform='2', method='0', telephone=data.username, password=data.password)
+    resp = {"access_token": access_token, "token_type": "bearer"}
+    await VadminLoginRecord.create_login_record(db, record, True, request, resp)
+    return resp
+
+
 @app.post("/login", summary="手机号密码登录", description="员工登录通道，限制最多输错次数，达到最大值后将is_active=False")
 async def login_for_access_token(
         request: Request,

kinit-api/apps/vadmin/record/models/login.py

@@ -6,7 +6,6 @@
 # @IDE            : PyCharm
 # @desc           : 登录记录模型
 import json
-
 from application.settings import LOGIN_LOG_RECORD
 from apps.vadmin.auth.utils.validation import LoginForm, WXLoginForm
 from utils.ip_manage import IPManage
@@ -14,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from db.db_base import BaseModel
 from sqlalchemy import Column, String, Boolean, TEXT
 from fastapi import Request
+from starlette.requests import Request as StarletteRequest
 from user_agents import parse
 
 
@@ -45,7 +45,7 @@ class VadminLoginRecord(BaseModel):
             db: AsyncSession,
             data: LoginForm | WXLoginForm,
             status: bool,
-            req: Request,
+            req: Request | StarletteRequest,
             resp: dict
     ):
         """
@@ -57,13 +57,17 @@ class VadminLoginRecord(BaseModel):
         header = {}
         for k, v in req.headers.items():
             header[k] = v
-        body = json.loads((await req.body()).decode())
+        if isinstance(req, StarletteRequest):
+            form = (await req.form()).multi_items()
+            params = json.dumps({"form": form, "headers": header})
+        else:
+            body = json.loads((await req.body()).decode())
+            params = json.dumps({"body": body, "headers": header})
         user_agent = parse(req.headers.get("user-agent"))
         system = f"{user_agent.os.family} {user_agent.os.version_string}"
         browser = f"{user_agent.browser.family} {user_agent.browser.version_string}"
         ip = IPManage(req.client.host)
         location = await ip.parse()
-        params = json.dumps({"body": body, "headers": header})
         obj = VadminLoginRecord(
             **location.dict(),
             telephone=data.telephone if data.telephone else data.code,