lzf_0626
01ee68034d
背景：客户 / 员工 / 管理员三种身份此前无法区分。但区分逻辑其实早就完备 ——
bootstrap.py 里各 Agent 的 allowed_roles 一直是分开的（CustomerServiceAgent 只要
customer、RiskAgent 要 risk_operator/admin、PlatformProbeAgent 只要 admin），
唯独缺"怎么证明你是谁"；sys_user.password_hash 字段也一直存在，只是全是占位符
（种子写 'x'、worker 身份写 !worker-only-no-password-login!），从没写过真实密码。
实现：
- app/service/auth_service.py：bcrypt 校验 + 签发只含 sub 的 JWT + 审计。令牌里只放 sub
是刻意的：角色/权限/数据范围由 IdentityService 每次请求查库解析
（identity_repository.load_context），权限变更因此立即生效，现有鉴权链路一行未改。
- app/api/controllers/auth.py + app/api/schemas/auth.py：POST /api/v1/auth/tokens，
响应含 roles/data_scope 供前端决定进哪个界面（鉴权仍以库里实时数据为准）。
- tools/set_user_password.py：设密码（客户 123456 / 员工 666666 / 管理员 88888888）。
⚠️ 脚本与文档均标注"仅限演示环境"，这三种弱口令上线前必须更换。
- pyproject / requirements 加 bcrypt（cryptography 只用于 JWT，不提供密码哈希）。
安全约定（逐条有实现与测试）：失败不区分原因 —— 用户不存在/密码错/账号停用返回同一条
401，否则接口就成了账号枚举器；用户不存在时也跑一次 bcrypt 以抹掉时序差异；
成功与失败都写 interaction_audit（actor_id 可空正是为失败场景准备的）；绝不记录密码。
过程中踩到一个自己挖的坑：给登录路由挂了通用的 enforce_rate_limit，而它声明依赖
build_request_context ⇒ 变成"要登录先登录"，所有登录都 401。改为新增
enforce_login_rate_limit：按客户端 IP 独立限流（60 秒 10 次）、不依赖认证上下文。
集成测试据此调整：注入恒放行替身隔离跨用例的计数累积，同时保留一个恒超限用例验证闸门
确实会拦 —— 不能因为加了替身就把这道防线测丢。
接口登记：docs/05 §19 加 A034；并更新 §11 —— 那里原写"JWT 签发、刷新、注销由统一身份
认证模块负责，Agent 平台不重复实现"，现注明平台只做登录这一步，刷新/注销仍归该模块。
验证：ruff 干净 / mypy 183 文件 0 错 / 文档守卫 37 份无重号（此前因 docs/21 重号失败）/
unit+contract 1140 passed / integration 90 passed。
50 lines
2.3 KiB
Python
50 lines
2.3 KiB
Python
"""登录接口：账号密码换访问令牌。
|
|
|
|
`docs/05` §11 把 JWT 的签发划给统一身份认证模块，本路由只做**登录**这一步，
|
|
刷新与注销留待后续（`app/core/security.py` 的 `RevocationStore` 协议已经留好）。
|
|
|
|
三点与其它接口不同的地方，都是有意为之：
|
|
|
|
1. **不依赖 `build_request_context`** —— 登录时本来就还没有身份，要求带令牌就成了
|
|
"要登录先登录"。追踪标识改从 `X-Trace-ID` 请求头取，与 `auth.py:38` 的取法一致。
|
|
2. **挂在 `enforce_rate_limit` 上** —— 这是全平台最需要限流的端点（密码爆破的入口）。
|
|
3. **不写 `Audit` 之外的东西、也不回显失败原因** —— 失败一律 401，消息由
|
|
`AuthService` 统一给出，见那里的模块文档第 1 条。
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.dependencies.database import get_session
|
|
from app.api.dependencies.rate_limit import enforce_login_rate_limit
|
|
from app.api.schemas.auth import LoginRequest
|
|
from app.service.auth_service import AuthService
|
|
|
|
router = APIRouter(
|
|
prefix="/api/v1/auth",
|
|
tags=["auth"],
|
|
# 用 `enforce_login_rate_limit` 而不是通用的 `enforce_rate_limit`：后者声明依赖
|
|
# `build_request_context`，挂在这里就成了"要登录先登录"。详见该函数的文档字符串。
|
|
dependencies=[Depends(enforce_login_rate_limit)],
|
|
)
|
|
|
|
|
|
@router.post("/tokens")
|
|
async def create_access_token(
|
|
payload: LoginRequest,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_session), # noqa: B008
|
|
) -> dict[str, object]:
|
|
"""用工号/账号与密码换一个访问令牌。
|
|
|
|
成功响应是 `docs/05` §3.3 的信封：业务字段全在 `data` 里，`meta` 只有 `trace_id`。
|
|
`data.roles` 与 `data.data_scope` 是**给前端决定进哪个界面用的**——
|
|
真正的鉴权每次请求都由 `IdentityService` 查库解析，不看这里。
|
|
"""
|
|
# 认证失败时请求上下文尚未建立，请求头是唯一可复用的追踪标识（同 `auth.py:38`）。
|
|
trace_id = request.headers.get("X-Trace-ID") or ""
|
|
data = await AuthService(session).login(
|
|
payload.username, payload.password, trace_id=trace_id
|
|
)
|
|
return {"data": data, "meta": {"trace_id": trace_id}}
|