《AI Agent 工程实战》4.1 认证：JWT 与 API Key 中间件

4.1 认证：JWT 与 API Key 中间件

第 3 章我们把 ResearchCopilot 的 REST 与 SSE 接口都开出来了，但它们现在对全世界敞开：任何人拿到 URL，就能读别人的会话、跑别人的任务。这一节补上「你是谁」这一课——只有先确认了身份，后面「你能做什么」（4.2 授权）和「危险动作要不要二次确认」（4.3 审批）才有意义。认证（authentication）与授权（authorization）是两件事：前者回答「你是谁」，后者回答「你能碰哪些资源」。混在一起写，是很多项目越改越乱的起点。

本节把 ResearchCopilot 推进到「有门」：新增 api/auth.py，用 FastAPI 的依赖注入实现认证，同一个依赖同时接受 Bearer JWT 与 X-API-Key，并让所有认证失败都走 3.3 定下的统一错误契约。

4.1.1 认证为什么必须做成「依赖」，而不是散落各处

最容易写错的版本，是在每个路由函数开头手写一遍「读 header → 解析 token → 校验」。它有三宗罪：重复（每个接口抄一遍）、不一致（有人漏了过期检查，有人忘了校验签名）、不可组合（测试时要绕过认证，得挨个接口打补丁）。

FastAPI 的依赖注入（dependency injection）正好治这三宗罪。把认证写成一个依赖函数，路由用 Depends(authenticate) 声明它需要「已认证的调用方」：

@app.get("/v1/runs/{run_id}")
def get_run(run_id: str, principal: dict = Depends(authenticate)):
    ...

依赖函数要么返回一个 principal（含 workspace_id / user_id），要么直接抛 401。路由函数根本看不到凭证，它只拿到「已经验证过的人」。这样一来，认证逻辑只有一份，测试时用 app.dependency_overrides 一行就能替换掉它。这就是把「横切关注点」从业务代码里抽出来的标准手法。

顺带澄清一个常见疑问：为什么用依赖，而不是 ASGI 中间件？ 两者都能拦截请求，但职责不同：

认证放依赖里，是因为不是每个接口都要认证——健康检查、登录接口本身就不需要。中间件会强迫所有请求走一遍，还得在中间件里判断「这个路径要不要豁免」，反而更绕。

4.1.2 两种凭证：JWT 与 API Key 的分工

ResearchCopilot 面向两类调用方，需要的凭证也不同：

JWT（JSON Web Token） 是一段自包含的字符串：header.payload.signature 三段点分。payload 里放 sub（用户 id）、ws（工作区 id）、exp（过期时间），签名用服务端密钥签出。服务端校验时不需要查库——只要签名对、没过期，就认这个 token。这也是它的代价：签发后无法单方面撤销，只能靠短有效期 + 刷新机制把窗口压小。

API Key 则相反：它本身没有信息，只是一个随机串，服务端拿它去表里查。好处是随时可吊销、可绑定租户；坏处是每次都要查一次存储（生产里用 Redis 缓存，见第 6 章）。

两者用同一个依赖处理，是因为它们最终都要回答同一个问题：这次调用属于哪个 workspace_id、哪个 user_id。把结果统一成 principal 字典，下游代码就不必关心调用方到底用哪种凭证来的。

4.1.3 认证依赖：两种凭证，一个出口

下面是 api/auth.py 的核心。注意 authenticate 先看 Authorization，没有再看 X-API-Key，都没有才拒绝——这个顺序让浏览器用户和脚本用户共用同一个依赖。

# api/auth.py（节选）
import time
import jwt
from fastapi import Header
from starlette.exceptions import HTTPException

# 仅供本地演示：真实环境从密钥管理注入（见 2.2），绝不能硬编码
SECRET = "dev-secret-not-for-production-32bytes"
ALGO = "HS256"

# API Key 表：演示用内存字典；生产在 DB / Redis，且只存哈希
API_KEYS = {"key_acme_demo": {"workspace_id": "ws_acme", "user_id": "u_api"}}


def make_token(sub="u_1", ws="ws_acme", ttl=3600, secret=SECRET) -> str:
    now = int(time.time())
    return jwt.encode(
        {"sub": sub, "ws": ws, "iat": now, "exp": now + ttl},
        secret, algorithm=ALGO,
    )


