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