JWT 完全指南

官方文档：https://jwt.io/ | RFC 7519：https://tools.ietf.org/html/rfc7519
适用版本：PyJWT 2.x（2026-05-07 核实）

相关文档：FastAPI完全指南 SQLModel完全指南 Pydantic完全指南

1. 基础概念

JWT 是什么

JWT（JSON Web Token）是一种开放标准（RFC 7519），用于在各方之间安全地传输 JSON 格式的声明信息，常用于 身份认证 和 信息交换。

JWT 结构

JWT 由三部分组成，用 . 分隔：

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9   ← Header（Base64Url）
.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkFsaWNlIiwiZXhwIjoxNzExNjgwMDAwfQ
.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c  ← Signature

常用 Payload 字段（Claims）

JWT vs Session

安装

pip install python-jose[cryptography]   # FastAPI 官方推荐
# 或
pip install PyJWT                        # 更轻量

2. 基础使用（PyJWT）

import jwt
from datetime import datetime, timedelta, timezone

SECRET_KEY = "your-secret-key-min-32-chars"
ALGORITHM = "HS256"

# 生成 Token
def create_token(user_id: int, expire_minutes: int = 30) -> str:
    payload = {
        "sub": str(user_id),
        "exp": datetime.now(timezone.utc) + timedelta(minutes=expire_minutes),
        "iat": datetime.now(timezone.utc),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

# 验证 Token
def decode_token(token: str) -> dict:
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

# 使用
token = create_token(user_id=42)
print(token)

try:
    payload = decode_token(token)
    print(payload["sub"])  # "42"
except jwt.ExpiredSignatureError:
    print("Token 已过期")
except jwt.InvalidTokenError:
    print("Token 无效")

3. Access Token + Refresh Token 方案

单 Token 方案的问题：短有效期则频繁重新登录，长有效期则安全风险高。

双 Token 方案：

• Access Token：有效期短（15 分钟 ~ 2 小时），用于接口认证
• Refresh Token：有效期长（7 ~ 30 天），专门用于换取新的 Access Token

from datetime import datetime, timedelta, timezone
from enum import Enum
import jwt

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

class TokenType(str, Enum):
    ACCESS = "access"
    REFRESH = "refresh"

def create_token(user_id: int, token_type: TokenType, expire: timedelta) -> str:
    payload = {
        "sub": str(user_id),
        "type": token_type,
        "exp": datetime.now(timezone.utc) + expire,
        "iat": datetime.now(timezone.utc),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_access_token(user_id: int) -> str:
    return create_token(user_id, TokenType.ACCESS, ACCESS_TOKEN_EXPIRE)

def create_refresh_token(user_id: int) -> str:
    return create_token(user_id, TokenType.REFRESH, REFRESH_TOKEN_EXPIRE)

def verify_token(token: str, expected_type: TokenType) -> dict:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    if payload.get("type") != expected_type:
        raise jwt.InvalidTokenError("Token 类型不匹配")
    return payload

4. 与 FastAPI 完整集成

# src/core/security.py
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

SECRET_KEY = "your-secret-key-at-least-32-chars"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: int

def create_access_token(user_id: int) -> str:
    payload = {
        "sub": str(user_id),
        "type": "access",
        "exp": datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(user_id: int) -> str:
    payload = {
        "sub": str(user_id),
        "type": "refresh",
        "exp": datetime.now(timezone.utc) + timedelta(days=7),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_access_token(token: str) -> TokenData:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "access":
            raise ValueError
        return TokenData(user_id=int(payload["sub"]))
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token 已过期",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token 无效",
            headers={"WWW-Authenticate": "Bearer"},
        )

# 依赖项：获取当前用户
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    session: AsyncSession = Depends(get_session),
) -> User:
    token_data = verify_access_token(token)
    user = await session.get(User, token_data.user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    if not user.is_active:
        raise HTTPException(status_code=403, detail="账号已被禁用")
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

# src/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/token", response_model=TokenResponse)
async def login(
    form: OAuth2PasswordRequestForm = Depends(),
    session: AsyncSession = Depends(get_session),
):
    user = await auth_user(session, form.username, form.password)
    if not user:
        raise HTTPException(status_code=400, detail="邮箱或密码错误")

    return TokenResponse(
        access_token=create_access_token(user.id),
        refresh_token=create_refresh_token(user.id),
    )

@router.post("/refresh", response_model=TokenResponse)
async def refresh(refresh_token: str, session: AsyncSession = Depends(get_session)):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise ValueError
        user_id = int(payload["sub"])
    except Exception:
        raise HTTPException(status_code=401, detail="Refresh Token 无效")

    user = await session.get(User, user_id)
    if not user or not user.is_active:
        raise HTTPException(status_code=401, detail="用户不存在或已禁用")

    return TokenResponse(
        access_token=create_access_token(user.id),
        refresh_token=create_refresh_token(user.id),
    )

# 受保护的接口
@router.get("/me", response_model=UserResponse)
async def get_me(current_user: CurrentUser):
    return current_user

5. 密码哈希（配套使用）

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)

async def auth_user(session: AsyncSession, email: str, password: str) -> User | None:
    result = await session.exec(select(User).where(User.email == email))
    user = result.first()
    if not user or not verify_password(password, user.hashed_password):
        return None
    return user

6. Refresh Token 撤销（黑名单）

JWT 无状态，Token 签发后无法直接撤销。解决方案：用 Redis 维护黑名单。

import redis.asyncio as aioredis
from datetime import datetime, timezone
import jwt

redis_client = aioredis.from_url("redis://localhost:6379/0")

async def revoke_token(token: str) -> None:
    """将 token 加入黑名单"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        exp = payload.get("exp", 0)
        ttl = exp - int(datetime.now(timezone.utc).timestamp())
        if ttl > 0:
            await redis_client.setex(f"blacklist:{token}", ttl, "1")
    except Exception:
        pass

async def is_token_revoked(token: str) -> bool:
    return bool(await redis_client.exists(f"blacklist:{token}"))

# 在验证函数中增加黑名单检查
async def verify_access_token_with_blacklist(token: str) -> TokenData:
    if await is_token_revoked(token):
        raise HTTPException(status_code=401, detail="Token 已失效")
    return verify_access_token(token)

7. 常用代码段

从请求头提取 Token（不用 OAuth2PasswordBearer）

from fastapi import Request, HTTPException

def get_token_from_header(request: Request) -> str:
    auth = request.headers.get("Authorization")
    if not auth or not auth.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="缺少认证信息")
    return auth.split(" ")[1]

生成临时操作 Token（邮件验证、重置密码）

def create_email_verify_token(email: str) -> str:
    payload = {
        "sub": email,
        "type": "email_verify",
        "exp": datetime.now(timezone.utc) + timedelta(hours=24),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_email_token(token: str) -> str:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    if payload.get("type") != "email_verify":
        raise ValueError("Token 类型错误")
    return payload["sub"]  # 返回 email

8. 最佳实践

SECRET_KEY 管理

# 生成安全的随机密钥
import secrets
print(secrets.token_hex(32))  # 生成 64 字符的十六进制密钥

# 从环境变量读取（不要硬编码）
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

settings = Settings()  # 从 .env 或环境变量读取

Access Token 有效期建议

9. 踩坑与注意事项

JWT 的 Payload 不是加密的

JWT 只是签名（防篡改），不是加密。Payload 可以被任何人解码（Base64Url 解码即可），不要在 Payload 中存放敏感信息（密码、银行卡号等）。

时区问题

# 必须使用带时区的 datetime，否则 PyJWT 在新版本会报 warning
from datetime import datetime, timezone

# 正确
exp = datetime.now(timezone.utc) + timedelta(minutes=30)

# 错误（naive datetime）
exp = datetime.utcnow() + timedelta(minutes=30)

算法指定不能省略

验证时必须显式指定 algorithms 参数，防止算法混淆攻击（如 RS256 → HS256 降级攻击）：

# 正确
jwt.decode(token, SECRET_KEY, algorithms=["HS256"])

# 危险（允许任意算法）
jwt.decode(token, SECRET_KEY, algorithms=["HS256", "RS256", "none"])

最佳实践

Payload 只存最少信息，不存敏感数据：JWT payload 是 Base64 编码（非加密），任何人解码即可读取。只存 user_id、角色等标识符，绝对不存密码、手机号、身份证等。

Access Token 短有效期 + Refresh Token 长有效期：Access Token 15–30 分钟过期减少泄露风险，用 Refresh Token（7–30 天）换取新 Access Token，无需用户重新登录。

access_token = create_token(user_id, expires_minutes=15)
refresh_token = create_token(user_id, expires_minutes=60*24*7)  # 7天

服务端验证所有需要保护的端点：不要假设"内部接口"不需要验证。每个需要认证的路由都必须验证 JWT。

用 RS256 替代 HS256（多服务场景）：HS256 使用同一密钥签名和验证，所有服务都需要知道密钥，密钥泄露风险高。RS256 只有认证服务持有私钥，其他服务只需公钥验证。

实现 Token 黑名单以支持即时登出：JWT 在过期前始终有效，用户登出后 Token 仍可使用。将已登出的 JTI（JWT ID）存入 Redis，验证时检查是否在黑名单。

import uuid
jti = str(uuid.uuid4())
payload = {"sub": user_id, "jti": jti, "exp": ...}
# 登出时：redis.setex(f"blacklist:{jti}", ttl, "1")
# 验证时：if redis.exists(f"blacklist:{payload['jti']}"): raise InvalidToken

常见陷阱

陷阱：未验证算法导致 "alg:none" 攻击

现象： 攻击者将 JWT header 中的 alg 改为 none，删除签名部分，服务端仍然接受。

原因： 早期 JWT 库允许 alg:none 跳过签名验证；若不显式指定 algorithms，库可能默认接受所有算法。

解决： 验证时始终显式传入允许的算法列表，拒绝 none。

# 危险
payload = jwt.decode(token, key)             # 可能接受 alg:none

# 安全
payload = jwt.decode(token, key, algorithms=["HS256"])

陷阱：使用 naive datetime 导致 exp 验证失败

现象： 生成 Token 时用 datetime.utcnow()，在某些 PyJWT 版本下验证时报 TypeError: can't compare offset-naive and offset-aware datetimes。

原因： PyJWT 2.x 要求 exp 是 aware datetime（带时区信息），datetime.utcnow() 是 naive（无时区）。

解决： 改用 datetime.now(timezone.utc) 或直接用 int(time.time()) 传 Unix 时间戳。

from datetime import datetime, timezone, timedelta

# 正确：aware datetime
exp = datetime.now(timezone.utc) + timedelta(minutes=30)

# 或：Unix 时间戳
import time
exp = int(time.time()) + 30 * 60

陷阱：Token 泄露无法即时撤销

现象： 用户举报账号被盗，但只能等 Access Token 自然过期（可能还有 29 分钟），无法立即失效。

原因： JWT 是无状态的，服务端不存储已发放的 Token 列表，无法主动撤销。

解决： 实现 Token 黑名单（Redis）；或缩短 Access Token 有效期；或在高安全场景改用有状态的 session token。

参见

• FastAPI完全指南
• SQLModel完全指南
• Pydantic完全指南
• Redis完全指南