FastAPI-06.鉴权与安全

本系列：00 导读 · 01 心智模型 · 02 路由与数据模型 · 03 依赖注入与分层 · 04 中间件异常日志 · 05 异步后台与流式 · 06 鉴权与安全（本文） · 07 测试与项目骨架 · 08 实战 HTTP↔MCP

行文：T3 + T4　｜　本篇方法：刻意练习 + 主动回忆　｜　辅助：间隔重复（清单闪卡）

1. 选型：先回答两个问题

问题	倾向
调用方是脚本/内部服务？	API Key（Header）
调用方是用户+前端登录？	OAuth2 密码流 / 授权码 + JWT
需要细粒度权限？	JWT claims + 依赖里校验 role/scope

段末注释：JWT（JSON Web Token）是一种签名的声明载体，常用于无状态鉴权；OAuth2 是授权框架，FastAPI 内置工具类便于对接。

本系列基线：掌握 API Key + JWT Bearer；OAuth2 授权码留给对接 IdP 时查官方文档。

2. API Key 组块（内部服务）

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import APIKeyHeader

app = FastAPI()

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

VALID_KEYS = {"dev-key-change-me"}  # 生产：环境变量 / Secret


async def verify_api_key(key: str | None = Security(api_key_header)) -> str:
    if key is None or key not in VALID_KEYS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing API Key",
        )
    return key


ApiKeyDep = Annotated[str, Depends(verify_api_key)]


@app.get("/internal/metrics", dependencies=[Depends(verify_api_key)])
async def metrics():
    return {"ok": True}

误用：Key 写进仓库；应 Settings + 环境变量（07）。
误用：HTTPS 外网仍只信 API Key——中间人可嗅探，必须 TLS。

3. JWT Bearer 组块（用户会话）

from datetime import datetime, timedelta, timezone

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

app = FastAPI()

SECRET_KEY = "change-me-use-env"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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


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


async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")


@app.post("/token", response_model=Token)
async def login(form: OAuth2PasswordRequestForm = Depends()):
    # 演示：真实项目查库 + verify_password
    if form.username != "demo" or form.password != "demo":
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    return Token(access_token=create_access_token(form.username), token_type="bearer")


@app.get("/me")
async def read_me(user: str = Depends(get_current_user)):
    return {"username": user}

/docs 点 Authorize，用 /token 拿到的 bearer token 访问 /me。

4. 安全基线清单（间隔重复闪卡源）

#	项	做法
S1	密钥	`SECRET_KEY`、DB 密码进环境变量 / K8s Secret
S2	传输	生产全站 HTTPS；HSTS 由网关处理
S3	CORS	`allow_origins` 白名单，禁 `*` 配凭据
S4	文档	生产可 `docs_url=None` 或加网关鉴权
S5	限流	网关或中间件（本系列不展开库选型）
S6	输入	靠 Pydantic；文件上传限制大小与类型
S7	错误	500 不返回栈（见 04）
S8	依赖	鉴权放 `Depends`，不散落每个 if
S9	密码	bcrypt/argon2，禁止明文存库
S10	Token	短过期 + 刷新策略（按需）

5. 辨析排错

现象	根因	修法
401 但 Header 明明带了 Key	头名与 `APIKeyHeader(name=)` 不一致	对齐 `X-API-Key`
`/docs` 授权后仍 401	`tokenUrl` 路径错	与 `@app.post("/token")` 一致
JWT 过期仍能通过	未校验 `exp` 或时钟漂移	`jwt.decode` 默认校验 exp
CORS + Cookie 登录失败	`allow_credentials` 与 `*` 冲突	指定 origin 列表

6. Drills

#	任务	验收
D1	无 Key 访问 `/internal/metrics`	401
D2	`/token` 取 token 访问 `/me`	200 + username
D3	把 `SECRET_KEY` 挪到 `Settings`	从 `.env` 读取
D4	给 `APIRouter` 整组加 `verify_api_key`	组内路由均受保护

7. 闪卡候选

正面	背面
API Key 放哪？	Header `X-API-Key`（名可自定）
OAuth2PasswordBearer 的 tokenUrl？	换 token 的 POST 路径
401 vs 403？	未认证 vs 已认证无权限
生产能把 SECRET 写代码里吗？	不能

小结

内部用 API Key；用户会话用 OAuth2 + JWT；都通过 Depends 注入。
清单 S1–S10 建议做成闪卡，部署前过一遍。
下一篇 07 测试与项目骨架：可运行、可测的目录与 pytest。