fastapi添加全局token驗證並解決自帶接口文檔展示問題


本篇文章不太適合從未使用過fastapi的用戶閱讀

為什么會有這篇文章

假期因為疫情沒能回老家,出去玩又擔心疫情,在出租屋待着實在無聊,偶然想起之前朋友推薦過fastapi不錯,所以就初步學習了一下,如有錯誤歡迎大家指正。

官方文檔添加Token的方法

在官方教程中,已經介紹了如何使用(哈希)密碼和JWT Bearer令牌的OAuth2,大家跟着教程來,正常都是可以完成的,只需要把數據的操作替換成從數據庫操作即可。

按照教程開發完以后打開fastapi自帶的接口文檔,發現的確是除了token接口,其他兩個接口/user/me/和/user/me/iterm/都需要登錄以后才能訪問。但是如果你再開發一個新的接口,不依賴get_current_active_user方法,你會發現這個接口就不需要登錄可以直接訪問,並且接口文檔上也沒有鎖的標識。

那么接下來需要解決的問題就有兩個:1、如何給所有接口添加token校驗;2、文檔上展示出來需要登錄后才能訪問的標識(就是個鎖)

如何全局添加token驗證

所謂的token校驗,就是先從請求的請求頭中獲取token,然后再校驗token是否合法,解決方法有以下幾個

方法1:添加全局依賴

然后再讀文檔全局依賴項,你會發現,可以在實例fastapi的時候添加登錄依賴,來實現所有接口均進行token驗證

# 官方例子   是單個APP的情況
from fastapi import Depends, FastAPI, Header, HTTPException

async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])




# 多個APP的時候
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi import APIRouter
async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

# app01里面所有的接口都會校驗token,app02里面所有的接口都不校驗token
app = FastAPI()
app01 = APIRouter(dependencies=[Depends(verify_token), Depends(verify_key)])
app02 = APIRouter(

app.include_router(app01)
app.include_router(app02)

那么有同學肯定會問了,那登錄接口豈不是也會驗證token,我都沒登錄成功哪里來的token!

解決/login接口也會被校驗token的問題

最直接的方法就是修改上面的代碼,過濾掉不需要驗證token的接口,假設我們登錄接口就是/login,修改后的代碼如下,如果多個接口都不需要驗證登錄態,那你就繼續加嘛,就不展開了

# 官方例子修改
from fastapi import Depends, FastAPI, Header, HTTPException,Request

async def verify_token(x_token: str = Header(...),request: Request):
  	path: str = request.get('path')
    if path.startswith('/login'):
      pass
    else x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...),request: Request):
  	path: str = request.get('path')
    if path.startswith('/login'):
      pass
    else x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])

方法2:Oauth2的scopes屬性

這個方法我沒有深入研究,大家可以參考官方文檔:https://fastapi.tiangolo.com/zh/advanced/security/oauth2-scopes/

方法3:使用fastapi的中間件

通過fastapi的中間件,每次接口請求過來先進行token校驗,通過后再進行后續操作,否則直接返回

from fastapi import Depends, FastAPI,Request,status,HTTPException
app = Fastapi()


@app.middleware("http")
async def verify_token(request: Request, call_next):
    auth_error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid authentication credentials",
        headers={"WWW-Authenticate": "Bearer"},  # OAuth2的規范,如果認證失敗,請求頭中返回“WWW-Authenticate”
    )
    # 獲取請求路徑
    path: str = request.get('path')
    # 登錄接口、docs文檔依賴的接口,不做token校驗
    if path.startswith('/login') | path.startswith('/docs') | path.startswith('/openapi'):
        response = await call_next(request)
        return response
    else:
        try:
            # 從header讀取token
            authorization: str = request.headers.get('authorization')
            if not authorization:
                response = Response(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    content="Invalid authentication credentials",
                    headers={"WWW-Authenticate": "Bearer"},  # OAuth2的規范,如果認證失敗,請求頭中返回“WWW-Authenticate”
                )
                return response
            # 拿到token值
            token = authorization.split(' ')[1]
            # 這個是我自己封裝的校驗token的邏輯,大家可以自己替換成自己的
            with SessionLocal() as db:
                if secret.verify_token(db, token):
                    logger.info("token驗證通過")
                    response = await call_next(request)
                    return response
                else:
                    raise auth_error
        except Exception as e:
            logger.error(e)
            raise auth_error