def authenticate(
    authorization: str | None = Header(default=None),
    x_api_key: str | None = Header(default=None, alias="X-API-Key"),
) -> dict:
    if authorization:
        scheme, _, token = authorization.partition(" ")
        if scheme.lower() != "bearer" or not token:
            raise HTTPException(401, "malformed_authorization")
        try:
            claims = jwt.decode(token, SECRET, algorithms=[ALGO])
        except jwt.ExpiredSignatureError:
            raise HTTPException(401, "token_expired")
        except jwt.InvalidTokenError:
            raise HTTPException(401, "invalid_token")
        return {"workspace_id": claims["ws"], "user_id": claims["sub"], "auth": "jwt"}

    if x_api_key:
        rec = API_KEYS.get(x_api_key)
        if rec is None:
            raise HTTPException(401, "invalid_api_key")
        return {**rec, "auth": "api_key"}

    raise HTTPException(401, "missing_credentials")

几个刻意的选择：

• algorithms=[ALGO] 必须显式传。jwt.decode 如果允许调用方从 token header 里自选算法，就中了经典的 alg=none / 算法混淆攻击——攻击者把 alg 改成 none 或换成对称算法，服务端就可能用错密钥去验。锁死算法列表是硬性纪律。
• 异常分类。ExpiredSignatureError 与其它 InvalidTokenError 分开抛，是因为它们的语义不同：过期是「正常的生命周期结束」，应当引导用户刷新；签名无效是「可疑」，要记审计日志（第 15 章）。
• x_api_key 用 alias="X-API-Key"，因为 Python 参数名不能带连字符。
• API_KEYS 只存明文 key 在演示里可行，生产里要存哈希——明文存储意味着一旦库被读走，所有 key 立刻可用。

4.1.4 真跑：合法 / 过期 / 篡改三种情况

把上面的依赖挂到一个最小 app 上，用 TestClient 跑六种输入。注意 app.exception_handler 把 HTTPException 重包成 3.3 的统一错误体，并注入 request_id：

# demo_41.py（节选）
from fastapi import Depends, FastAPI, Header, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.middleware("http")
async def attach_request_id(request: Request, call_next):
    request.state.request_id = "req_" + request.headers.get("X-Request-Id", "local")
    return await call_next(request)

@app.exception_handler(StarletteHTTPException)
async def error_envelope(request: Request, exc: StarletteHTTPException):
    return JSONResponse(status_code=exc.status_code, content={
        "code": exc.detail,
        "message": MESSAGES.get(exc.detail, exc.detail),
        "details": {},
        "request_id": getattr(request.state, "request_id", "req_?"),
    })

@app.get("/v1/runs/{run_id}")
def get_run(run_id: str, principal: dict = Depends(authenticate)):
    return {"run_id": run_id, "workspace_id": principal["workspace_id"],
            "user_id": principal["user_id"], "auth": principal["auth"]}

client = TestClient(app)
tampered = make_token()[:-3] + "AAA"   # 改掉签名最后三个字符

用 venv 里的 Python 3.14.6 + PyJWT 2.15.1 真跑，输出如下：

合法 JWT                 200  {'run_id': 'run_1', 'workspace_id': 'ws_acme', 'user_id': 'u_1', 'auth': 'jwt'}
过期 JWT                 401  token_expired  token 已过期
篡改签名 JWT              401  invalid_token  token 签名无效或结构损坏
合法 API Key             200  {'run_id': 'run_1', 'workspace_id': 'ws_acme', 'user_id': 'u_api', 'auth': 'api_key'}
无效 API Key             401  invalid_api_key  API Key 无效
无凭证                   401  missing_credentials  缺少凭证：请带 Bearer token 或 X-API-Key

三种失败各有一个稳定的 code，这正是下一小节要展开的契约。注意「篡改签名」用的是合法 token 改掉末尾三个字符——服务端重新计算 HMAC 后发现对不上，直接判 invalid_token，根本不会去看 payload 里的内容。这也是为什么 JWT 敢把 ws 放在客户端可见的 payload 里：内容可读，但改不动。

