FastAPI OAuth2与JWT鉴权完全指南

📂 所属阶段：第四阶段 — 安全与认证（安全篇）
🔗 相关章节：FastAPI依赖注入系统 · FastAPI密码哈希与安全实践

目录

为什么选择JWT认证？

传统Session认证 vs JWT认证

方案优势劣势适用场景
Session认证简单易用，服务器可控有状态，扩展性差，跨域困难单体应用，小型项目
JWT认证无状态，可扩展，跨域友好Token较大，无法主动失效分布式系统，微服务，API

JWT的核心价值

  1. 无状态：服务器无需存储Session信息
  2. 可扩展：适合水平扩展的微服务架构
  3. 跨域友好：天然支持跨域认证
  4. 自包含：Token本身包含用户信息
  5. 标准化：遵循RFC 7519标准

项目依赖安装

# 安装核心依赖
pip install fastapi python-jose[cryptography] passlib[bcrypt] bcrypt python-multipart uvicorn

快速上手：核心实现

JWT结构简析

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.
eyJzdWIiOiIxIiwiZW1haWwiOiJhbGljZUBleGFtcGxlLmNvbSIsInJvbGUiOiJhZG1pbiIsImV4cCI6MTc0MzEyNzIwMH0.
SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c
     ↓                ↓                ↓
  Header          Payload           Signature
  • Header：算法和类型 {"alg":"HS256","typ":"JWT"}
  • Payload：声明（用户信息、过期时间等）
  • Signature：签名，防止伪造

最小化完整示例

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from pydantic import BaseModel

# 配置
SECRET_KEY = "your-secret-key-change-this-in-production-at-least-32-chars"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 模拟数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "hashed_password": "$2b$12$EixZaYb4uX512Gpq5vWveu5G9.5G9.5G9.5G9.5G9.5G9.5G9.5G",
        "email": "johndoe@example.com",
        "role": "user"
    }
}

# 初始化
app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Pydantic模型
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

# 工具函数
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_user(db, username: str):
    if username in db:
        return db[username]

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user["hashed_password"]):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        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
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

# 路由
@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="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["username"], "role": user["role"]},
        expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

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

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

JWT工具模块

模块化JWT管理

# auth/jwt.py
import uuid
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import HTTPException, status

# 配置
SECRET_KEY = "your-secret-key-change-this-in-production-at-least-32-chars"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

# 密码哈希配置
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class JWTManager:
    """JWT管理器 - 集中管理JWT相关操作"""
    
    def __init__(self):
        self.secret_key = SECRET_KEY
        self.algorithm = ALGORITHM
        self.access_token_expire = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        self.refresh_token_expire = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    
    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        """验证密码"""
        return pwd_context.verify(plain_password, hashed_password)
    
    def hash_password(self, password: str) -> str:
        """哈希密码"""
        return pwd_context.hash(password)
    
    def create_access_token(
        self,
        data: Dict[str, Any],
        expires_delta: Optional[timedelta] = None
    ) -> str:
        """创建访问令牌"""
        to_encode = data.copy()
        expire = datetime.now(timezone.utc) + (expires_delta or self.access_token_expire)
        to_encode.update({
            "exp": expire.timestamp(),
            "iat": datetime.now(timezone.utc).timestamp(),
            "jti": str(uuid.uuid4()),
            "type": "access",
        })
        return jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
    
    def create_refresh_token(self, user_id: int) -> str:
        """创建刷新令牌"""
        return self.create_access_token(
            data={"sub": str(user_id), "type": "refresh"},
            expires_delta=self.refresh_token_expire
        )
    
    def decode_token(self, token: str) -> Dict[str, Any]:
        """解码并验证令牌"""
        try:
            return jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
        except JWTError as e:
            raise self._create_credentials_exception(f"Token无效: {str(e)}")
    
    def _create_credentials_exception(self, detail: str = "无法验证凭据") -> HTTPException:
        """创建认证异常"""
        return HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=detail,
            headers={"WWW-Authenticate": "Bearer"},
        )

# 全局JWT管理器实例
jwt_manager = JWTManager()

认证服务与路由

认证服务层

# services/auth_service.py
from typing import Optional
from models.user import User
from auth.jwt import jwt_manager

# 模拟数据库
fake_users_db = {
    1: {
        "id": 1,
        "username": "johndoe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYb4uX512Gpq5vWveu5G9.5G9.5G9.5G9.5G9.5G9.5G9.5G",
        "role": "user",
        "is_active": True,
    }
}

