JWT 认证与 OAuth2 Password Flow

第 7 章 · JWT 认证与 OAuth2 Password Flow

本章目标：在 svc-demo 中实现用户 注册、登录、获取当前用户；使用 python-jose（或 PyJWT）签发与校验 JWT；配置 OAuth2PasswordBearer 与 get_current_user 依赖；用 Bearer Token 保护 ch06 的商品写接口；理解 Access Token 生命周期与安全基线。

学时建议：5～6 小时（含 2 小时 Token 联调）

前置：完成 fastapi-web ch06（Product CRUD、get_db、Repository）；Python 密码学基础见 python-dev 相关章节。

7.1 为什么 API 需要 JWT

前后端分离场景下，浏览器不再依赖服务端 Session Cookie；移动端与 SPA 通过 Authorization: Bearer <token> 携带身份。

客户端                          svc-demo
  POST /auth/login  ──────►  校验邮箱密码
       ◄──────────────────  返回 access_token (JWT)
  POST /api/v1/products
  Header: Authorization: Bearer eyJ...
       ──────────────────►  get_current_user 解码 JWT
       ◄──────────────────  201 Created

示例域名 https://api.example.com 仅用于说明生产网关地址；本章代码均在本地 127.0.0.1:8000 调试。

7.2 安装依赖与安全模块

pip install "python-jose[cryptography]>=3.3,<4"
pip install "passlib[bcrypt]>=1.7,<2"
pip install "python-multipart>=0.0.9"   # OAuth2 表单登录

app/core/security.py：

from datetime import datetime, timedelta, timezone
from typing import Any

from jose import JWTError, jwt
from passlib.context import CryptContext

from app.core.config import settings

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

ALGORITHM = "HS256"


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


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


def create_access_token(subject: str | int, extra: dict[str, Any] | None = None) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": str(subject), "exp": expire, "type": "access"}
    if extra:
        payload.update(extra)
    return jwt.encode(payload, settings.SECRET_KEY, algorithm=ALGORITHM)


def decode_token(token: str) -> dict[str, Any]:
    return jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])

app/core/config.py 增补：

class Settings(BaseSettings):
    # ...
    SECRET_KEY: str = "change-me-in-production-use-openssl-rand"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7

7.3 User 模型与迁移

app/models/user.py：

from datetime import datetime

from sqlalchemy import Boolean, DateTime, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.core.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    full_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    is_superuser: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

alembic revision --autogenerate -m "add users"
alembic upgrade head

app/schemas/user.py：

from pydantic import BaseModel, EmailStr, Field, ConfigDict


class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8, max_length=72)
    full_name: str | None = None


class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: EmailStr
    full_name: str | None
    is_active: bool


class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"


class TokenPayload(BaseModel):
    sub: str | None = None

7.4 OAuth2PasswordBearer 与 get_current_user

app/api/deps.py 扩展：

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import decode_token
from app.models.user import User

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


async def get_current_user(
    token: str = Depends(oauth2_scheme),