本篇文章不太適合從未使用過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自帶的文檔時,接口后面沒有帶有登錄的標識。

 大家可以看到,只有/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下所有的接口帶有登錄標識
 
        至此,問題不完美的解決了~~
