OAuth2（含密码哈希）与带 JWT 令牌的 Bearer 认证

FastAPI 教程示例

这段代码是一个基于 FastAPI 框架实现的、完整的用户认证与授权系统。它使用了行业标准的 OAuth2 协议和 JWT (JSON Web Token) 技术来安全地验证用户身份，并保护后续的 API 接口。

from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from pydantic import BaseModel

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
        "disabled": False,
    }
}


class Token(BaseModel):
    access_token: str
    token_type: str


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


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


password_hash = PasswordHash.recommended()

DUMMY_HASH = password_hash.hash("dummypassword")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return password_hash.verify(plain_password, hashed_password)


def get_password_hash(password):
    return password_hash.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        verify_password(password, DUMMY_HASH)
        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: Annotated[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 = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    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 Token(access_token=access_token, token_type="bearer")


@app.get("/users/me/")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)],
) -> User:
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return [{"item_id": "Foo", "owner": current_user.username}]

核心配置与模拟数据库

这部分定义了生成和验证 JWT 令牌所需的关键参数，以及一个模拟的用户数据库。

# 用于对 JWT 令牌进行签名和验证的密钥。在生产环境中，必须保密！
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
# 使用的加密算法，这里是 HMAC-SHA256
ALGORITHM = "HS256"
# 令牌有效期，30分钟
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 模拟的用户数据库，实际应用中应替换为真实数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        # 这是用户密码的哈希值，而非明文密码
        "hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
        "disabled": False,
    }
}

专业知识点：密码学安全
- SECRET_KEY: 用于对 JWT 令牌进行签名的密钥。服务端用它来“盖章”（签名）令牌，确保其未被篡改。这个密钥必须严格保密，一旦泄露，攻击者就可以伪造任意令牌。
- ALGORITHM: 指定的 HS256 是一种对称加密算法，意味着签名和验证使用的是同一个密钥。对于分布式系统，有时会使用非对称算法（如 RSA）。

数据模型定义 (Pydantic)

这部分定义了 API 交互中使用的数据结构。

# 令牌模型，用于定义登录成功后返回给客户端的数据格式
class Token(BaseModel):
    access_token: str  # 令牌字符串本身
    token_type: str    # 令牌类型，通常是 "bearer"

# 令牌数据模型，用于定义令牌中携带的用户信息的结构
class TokenData(BaseModel):
    username: str | None = None  # 这里只存放了用户名

# 用户模型，定义了用户的基本信息
class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

# 用户数据库模型，继承自User，并增加了hashed_password字段
class UserInDB(User):
    hashed_password: str

专业知识点：数据验证与序列化
- Pydantic Models: 这些类继承自 BaseModel，是 Pydantic 库的核心功能。它们用于： 1) 数据验证：确保传入或传出的数据符合预定义的类型和结构。 2) 数据序列化/反序列化：方便地将 Python 对象转换为 JSON 格式（发送给客户端），或从 JSON 格式转换为 Python 对象（接收自客户端）。

核心安全工具函数

这部分包含了密码哈希处理和 JWT 令牌创建的核心逻辑。

# 初始化密码哈希工具，使用推荐的算法（通常是 Argon2）
password_hash = PasswordHash.recommended()

# 一个用于防止时序攻击的dummy哈希值
DUMMY_HASH = password_hash.hash("dummypassword")

# 验证用户输入的密码是否与存储的哈希值匹配
def verify_password(plain_password, hashed_password):
    return password_hash.verify(plain_password, hashed_password)

# 对用户密码进行哈希处理，用于存储
def get_password_hash(password):
    return password_hash.hash(password)

# 根据用户名从数据库中获取用户信息
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

# 认证用户：验证用户名和密码
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        # 关键安全实践：即使用户不存在，也进行一次哈希验证，防止时序攻击
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

# 创建访问令牌（JWT）
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

