Skip to content

36. FastAPI 登录认证 密码与TOKEN(JWT) ​

登录认证的四个步骤 ​

用户提交账号密码 → 服务端核对密码散列 → 签发短期访问令牌 → 客户端在后续请求中携带令牌。认证回答“你是谁”,授权回答“你能做什么”。JWT 只是令牌的一种格式。

密码应使用 Argon2 等专门密码散列算法,不能明文存储,也不能只用快速的 MD5/SHA256。JWT 通常包含可读的编码载荷和签名;签名防篡改,不隐藏正文,因此不要把密码和密钥装进载荷。

安装与配置 ​

powershell
uv add "pwdlib[argon2]" PyJWT python-multipart
uv run python -c "import secrets; print(secrets.token_hex(32))"
$env:JWT_SECRET = '将上一步生成的随机密钥填在这里'
$env:DEMO_PASSWORD = '设置一个本地练习用密码'

安装包叫 PyJWT,导入名叫 jwt。不要安装名称相近但 API 不同的包。密钥与演示密码由环境变量提供,不写入代码或版本库。

完整的 main.py ​

python
import os
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from pydantic import BaseModel
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

SECRET = os.environ["JWT_SECRET"]
if len(SECRET) < 32:
    raise RuntimeError("请配置足够长的随机 JWT 密钥")
ALGORITHM = "HS256"
ISSUER = "fastapi-study"
AUDIENCE = "study-client"
password_hash = PasswordHash.recommended()
DEMO_HASH = password_hash.hash(os.environ["DEMO_PASSWORD"])
# 仅为可运行示例:重启后重建;实际用户和散列应存数据库。
USERS = {"alice": {"id": "1", "username": "alice",
                   "password_hash": DEMO_HASH, "disabled": False}}

app = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl="token")

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

class UserOut(BaseModel):
    id: str
    username: str

def unauthorized():
    return HTTPException(401, "账号、密码或令牌无效",
                         headers={"WWW-Authenticate": "Bearer"})

@app.post("/token", response_model=TokenOut)
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = USERS.get(form.username)
    # 未知账号也做一次散列验证,减少明显的时序差异。
    stored_hash = user["password_hash"] if user else DEMO_HASH
    valid = password_hash.verify(form.password, stored_hash)
    if user is None or not valid:
        raise unauthorized()
    if user["disabled"]:
        raise HTTPException(403, "账号已停用")
    now = datetime.now(timezone.utc)
    token = jwt.encode({
        "sub": user["id"], "iat": now, "exp": now + timedelta(minutes=15),
        "iss": ISSUER, "aud": AUDIENCE,
    }, SECRET, algorithm=ALGORITHM)
    return TokenOut(access_token=token)

def current_user(token: Annotated[str, Depends(oauth2)]):
    try:
        payload = jwt.decode(
            token, SECRET, algorithms=[ALGORITHM],
            issuer=ISSUER, audience=AUDIENCE,
            options={"require": ["sub", "exp", "iat", "iss", "aud"]},
        )
        subject = payload["sub"]
        if not isinstance(subject, str):
            raise unauthorized()
    except InvalidTokenError:
        raise unauthorized()
    user = next((u for u in USERS.values() if u["id"] == subject), None)
    if user is None:
        raise unauthorized()
    if user["disabled"]:
        raise HTTPException(403, "账号已停用")
    return user

@app.get("/me", response_model=UserOut)
def me(user: Annotated[dict, Depends(current_user)]):
    return user

启动 uv run uvicorn main:app --reload。在 /docs 的 Authorize 中填写 alice 和演示密码,授权后请求 /me,应只返回 ID 和用户名。

PowerShell 调用方式,以下在已设置环境变量的终端中执行:

powershell
$token = Invoke-RestMethod -Method Post -Uri http://127.0.0.1:8000/token -Body @{username='alice';password=$env:DEMO_PASSWORD}
Invoke-RestMethod http://127.0.0.1:8000/me -Headers @{Authorization="Bearer $($token.access_token)"}

登录端点接收表单而不是 JSON;访问业务接口使用 Authorization: Bearer ...。tokenUrl 用于说明获取令牌的位置,不会替你创建 /token 路由。

验证 JWT 时不能省的内容 ​

固定允许算法,而不是让令牌自行选择;验证签名、过期时间、签发者和受众;要求关键声明存在;sub 使用稳定的用户 ID 字符串。校验后仍要查询用户是否存在、是否停用,不能只解码就信任。

访问令牌短期有效。修改密码不会自动让所有旧 JWT 失效;登出如果只删除客户端令牌,已泄露副本仍可能有效。需要强制失效时可引入令牌版本、撤销列表或服务端会话;刷新令牌还需安全存储、轮换与重放检测。

上线前如何扩展 ​

用数据库替换演示用户表,注册时散列密码;用户查询和散列验证可放同步依赖或线程池,避免阻塞事件循环。登录增加速率限制、必要的验证码或多因素认证;外部身份平台通常应使用其标准授权流程。

真实传输使用 HTTPS。浏览器令牌存放位置要结合 XSS、CSRF 和会话策略设计;Cookie 与本地存储各有取舍,不应只为了方便默认长期存明文令牌。多 worker 必须使用一致且可管理的签名密钥,轮换时设计过渡期。

自测清单与练习 ​

验证正确登录、错误密码、未知账号、无令牌、签名篡改、过期令牌、错误受众、停用用户。不要只测试成功路径。练习在 current_user 下增加角色依赖,区分 401 与 403;将演示用户迁入第 21 篇模型时,另加密码散列字段且不放入公开响应。