欢迎回到FastAPI教程系列，我们又见面了！到目前为止，我们已经介绍了设置FastAPI、CRUD操作、使用Pydantic进行数据验证、使用依赖注入的异步编程。现在，我们将深入讲解Web API的一个关键部分：认证和授权。确保只有授权用户才能访问特定资源这一点，对于构建安全的应用程序来说至关重要。FastAPI提供了强大的工具来帮助管理这一点。
在这部分，我们将实现基于JSON Web Tokens (JWT)的认证，并且保护敏感的API端点。我们还将介绍权限管理，为用户提供不同的访问权限。
第一步：理解，认证和授权在 FastAPI 中的作用在 FastAPI 中，有两种主要方式来控制访问：
- 身份验证：验证用户的身份，通常通过令牌和凭证。
- 授权：基于用户的角色和权限，确定用户能访问的资源。
我们将使用 JWT 令牌来实现安全的身份验证，并将某些端点仅对经过认证的用户开放。
为什么使用JWT (JSON Web Token) 进行身份验证？- 无状态：不需要服务器端会话存储。
- 安全：令牌经过签名，使其无法被篡改。
- 广泛支持：JWT 可以在许多框架和库中使用。
首先，我们需要一个包来创建和验证JWT。让我们安装一下PyJWT来处理JWT。
您可以使用以下命令安装PyJWT：pip install PyJWT
在 main.py 文件中，我们将定义如下基本配置，包括两个重要的端点。
- 一个
/login端点用于生成 JWT 认证令牌。 - 一个只有通过验证的用户才能访问的
/protected端点。
- 定义 JWT 认证实用程序
我们将创建实用函数来生成和验证JWT。请将 "SECRET_KEY" 替换为既安全又独特的密钥，在生产环境中使用。
import jwt
from datetime import datetime, timedelta
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
SECRET_KEY = "your_secret_key" # 请在生产环境中使用安全的密钥
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # ACCESS_TOKEN过期分钟数
# 生成 JWT 令牌
def create_jwt_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# 检查 JWT 令牌
def verify_jwt_token(token: str):
try:
decoded_token = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return decoded_token if decoded_token["exp"] >= datetime.utcnow() else None
except jwt.PyJWTError: # 处理 JWT 错误
return None- create_jwt_token : 接受数据并生成 JWT 令牌。
- verify_jwt_token : 解码并验证令牌，若有效则返回载荷。
让我们创建一个 /login 接口，用户可以在这里通过提供凭据来获取 JWT 令牌。
从 fastapi 导入 FastAPI, Depends, HTTPException, status
从 pydantic 导入 BaseModel
app = FastAPI()
# 用于演示的假用户数据库
fake_users_db = {
"johndoe": {"username": "johndoe", "password": "secretpassword"}
}
# 登录模型
class Login(BaseModel):
username: str
password: str
@app.post("/login")
async def login(user: Login):
db_user = fake_users_db.get(user.username)
if not db_user or db_user['password'] != user.password:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的用户名或密码",
)
token_data = {"sub": user.username}
token = create_jwt_token(data=token_data)
return {"access_token": token, "token_type": "bearer"}- fake_users_db : 一个假用户数据库。
- /登录接口 : 验证用户成功后返回访问令牌（access token）。
当用户使用有效凭证登录后，会收到一个JWT令牌，必须用它来访问受保护的资源。
第五步：使用 JWT 认证机制保护端点安全为了保护端点的安全，我们将创建一个自定义依赖来通过HTTPBearer类来验证JWT令牌。
security = HTTPBearer()
async def get_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
token = credentials.credentials
payload = verify_jwt_token(token)
if payload 为空:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="令牌无效或已过期",
)
return payload- get_current_user : 从Authorization头中提取并验证JWT令牌。
- HTTPBearer : 提供Bearer令牌的安全保障，通常用于JWT的认证。
接下来，让我们把 获取当前用户 作为依赖来保护这些端点的安全。
让我们加一个 /protected 的端口，只有通过认证的用户才能访问。
@app.get("/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
return {"信息": f"嘿，{current_user['sub']}! 你已成功登录。"}通过 Depends(get_current_user)，端点会检查用户是否已认证并拥有有效令牌，然后才会授予访问权限。这种设置仅允许授权用户访问。
为了增强控制，你可以通过在JWT负载中添加用户角色来实现基于角色的权限控制。
角色授权步骤- 更新 create_jwt_token 函数：添加角色参数。
- 修改 Token 负载内容：在创建令牌时包含用户的角色。
def 创建_jwt_token(data: dict, role: str):
待编码的数据 = data.copy()
待编码的字典.update({"role": role, "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)})
编码后的_jwt = jwt.encode(待编码的字典, SECRET_KEY, algorithm=ALGORITHM)
return 编码后的_jwt3. 保护端点安全：根据角色定义访问权限
@app.get("/admin")
async def admin_route(current_user: dict = Depends(get_current_user)):
if current_user.get("role") != "admin":
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权限访问")
return {"message": "欢迎，管理员!"}只有拥有“admin”角色的用户才能访问 /admin，为 API 增加了一层权限保障。
FastAPI 自动同步 Swagger UI，以包含你的 JWT 鉴权端点。
-
使用 /login 端点获取 JWT 令牌。
-
授权：在 Swagger UI 中，通过点击“Authorize”按钮来输入 JWT 令牌。
- 访问受保护的端点：试着访问
/protected和/admin端点。
有了这样设置，Swagger UI 让测试认证和基于角色的权限管理变得简单。
结论部分.通过在 FastAPI 中使用 JWT (JSON Web Tokens) 认证 和 基于角色的 (Role-Based) 授权，我们创建了一个既强大又安全的 API，该 API 确保只有授权用户可以访问。这些技术对于保护实际应用中的敏感数据和资源来说非常重要。
在第六部分中，我们将探讨如何添加中间件和后台任务，以扩展我们的FastAPI应用，以实现自定义请求处理和后台处理。敬请关注！