class AuthService:
    """认证服务 - 处理用户认证相关业务逻辑"""
    
    def authenticate_user(self, username: str, password: str) -> Optional[dict]:
        """验证用户凭据"""
        user = next((u for u in fake_users_db.values() if u["username"] == username), None)
        if not user:
            return None
        if not jwt_manager.verify_password(password, user["hashed_password"]):
            return None
        if not user["is_active"]:
            return None
        return user
    
    def get_user_by_id(self, user_id: int) -> Optional[dict]:
        """根据ID获取用户"""
        return fake_users_db.get(user_id)

认证路由

# routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from datetime import timedelta

from auth.jwt import jwt_manager
from services.auth_service import AuthService
from schemas.auth import Token

router = APIRouter(prefix="/auth", tags=["认证"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
auth_service = AuthService()

@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """OAuth2 密码模式登录"""
    user = auth_service.authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = jwt_manager.create_access_token(
        data={"sub": str(user["id"]), "username": user["username"], "role": user["role"]},
    )
    
    refresh_token = jwt_manager.create_refresh_token(user["id"])
    
    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer",
        expires_in=30 * 60,
    )

依赖注入与受保护路由

依赖注入实现

# dependencies.py
from fastapi import Depends, HTTPException, status
from auth.jwt import jwt_manager
from services.auth_service import AuthService
from routers.auth import oauth2_scheme

auth_service = AuthService()

async def get_current_user(token: str = Depends(oauth2_scheme)):
    """获取当前认证用户"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = jwt_manager.decode_token(token)
        user_id: int = int(payload.get("sub"))
        if user_id is None:
            raise credentials_exception
    except Exception:
        raise credentials_exception
    
    user = auth_service.get_user_by_id(user_id)
    if user is None:
        raise credentials_exception
    
    return user

async def get_current_active_user(current_user: dict = Depends(get_current_user)):
    """获取当前活跃用户"""
    if not current_user["is_active"]:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户账户未激活"
        )
    return current_user

async def get_current_admin_user(current_user: dict = Depends(get_current_user)):
    """获取当前管理员用户"""
    if current_user["role"] != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="权限不足：需要管理员权限"
        )
    return current_user

受保护路由

# routers/protected.py
from fastapi import APIRouter, Depends
from dependencies import get_current_active_user, get_current_admin_user

router = APIRouter(prefix="/protected", tags=["受保护路由"])

@router.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_active_user)):
    """获取用户个人信息 - 需要认证用户"""
    return {k: v for k, v in current_user.items() if k != "hashed_password"}

@router.get("/admin/dashboard")
async def admin_dashboard(current_user: dict = Depends(get_current_admin_user)):
    """管理员仪表板 - 需要管理员权限"""
    return {
        "message": "管理员仪表板",
        "user": current_user["username"],
        "role": current_user["role"],
    }

安全最佳实践

核心安全建议

  1. 使用HTTPS：始终通过HTTPS传输令牌
  2. 安全的密钥：使用至少32个字符的强密钥，定期轮换
  3. 短有效期：访问令牌有效期要短（15-60分钟），使用刷新令牌获取新的访问令牌
  4. 令牌黑名单：实现令牌黑名单机制，支持用户主动登出
  5. 密码安全：使用bcrypt等强哈希算法，不要存储明文密码
  6. 速率限制：防止暴力破解和DDoS攻击
  7. 安全头：设置适当的安全头，如X-Content-Type-Options、X-Frame-Options等

企业级安全配置

# config/security.py
import os

class SecurityConfig:
    """安全配置类"""
    
    # JWT配置 - 生产环境使用非对称加密
    JWT_ALGORITHM: str = os.getenv("JWT_ALGORITHM", "HS256")
    JWT_SECRET_KEY: str = os.getenv("JWT_SECRET_KEY", "")
    
    # 密码策略
    PASSWORD_MIN_LENGTH: int = 12
    PASSWORD_REQUIRE_UPPERCASE: bool = True
    PASSWORD_REQUIRE_LOWERCASE: bool = True
    PASSWORD_REQUIRE_NUMBERS: bool = True
    PASSWORD_REQUIRE_SYMBOLS: bool = True
    
    # 速率限制
    LOGIN_RATE_LIMIT: str = "5/minute"
    API_RATE_LIMIT: str = "100/minute"
    
    # 会话管理
    MAX_ACTIVE_SESSIONS: int = 5
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 15
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7

security_config = SecurityConfig()

总结

FastAPI中的OAuth2与JWT认证提供了强大而灵活的安全机制：

  1. 无状态设计：JWT令牌自包含用户信息，适合分布式系统
  2. 灵活的角色控制：通过依赖注入实现精细的权限管理
  3. 安全的令牌管理：支持令牌刷新、黑名单等高级功能
  4. 可扩展的架构：易于集成第三方认证服务

通过遵循安全最佳实践，可以构建企业级的安全认证系统。

💡 关键要点：始终使用HTTPS传输令牌，定期轮换JWT密钥，实现适当的速率限制和审计日志。

🔗 扩展阅读