专业知识点：密码存储安全与时序攻击
- 密码哈希 (Password Hashing): 密码绝不能以明文形式存储。pwdlib 库使用 Argon2 算法（2015年密码哈希竞赛 winner），它通过加盐（salt）和多次迭代，极大增加了暴力破解的难度。
- 防止时序攻击 (Timing Attack): 在 authenticate_user 函数中，如果用户不存在，代码仍会调用 verify_password。这是为了确保用户存在和不存在两种情况下的函数执行时间大致相同，防止攻击者通过精确测量响应时间来判断用户名是否存在。
专业知识点：JWT (JSON Web Token)
- 结构: 由三部分组成：Header（算法）、Payload（数据，如用户名、过期时间）和 Signature（签名）。
- 无状态认证: 服务端生成令牌后，自身无需保存会话（Session）。客户端在后续请求中携带此令牌，服务端通过验证签名即可确认用户身份，非常适合分布式和微服务架构。

认证依赖项 (FastAPI Dependencies)

这部分是 FastAPI 安全机制的核心，通过依赖注入（Dependency Injection）实现认证逻辑的复用。

# 初始化 OAuth2PasswordBearer，它会从请求头中提取令牌
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 从令牌中获取当前用户信息
async def get_current_user(token: Annotated[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 = payload.get("sub")  # "sub" 是 JWT 标准中代表主题（用户）的字段
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

# 获取当前的活跃用户（非禁用状态）
async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

专业知识点：依赖注入 (Dependency Injection)
- Depends: FastAPI 的核心特性。它允许您将一个函数（依赖项）注入到路由处理函数中。FastAPI 会自动在请求处理前执行这个依赖项，并将其返回值传递给处理函数。这实现了逻辑的解耦和复用。
- oauth2_scheme: 一个内置的依赖项，它会自动从请求的 Authorization 头中提取 Bearer 类型的令牌字符串。

API 路由 (Endpoints)

这部分定义了两个核心 API：一个用于获取令牌，另一个是受保护的接口。

# 登录端点，接受用户名和密码，返回访问令牌
@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    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 Token(access_token=access_token, token_type="bearer")

# 受保护的端点，需要有效的令牌才能访问
@app.get("/users/me/")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)],
) -> User:
    return current_user

# 另一个受保护的端点
@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return [{"item_id": "Foo", "owner": current_user.username}]

专业知识点：OAuth2 协议与 API 安全
- /token 端点: 实现了 OAuth2 的 Resource Owner Password Credentials Grant 流程。客户端通过此接口提交用户名和密码，换取一个访问令牌（Access Token）。
- 受保护的端点: 通过在函数参数中声明 Depends(get_current_active_user)，FastAPI 会在执行函数前自动运行整个认证链： 1) 从请求头获取令牌 (oauth2_scheme) 2) 验证令牌并解析用户信息 (get_current_user) 3) 检查用户是否处于活跃状态 (get_current_active_user)
- 如果任何一步失败，都会直接返回 401 或 400 错误，无法访问受保护的资源。

总结：核心知识点图谱

FastAPI 框架: 利用其高性能、依赖注入（Depends）和自动文档生成等现代特性构建 API。
OAuth2 协议: 一个开放标准的授权框架，用于在不暴露用户密码的情况下，让客户端应用获取对用户资源的有限访问权限。
JWT (JSON Web Token): 一种开放标准，用于在网络应用环境间安全地传输信息。它通过数字签名保证信息的完整性与认证性。
密码学安全: 使用强哈希算法（如 Argon2）存储密码，并通过设置强密钥（SECRET_KEY）和令牌有效期来保障系统安全。
安全最佳实践: 包括防止时序攻击、使用 HTTPS 传输令牌、将令牌存储在安全的地方（如 HTTP Only Cookie）等。

（采用 CC BY-NC-SA 4.0 许可协议进行授权）

本文标题：《 OAuth2（含密码哈希）与带 JWT 令牌的 Bearer 认证》

本文链接：https://yucol.top/tech/OAuth2-JWT.html

本文最后一次更新为天前，文章中的某些内容可能已过时！