feat: api秘钥只能访问api接口，access token只能访问内部接口

app/controller/add_api_route.py

@@ -11,7 +11,7 @@ from app.tools.tool_retrieve_documents import tool_retrieve_documents
 
 def add_api_route(app: FastAPI):
   # 用于dify自定义插件授权认证接口
-  @app.post('/api/account')
+  @app.post('/api/me')
   async def api_account(
     request: Request
   ):

app/controller/add_user_route.py

@@ -5,6 +5,7 @@ from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
 from jwt import InvalidTokenError
 from sqlmodel import select
 from starlette import status
+from starlette.requests import Request
 
 from app.config.env import env
 from app.model.UserModel import RegistryUser, UserModel, UserValidate, PublicUser, UserService
@@ -154,16 +155,8 @@ def add_user_route(app: FastAPI):
 
   # 获取用户信息接口
   @app.get("/users/me")
-  async def _me(session: AsyncSessionDep, current_user: PublicUser = Depends(get_current_user)):
+  async def _me(request: Request):
-    return await UserService.query_item(
+    return request.state.user
-      session=session,
-      row_dict={"id": current_user.id},
-    )
-
-  # 订单查询接口
-  @app.post("/order")
-  async def _query_order(product_name: str, current_user: PublicUser = Depends(get_current_user)):
-    return [product_name]
 
 
 # 用户登录信息验证处理,验证账号密码通过会返回用户信息，否则返回None
@@ -197,11 +190,11 @@ unauthorized_exception = HTTPException(
 
 
 # 获取当前用户信息，通过注入的token来获取当前用户信息，如果token有效则返回用户信息，无效则抛出异常
-async def get_current_user(session: AsyncSessionDep, token: str = Depends(oauth2_scheme)):
+async def get_current_user(session: AsyncSessionDep, token: str):
   try:
     token_info: TokenInfo = CryptUtils.get_token_info(token)
     username = token_info.get('username')
-    if not username or token_info.get('type') != 'access':
+    if not username:
       raise unauthorized_exception
   except InvalidTokenError:
     raise unauthorized_exception

app/middlewares/app_middlewares.py

@@ -10,6 +10,7 @@ from starlette.responses import JSONResponse
 from app.config.env import env
 from app.controller.add_user_route import unauthorized_exception, get_current_user
 from app.model.ApiSecretModel import ApiSecretService
+from app.utils.CrpyUtils import TokenInfo, CryptUtils
 from app.utils.api_secret_utils import api_secret_utils, ApiSecretStatus
 from app.utils.db_utils import async_session
 
@@ -49,9 +50,20 @@ def add_app_middlewares(app: FastAPI):
     if oauth_header and oauth_header.startswith("Bearer "):
       token = oauth_header.split(" ")[1].strip()
 
+    # 没有得到token信息，直接返回错误
     if not token:
       raise unauthorized_exception
 
+    # 从token秘钥中解析token信息
+    try:
+      token_info: TokenInfo = CryptUtils.get_token_info(token)
+      # 既不是access token，也不是api token，直接返回错误信息
+      if token_info.get('type') != 'access' and token_info.get('type') != 'api':
+        raise unauthorized_exception
+    except InvalidTokenError:
+      raise unauthorized_exception
+
+    # 根据token信息获取用户信息
     async with async_session() as session:
       try:
         public_user = await get_current_user(session, token)
@@ -63,6 +75,9 @@ def add_app_middlewares(app: FastAPI):
     # 如果请求的是api接口
     # api开头的接口需要额外验证秘钥
     if request.url.path.startswith("/api/"):
+      # /api/开头的接口，只能使用 api token 访问
+      if token_info.get('type') != "api":
+        raise unauthorized_exception
       # 验证秘钥状态
       secret_status = await api_secret_utils.verify_secret(token)
       print("secret_status", secret_status)
@@ -83,7 +98,10 @@ def add_app_middlewares(app: FastAPI):
       else:
         # 秘钥有效
         pass
-
+    else:
+      # 非/api/接口，智能使用 access token 访问
+      if token_info.get('type') != "access":
+        raise unauthorized_exception
     response = await call_next(request)
     return response
 

app/model/ApiSecretModel.py

@@ -20,7 +20,7 @@ class ApiSecretModel(BasicModel, table=True):
 # 插入数据之前，查询当前用户信息，使用username生成一个access_token作为秘钥保存到 ApiSecretModel中
 async def before_insert(row_dict, session):
   user: UserServiceModel = await UserService.query_item(session, {"id": row_dict.get("created_by")})
-  row_dict["secret"] = CryptUtils.create_token(user.username, "access", timedelta(days=9999))
+  row_dict["secret"] = CryptUtils.create_token(user.username, "api", timedelta(days=9999))
 
 
 # 删除ApiSecret凭据之前，先清理掉缓存中的秘钥信息

app/utils/CrpyUtils.py

@@ -9,7 +9,7 @@ from app.config.env import env
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 # token的类型，access用于接口认证，refresh用于刷新access token，verify用于激活用户账号
-AccessTokenType: TypeAlias = Literal["access", "refresh", "verify"]
+AccessTokenType: TypeAlias = Literal["access", "refresh", "verify", "api"]
 
 
 class TokenInfo(TypedDict):