fastapi自帶的接口文檔,不展示登錄標識的問題

通過上面的方法,大家可以發揮自己的想象玩起來token驗證了,但是不知道大家有沒有發現,在打開fastapi自帶的文檔時,接口后面沒有帶有登錄的標識。

image-20220209125545126
大家可以看到,只有/users/me接口攜帶了標識,這個問題怎么解決呢。

首先/users/me是我根據fastapi的教程來寫的接口,通過看代碼和實現可以發現是因為依賴了OAuth2PasswordBearer,可以看代碼

# 代碼有刪減,只是為了說明情況,全部代碼參考:https://fastapi.tiangolo.com/zh/tutorial/security/oauth2-jwt/
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

如果每個接口都添加OAuth2PasswordBearer實例依賴顯然不符合我寫Python的習慣,想到了fast的全局依賴,結合fastapi的全局依賴,那我完全可以把OAuth2PasswordBearer的實例添加到全局dependencies中啊,這么一來就實現了,當前app下所有的接口在接口文檔出都展示需要登錄的標識,新問題就是不需要登錄的接口也展示了需要登錄的標識,但是對我來說問題不大,就沒深追,如果有小伙伴解決了這個問題歡迎留言

from fastapi import FastAPI
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

app = FastAPI(
	dependencies=[Depends(oauth2_scheme)]
	)

上面這個代碼引入了一個新的問題,那就是所有的接口都會去校驗token,大家可以看下OAuth2PasswordBearer這個類的源碼,這個類有個call方法(call方法簡單說就是可以把實例當方法一樣調用,具體的解釋大家可以查一下,我解釋的不太准確),call方法里面會從每次請求的header里面獲取token信息,登錄接口肯定是沒有token信息的,所以登錄接口又不能用了,那怎么辦呢,最簡單的方法就是繼承這個類,重寫call方法,修改后的代碼

from typing import Optional
from fastapi import Request, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.security.utils import get_authorization_scheme_param


class MyOAuth2PasswordBearer(OAuth2PasswordBearer):
    '''
    docs文檔的接口,如果需要登錄后才能訪問,需要添加OAuth2PasswordBearer的依賴才會展示登錄入口,並且依賴了OAuth2PasswordBearer的接口
    才會帶有登錄信息
    全局添加OAuth2PasswordBearer依賴,則登錄接口會陷入死循環,因為登錄接口沒有OAuth2PasswordBearer的信息
    重寫OAuth2PasswordBearer,對於登錄接口,或者指定的接口不讀取OAuth2PasswordBearer,直接返回空字符串
    '''

    def __init__(self, tokenUrl: str):
        super().__init__(
            tokenUrl=tokenUrl,
            scheme_name=None,
            scopes=None,
            description=None,
            auto_error=True
        )

    async def __call__(self, request: Request) -> Optional[str]:
        path: str = request.get('path')
        if path.startswith('/login') | path.startswith('/docs') | path.startswith('/openapi'):
            return ""
        authorization: str = request.headers.get("Authorization")
        scheme, param = get_authorization_scheme_param(authorization)
        if not authorization or scheme.lower() != "bearer":
            if self.auto_error:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Not authenticated",
                    headers={"WWW-Authenticate": "Bearer"},
                )
            else:
                return None
        return param


my_oauth2_scheme = MyOAuth2PasswordBearer(tokenUrl='token')
# 單個app --start
app = FastAPI(dependencies=[Depends(my_oauth2_scheme)])
# 單個app --end

# 多個app
# 掛載在主APP時,會讓所有的接口都帶有登錄標識
app = FastAPI(dependencies=[Depends(my_oauth2_scheme)])
app01 = APIRouter()
app02 = APIRouter(
app.include_router(app01)
app.include_router(app02)
  
# 掛載在子APP時,只有子APP所有的接口都帶有登錄標識
app = FastAPI()
app01 = APIRouter(dependencies=[Depends(my_oauth2_scheme)])
app02 = APIRouter(
app.include_router(app01)
app.include_router(app02)
# 只有app01下所有的接口帶有登錄標識

至此,問題不完美的解決了~~


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM