您的位置:

使用FastAPIJWT构建Web应用的鉴权系统

在Web应用开发中,鉴权系统是必不可少的一个环节。FastAPIJWT是使用JSON Web Token(以下简称JWT)在FastAPI中构建鉴权系统的一种简单、快速但功能强大的方法。JWT是一种开放标准(RFC 7519),用于在网络应用之间传递声明。JWT中的信息使用JSON对象进行编码,以数字签名方式进行验证。

一、在FastAPI中使用FastAPIJWT

在开始使用FastAPIJWT之前,需要使用pip安装FastAPI和FastAPIJWT库:

pip install fastapi
pip install fastapi-jwt-auth

接下来,我们需要在FastAPI应用程序中添加FastAPIJWT的配置项和路由以启用鉴权系统。

在应用程序的主文件中,导入FastAPI、FastAPIJWT库并创建应用程序实例:

from fastapi import FastAPI
from fastapi_jwt_auth import AuthJWT

app = FastAPI()

接下来,我们需要配置FastAPIJWT并在应用程序中注册路由以处理jwt相关的请求。这里我们使用AuthJWT装饰器,该装饰器接受应用程序实例作为参数。配置中我们需要设置SECRET_KEY、ACCESS_TOKEN_EXPIRE_MINUTES和ALGORITHM参数,在路由中我们可以通过AuthJWT.current_user()方法获取当前用户的信息:

# 必须设置的Secret Key以进行JWT加密和解密
app.config["SECRET_KEY"] = "your-secret-key"

# JWT过期时间(分钟)
app.config["ACCESS_TOKEN_EXPIRE_MINUTES"] = 30

# JWT算法
app.config["ALGORITHM"] = "HS256"

# JWT鉴权路由
@app.post("/login")
def login(auth: AuthJWT = Depends()):
    # 获取用户名和密码
    username = form_data.username
    password = form_data.password

    # 验证用户
    is_authenticated = verify_user(username, password)
    if is_authenticated:
        # 创建JWT令牌
        access_token = auth.create_access_token(subject=username)
        return {"access_token": access_token}
    else:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

@app.get("/protected")
def protected_route(user: User = Depends(get_current_user)):
    return {"message": "Hello, {0}!".format(user.username)}

在上面的示例中,我们使用了两个路由:/login和/protected。在/login路由中,我们获取用户的用户名和密码,并验证用户身份。如果用户身份验证成功,则调用AuthJWT.create_access_token()方法生成JWT令牌并返回给客户端。在/protected路由中,我们使用@Depends装饰器中的get_current_user方法来保护该路由。

二、使用FastAPIJWT保护路由

在上一节中,我们已经使用FastAPIJWT创建了鉴权系统。在本节中,我们将使用FastAPIJWT来保护我们的路由,仅允许经过身份验证的用户访问。

首先,我们需要创建一个获取当前用户的函数(也称为依赖项),该函数将在路由中使用。

from fastapi_jwt_auth import AuthJWT
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends

jwt_scheme = HTTPBearer()

# 获取当前用户的函数
def get_current_user(auth: HTTPAuthorizationCredentials = Depends(jwt_scheme), jwt: AuthJWT = Depends()):
    try:
        # 调用AuthJWT.verify_token()方法验证token并返回用户信息
        jwt.verify_token(auth.credentials)
        current_user = jwt.get_jwt_subject()
    except:
        # 验证失败时抛出HTTPException
        raise HTTPException(status_code=401, detail="Invalid token")

    return {"username": current_user}

在上述get_current_user方法中,我们使用@Depends装饰器指定了两个依赖项:auth和jwt。其中auth参数的值为FastAPI的HTTPBearer OAuth类型,用于获取HTTP头部中的JWT密码。jwt参数的值为AuthJWT实例,用于验证JWT密码和获取JWT主题(即用户标识符)。如果验证成功,则返回包含用户名的JSON对象。如果验证失败,则返回401 UnauthorizedHTTPException。接下来,我们可以在要保护的路由中使用该get_current_user方法。

# 保护路由
@app.get("/protected")
def protected_route(user: User = Depends(get_current_user)):
    return {"message": "Hello, {0}!".format(user.username)}

在上述示例中,我们使用fastapi的@Depends装饰器来指定get_current_user方法作为get_protected_route路由的依赖项。如果用户通过身份验证,则返回包含用户信息的JSON对象。如果用户未通过身份验证,则返回401 Unauthorized HTTPException。

三、使用Refresh Token实现JWT自动续期

JWT的过期时间是有限的。但是,如果用户一直在使用应用程序,他们可能希望在该时间段内自动续订他们的JWT。为了实现这一点,FastAPIJWT提供了一个自动续订机制,该机制使用Refresh Token。

在FastAPIJWT中,Refresh Token是一种与访问令牌(access token)相关的短暂的JWT。Refresh Token创建后的短暂有效期内可以用来获取新的访问令牌,从而允许用户无缝延长访问令牌的有效期。

要使用Refresh Token,请指定一个可选的REFRESH_TOKEN_EXPIRE_MINUTES选项,该选项指定Refresh Token的过期时间。以下是实现JWT自动续期的示例:

# JWT自动续订路由
@app.post("/refresh")
def refresh_token(authorize: AuthJWT = Depends()):
    # 验证Refresh Token
    authorize.jwt_refresh_token_required()

    # 创建新的AccessToken
    current_user = authorize.get_jwt_subject()
    new_access_token = authorize.create_access_token(subject=current_user)

    # 返回新的AccessToken
    return {"access_token": new_access_token, "refresh_token": authorize.create_refresh_token(subject=current_user)}

在上述示例中,我们使用AuthJWT装饰器和create_refresh_token()方法创建了Refresh Token。如果Refresh Token存在且有效,则可以调用create_access_token()方法创建新的Access Token。

此时,我们需要相应地更新前一节中的login()路由和get_current_user()函数。

# Login路由
@app.post("/login")
def login(form_data: LoginForm, authorize: AuthJWT = Depends()):
    # 获取用户名和密码
    username = form_data.username
    password = form_data.password

    # 验证用户
    is_authenticated = verify_user(username, password)
    if is_authenticated:
        # 创建JWT令牌和Refresh Token
        access_token = authorize.create_access_token(subject=username)
        refresh_token = authorize.create_refresh_token(subject=username)
        return {"access_token": access_token, "refresh_token": refresh_token}
    else:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

# 获取当前用户的函数
def get_current_user(authorize: AuthJWT = Depends()):
    try:
        # 调用AuthJWT.refresh_jwt_token()方法来刷新Token
        authorize.jwt_refresh_token_required()
        authorize.refresh_jwt_token()
        current_user = authorize.get_jwt_subject()
    except:
        # 验证失败时抛出HTTPException
        raise HTTPException(status_code=401, detail="Invalid token")

    return {"username": current_user}

在上述示例中,我们使用AuthJWT.refresh_jwt_token()方法来刷新JWT。如果Refresh Token有效,则会自动创建新的Access Token并返回给客户端。

四、结语

到目前为止,我们已经使用FastAPIJWT构建了一个功能齐全的Web应用程序鉴权系统,实现了访问控制和自动续订功能。FastAPIJWT是使用FastAPI创建鉴权系统的最佳选择之一,因为它结合了FastAPI的速度和易用性,以及JWT提供的强大的安全保障。