当前位置：首页 > news >正文

FastAPI 与 JWT 身份验证：保护你的 API

news 来源：原创 2025/6/10 17:54:33

一、引言

在当今的 Web 开发中，保护 API 不受未授权访问至关重要。JWT（JSON Web Token）是一种流行的用于身份验证的开放标准，它允许用户在客户端和服务器之间安全地传输信息。FastAPI 是一个现代、快速的 Web 框架，用于构建 API，它具有许多优秀特性，如自动文档生成功能和依赖注入系统。在本文中，我们将探讨如何在 FastAPI 中实现 JWT 身份验证。

二、什么是 JWT？

JWT 是一种紧凑、自包含的方式，用于在各方之间作为 JSON 对象安全地传输信息。每个令牌由三部分组成，它们之间用点（.）分隔：

头部（Header）：包含令牌类型（通常是 JWT）和使用的签名算法，例如 HMAC SHA256 或 RSA。
载荷（Payload）：包含声明，即关于实体（通常是用户）的某些信息，以及其他的额外信息。
签名（Signature）：用于验证消息在传输过程中未被篡改。

三、为什么使用 JWT？

无状态：服务器不需要在服务器端存储会话信息，减少了服务器的负载。
可扩展：由于服务器不需要保存会话状态，因此更容易扩展到多个服务器。
灵活性：可以在令牌中包含关于用户或会话的任何类型的信息。

四、在 FastAPI 中实现 JWT

1. 项目设置

首先确保安装了 FastAPI 和 Uvicorn。可以使用以下命令安装：

pip install fastapi uvicorn

然后安装 Python_jwt 和 pyjwt 库：

pip install python-jwt pyjwt

2. 导入必要的库

在代码中导入 FastAPI 和其他必要的库：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt
from jwt import PyJWTError
from datetime import datetime, timedelta
from typing import Optional
from pydantic import BaseModel

3. 配置 JWT 密钥和算法

SECRET_KEY = "your-secret-key"  # 在实际应用中使用更安全的方法生成和存储密钥
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

4. 创建用户模型和数据库模拟

fake_users_db = {"john": {"username": "john","password": "secret","email": "john@example.com",}
}class Token(BaseModel):access_token: strtoken_type: strclass TokenData(BaseModel):username: Optional[str] = None

5. 创建身份验证函数

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")def verify_password(plain_password, hashed_password):return plain_password == hashed_password  # 实际应用中使用更安全的哈希方法def get_user(db, username: str):if username in db:user_dict = db[username]return user_dictdef authenticate_user(fake_db, username: str, password: str):user = get_user(fake_db, username)if not user:return Falseif not verify_password(password, user['password']):return Falsereturn userdef create_access_token(data: dict, expires_delta: Optional[timedelta] = None):to_encode = data.copy()if expires_delta:expire = datetime.utcnow() + expires_deltaelse:expire = datetime.utcnow() + timedelta(minutes=15)to_encode.update({"exp": expire})encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)return encoded_jwt

6. 创建 API 路由

app = FastAPI()@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):user = authenticate_user(fake_users_db, form_data.username, form_data.password)if not user:raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Incorrect username or password",headers={"WWW-Authenticate": "Bearer"},)access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)access_token = create_access_token(data={"sub": user['username']}, expires_delta=access_token_expires)return {"access_token": access_token, "token_type": "bearer"}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_exceptiontoken_data = TokenData(username=username)except PyJWTError:raise credentials_exceptionuser = get_user(fake_users_db, username=token_data.username)if user is None:raise credentials_exceptionreturn user@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):return current_user