演示密钥 dev-secret-not-for-production-32bytes 仅用于本地，长度刻意 ≥32 字节——PyJWT 2.15.1 对短于 32 字节的 HMAC 密钥会发 InsecureKeyLengthWarning（RFC 7518 §3.2 要求），生产密钥应由密钥管理生成并轮换。

4.1.5 认证失败的错误契约（与 3.3 统一）

3.3 定下的错误体是扁平四字段 code / message / details / request_id。认证失败不另起炉灶，只是 code 取认证专属的一组值：

落到响应体上，就是 3.3 那个错误体里装进认证专属的 code：

{
  "code": "token_expired",
  "message": "token 已过期",
  "details": {},
  "request_id": "req_8f3c1a"
}

request_id 不是装饰。客户端报「我 401 了」，你靠它能在日志里精确定位这一次请求；没有它，你只能在一堆 401 里猜。所以认证失败也必须带 request_id，不能因为是「安全相关的错误」就少给上下文。

有一条安全纪律必须写进契约：认证失败一律 401，且响应体不透露「这个 key 存在但过期」还是「这个 key 根本不存在」的差别（invalid_token 里刻意不区分「签名错」与「用户不存在」）。响应体越具体，攻击者越省事。唯一的例外是 token_expired——它是给合法用户的刷新信号，暴露它利大于弊。

还有一个容易漏的点：认证失败不能带 WWW-Authenticate 之外的可枚举信息，并且要限速。否则 401 接口本身就成了撞库入口。限速属于第 6 章的配额体系，这里先把契约钉死。

4.1.6 凭证的生命周期：TTL、刷新与吊销

认证一旦上线，「签出去就不管」是最危险的姿态。两种凭证的生命周期管理完全不同：

JWT 的 TTL 选择是个取舍。 太短，用户频繁刷新、体验差；太长，一旦泄露、在过期前都无法撤销。工程上的标准做法是双 token：access_token 短命（如 15 分钟）且无状态校验，refresh_token 长命（如 30 天）且存库可吊销。access 过期就用 refresh 换新的；发现异常就把 refresh 吊销，攻击者最多用 15 分钟。

API Key 反过来：它靠「可吊销」而不是「短命」。 因为 key 常常硬编码在 CI 配置里，换 key 要走发布流程，不能频繁。所以 key 的安全靠三点：只存哈希（脱库也拿不到明文）、可即时吊销（从表里标记即可）、绑定租户与用途（一个 key 只属于一个 workspace，泄露影响面受限）。

还有一个必须提前想的场景：密钥轮换。服务端签名密钥（SECRET）一旦怀疑泄露，要能换掉。但直接换会让所有在飞的 token 立刻失效——用户被迫重新登录。所以生产 JWT 会引入 kid（key id）：token header 里带上 kid，服务端按 kid 选对应密钥验签。签发用新密钥，验签同时认新旧两把，等老 token 自然过期后再撤掉老密钥，全程用户无感。演示里省略了 kid，但它是「无感轮换」的必要条件。

4.1.7 真实环境下会怎样

本节用内存 API_KEYS 字典和固定 SECRET，是为了让示例自足可跑。真实环境有三个必改项：

真实 JWT 还会引入 kid（密钥 id） 支持多密钥轮换：签发用新密钥，验签按 kid 选对应公钥，老 token 到期前仍可验。这是「无感轮换」的标准做法，演示里省略了。

验收清单：

• 认证是一个依赖函数，业务路由只拿到 principal，看不到凭证。
• jwt.decode 显式锁定算法列表，杜绝 alg=none。
• 过期与篡改分成两个 code，且都不泄露多余信息。
• 所有认证失败走 3.3 的统一错误体（code / message / details / request_id），带 request_id。

认证解决了「你是谁」，但「你」能碰哪些数据还没定。下一节把 workspace_id 变成所有资源查询的强制条件，做行级租户隔离。想回看认证失败要遵循的统一错误体，翻 3.3 幂等、分页与统一错误契约 。

阅读导航：上一节：3.3 幂等、分页与统一错误契约 · 下一节：4.2 授权与租户隔离：workspace 行级隔离 。