FastAPI: 认证与 JWT — 安全的 API 身份体系

最后更新：2026-08-26

JWT 就像酒店的房卡——Access Token 是日卡（短期有效），Refresh Token 是月卡（长期有效但可随时作废），前台（认证服务）负责发卡和验证。

1. 你将学到

JWT 原理与结构：Header.Payload.Signature 解读
python-jose 签发与验证 Access Token + Refresh Token 双令牌
OAuth2PasswordBearer 与 FastAPI 安全工具集成
密码哈希：passlib + bcrypt 最佳实践
Alice 场景：PriceTracker 的 SaaS 多租户认证——不同订阅计划的权限矩阵

2. Alice 的真实故事

(1) 痛点：API 无认证被恶意爬取

Alice 的 PriceTracker API 完全开放，任何人都可以调用所有端点。竞争对手写了个脚本，每分钟爬取百万条价格数据。更糟糕的是，有人用 API 提交虚假价格数据，污染了整个数据库。Alice 需要用户注册登录、API 认证、不同订阅级别访问控制，但不知道如何安全实现。

(2) JWT 认证的解法

JWT（JSON Web Token）是无状态认证方案：用户登录获取 Token，后续请求携带 Token，服务端验证签名即可——无需存储 Session，天然支持分布式。

PYTHON

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

@app.get("/me")
async def get_me(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)
    return user

(3) 收益

API 认证上线后，未注册用户无法访问受保护端点，恶意爬取被限流中间件+认证双重拦截。Pro 用户可批量导入，Free 用户限制 1000 条，竞争对手只能看到公开数据。

3. JWT 原理与结构

(1) JWT 三段式结构

JWT 由三部分组成：Header（算法）、Payload（数据）、Signature（签名），用 . 分隔。

sequenceDiagram
    participant Client as Bob Frontend
    participant Auth as /auth/login
    participant API as Protected API
    participant Verify as Token Verifier

    Client->>Auth: POST email + password
    Auth->>Auth: Verify credentials
    Auth-->>Client: Access Token + Refresh Token
    Client->>API: GET /products (Bearer Token)
    API->>Verify: Decode & verify signature
    Verify-->>API: User payload
    API-->>Client: 200 OK + data
    
    Note over Client,Auth: Access Token expires after 30 min
    Client->>Auth: POST /auth/refresh (Refresh Token)
    Auth-->>Client: New Access Token

部分	内容	示例
Header	算法类型	`{"alg": "HS256", "typ": "JWT"}`
Payload	用户数据（Claims）	`{"sub": "alice", "role": "admin", "exp": 1700000000}`
Signature	签名	`HMACSHA256(header.payload, secret)`

▶ 示例：JWT 编码与解码

PYTHON

from jose import jwt, JWTError
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=30))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise ValueError("Invalid token")

# Test
token = create_access_token({"sub": "alice", "role": "admin"})
print(f"Token: {token[:50]}...")
payload = verify_token(token)
print(f"Payload: {payload}")

输出：

TEXT 📖 仅展示

# 函数定义成功

(2) Access Token vs Refresh Token

维度	Access Token	Refresh Token
有效期	30 分钟	7 天
用途	访问 API 资源	刷新 Access Token
存储	内存（前端）	HttpOnly Cookie
暴露风险	高（每次请求携带）	低（仅刷新时使用）
作废方式	等待过期	服务端黑名单

4. OAuth2PasswordBearer 集成

▶ 示例：完整认证配置

PYTHON

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=30)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

app = FastAPI()

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=401,
        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
    except JWTError:
        raise credentials_exception
    # In production: query user from database
    user = {"username": username, "role": payload.get("role", "user")}
    return user

@app.post("/auth/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # In production: verify password from database
    if form_data.username != "alice" or form_data.password != "secret":
        raise HTTPException(status_code=401, detail="Incorrect credentials")
    
    access_token = create_access_token(
        data={"sub": form_data.username, "role": "admin"},
        expires_delta=ACCESS_TOKEN_EXPIRE,
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_user)):
    return current_user

输出：

TEXT 📖 仅展示

# 函数定义成功

5. 密码哈希

(1) passlib + bcrypt

密码绝不以明文存储，使用 bcrypt 算法加盐哈希。

▶ 示例：密码哈希与验证

PYTHON

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

# Test
hashed = hash_password("my-secret-password")
print(f"Hashed: {hashed[:30]}...")
print(f"Verify correct: {verify_password('my-secret-password', hashed)}")
print(f"Verify wrong: {verify_password('wrong-password', hashed)}")

输出：

TEXT 📖 仅展示

Hashed: $2b$12$K3YQ8z9wB5eF7gH1j...
Verify correct: True
Verify wrong: False

(2) 密码安全最佳实践

实践	说明
使用 bcrypt	自适应加盐哈希，防彩虹表
不用 MD5/SHA256	无盐、计算太快，容易被暴力破解
密码长度 ≥ 8	强制最小长度
验证前先哈希	避免时序攻击（passlib 已内置）
SECRET_KEY 足够长	≥ 32 字符随机字符串，存环境变量

6. SaaS 多租户权限矩阵

(1) 订阅级别权限设计

