在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提供的强大的安全保障。