▶ 示例：基于订阅的权限依赖

PYTHON

from fastapi import Depends, HTTPException

SUBSCRIPTION_LIMITS = {
    "free": {"max_import": 1000, "websocket": False, "export": False},
    "pro": {"max_import": 100000, "websocket": True, "export": True},
    "enterprise": {"max_import": 1000000, "websocket": True, "export": True},
}

def require_subscription(min_level: str):
    LEVELS = {"free": 0, "pro": 1, "enterprise": 2}
    
    async def check_subscription(user: dict = Depends(get_current_user)):
        user_level = user.get("subscription", "free")
        if LEVELS.get(user_level, 0) < LEVELS.get(min_level, 0):
            raise HTTPException(
                status_code=403,
                detail=f"Requires {min_level} subscription. Current: {user_level}",
            )
        return user
    return check_subscription

@app.get("/api/v1/analytics")
async def get_analytics(user=Depends(require_subscription("pro"))):
    return {"total_products": 1000000, "active_users": 5000}

@app.post("/api/v1/prices/bulk")
async def bulk_import(
    prices: list[PriceCreate],
    user=Depends(require_subscription("free")),
):
    limits = SUBSCRIPTION_LIMITS[user.get("subscription", "free")]
    if len(prices) > limits["max_import"]:
        raise HTTPException(
            status_code=403,
            detail=f"Import limit: {limits['max_import']} for {user['subscription']} plan",
        )
    return {"imported": len(prices)}

输出：

TEXT 📖 仅展示

# 函数定义成功

端点	Free	Pro	Enterprise
GET /products	1000/天	无限	无限
POST /prices	1000/批	100000/批	1000000/批
WebSocket /ws/prices	-	✓	✓
GET /analytics	-	✓	✓
CSV 导出	-	✓	✓
API 限流	100/分钟	1000/分钟	无限

❓ 常见问题

Q JWT 和 Session 有什么区别？

A JWT 无状态（服务端不存储），适合分布式和微服务；Session 有状态（服务端存储），适合单机应用。API 服务推荐 JWT。

Q Token 过期后怎么办？

A Access Token 过期后用 Refresh Token 换新 Access Token。Refresh Token 过期则需重新登录。

Q SECRET_KEY 怎么管理？

A 生产环境用环境变量或密钥管理服务（AWS KMS/HashiCorp Vault），绝不硬编码在代码中。长度至少 32 字符。

Q bcrypt 和 argon2 哪个好？

A Argon2 是密码哈希竞赛冠军，抗 GPU/ASIC 攻击更强。但 bcrypt 久经考验、生态成熟，两者都远好于 MD5/SHA。

Q OAuth2PasswordBearer 和 HTTPBearer 有什么区别？

A OAuth2PasswordBearer 实现了 OAuth2 密码模式，自动在 Swagger UI 显示登录表单；HTTPBearer 是通用 Bearer Token 提取器。推荐前者。

Q JWT 能否实现即时作废？

A JWT 本身无状态，不能即时作废。方案：短过期时间 + Redis 黑名单（存被作废的 Token ID 直到过期）。

📖 小节

JWT 由 Header.Payload.Signature 三段组成，无状态认证天然支持分布式
Access Token（短期）用于 API 访问，Refresh Token（长期）用于刷新，双令牌提升安全性
OAuth2PasswordBearer 集成 FastAPI 安全工具，Swagger UI 自动显示登录表单
密码用 bcrypt 哈希存储，passlib 封装加盐和验证逻辑
PriceTracker 三级订阅权限矩阵：Free/Pro/Enterprise，通过 require_subscription DI 实现

📝 作业

基础题（难度⭐）：实现 /auth/login 端点，接收用户名密码，验证后返回 JWT Access Token（有效期 30 分钟），在 /me 端点用 Depends(oauth2_scheme) 解析用户。提示：OAuth2PasswordRequestForm + jwt.encode
进阶题（难度⭐⭐）：添加 Refresh Token 机制，Access Token 有效期 15 分钟，Refresh Token 有效期 7 天，实现 /auth/refresh 端点用 Refresh Token 换新 Access Token。提示：两个 create_token 函数 + 不同过期时间
挑战题（难度⭐⭐⭐）：实现完整的注册/登录/权限系统——注册端点密码用 bcrypt 哈希存储，require_subscription(min_level) DI 链检查订阅级别，Free 用户限制导入 1000 条，Pro 用户无限制。提示：hash_password + verify_password + require_subscription DI

---|

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

FastAPI: 认证与 JWT — 安全的 API 身份体系

1. 你将学到

2. Alice 的真实故事

(1) 痛点：API 无认证被恶意爬取

(2) JWT 认证的解法

(3) 收益

3. JWT 原理与结构

(1) JWT 三段式结构

▶ 示例：JWT 编码与解码

(2) Access Token vs Refresh Token

4. OAuth2PasswordBearer 集成

▶ 示例：完整认证配置

5. 密码哈希

(1) passlib + bcrypt

▶ 示例：密码哈希与验证

(2) 密码安全最佳实践

6. SaaS 多租户权限矩阵

(1) 订阅级别权限设计

▶ 示例：基于订阅的权限依赖

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好