diff --git a/app/service/auth_service.py b/app/service/auth_service.py index e9ef4fe..185bbaa 100644 --- a/app/service/auth_service.py +++ b/app/service/auth_service.py @@ -103,6 +103,16 @@ def issue_access_token(user_id: int) -> tuple[str, int]: return token, expires_in +def hash_password(password: str) -> str: + """生成 bcrypt 哈希（成本因子用库默认值）。 + + 与 `verify_password` 放在一起，是为了让"写密码"和"校验密码"永远用同一套算法 —— + 两个工具脚本（`set_user_password.py` / `create_test_user.py`）都从这里取， + 避免第三次复制粘贴出不一致的实现。 + """ + return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") + + def verify_password(password: str, stored_hash: str | None) -> bool: """常数时间的密码校验；任何异常都当校验失败。 diff --git a/docs/29-Agent组员登录接口使用说明.md b/docs/29-Agent组员登录接口使用说明.md new file mode 100644 index 0000000..11a6db1 --- /dev/null +++ b/docs/29-Agent组员登录接口使用说明.md @@ -0,0 +1,248 @@ +# 登录接口使用说明（给 Agent 组员） + +> **面向**：需要联调登录 / 需要造测试账号的人 +> **接口**：`POST /api/v1/auth/tokens`（`docs/05` §19 的 **A034**） +> **归属**：认证是**平台级能力**，由平台侧维护。业务分支**不要**自己加登录路由 +> （会绕开公共鉴权，违反 `AGENTS.md` 规则 7）。 +> **日期**：2026-09-11 + +--- + +## 1. 一句话 + +用 **`username` + `password`** 换一个 **Bearer 令牌**，之后所有接口带 +`Authorization: Bearer ` 即可。**令牌里只有用户 id**，你的角色、权限、能看多少数据 +全部由服务端按库里的 RBAC 实时解析——所以业务代码**永远只从 `RequestContext` 取身份**， +不要自己解析令牌，也不要硬编码 `user_id`。 + +--- + +## 2. ⚠️ 先记住：登录用的是 `username`，不是用户 id + +演示账号的用户名**不是** `9001/9002/9003`： + +| 用户 id | **username（登录用这个）** | 密码 | 角色 | 进去该看到什么 | +|---|---|---|---|---| +| 9001 | **`cust_t`** | `123456` | `customer` | 客户界面：自己的会话、适当性、知识问答 | +| 9002 | **`risk_t`** | `666666` | `risk_operator` | 风控界面：预警队列、证据、日报 | +| 9003 | **`admin_t`** | `88888888` | `admin` | 管理界面：配置发布、模型端点、审计 | + +> 密码由 `tools/set_user_password.py` 设置。**这三种弱口令仅用于演示**， +> 上线前必须全部更换。 + +--- + +## 3. 接口契约 + +### 请求 + +```http +POST /api/v1/auth/tokens +Content-Type: application/json + +{"username": "cust_t", "password": "123456"} +``` + +- **不需要** `Authorization` 头（本来就没有令牌）。 +- 请求体**只接受这两个字段**（`extra="forbid"`）。多传 `roles`、`user_id` 之类会 **422** —— + 这是有意的，身份只能由服务端解析，不能由调用方声明。 +- 建议带 `X-Trace-ID`（便于和服务端日志对齐；不带也能用）。 + +### 成功响应（`200`，`docs/05` §3.3 统一信封） + +```json +{ + "data": { + "access_token": "eyJhbGciOiJSUzI1NiIs...", + "token_type": "Bearer", + "expires_in": 1800, + "user_id": "9001", + "roles": ["customer"], + "data_scope": "self" + }, + "meta": {"trace_id": "..."} +} +``` + +| 字段 | 用途 | +|---|---| +| `access_token` | 后续所有请求的 `Authorization: Bearer <它>` | +| `expires_in` | 有效期（秒），当前 **1800**（30 分钟） | +| `roles` | **前端据此决定进哪个界面** | +| `data_scope` | 该身份能看的数据范围（`self` / `own_customers` / `all`） | + +> ⚠️ `roles` 与 `data_scope` 是**给前端做界面分流用的**，不是权限凭证。 +> 真正的鉴权每次请求都由服务端查库解析，所以**权限被改后立刻生效**， +> 不用等令牌过期、也不要用它们在前端做安全判断。 + +### 失败响应 + +| 场景 | 状态码 | 错误码 | +|---|---|---| +| 用户名不存在 / 密码错 / 账号停用 / 从没设过密码 | **401** | `AUTHENTICATION_REQUIRED` | +| 缺少字段、多传字段 | **422** | `AGENT_INPUT_INVALID` | +| 登录尝试过于频繁（60 秒 10 次） | **429** | `RATE_LIMITED`（带 `Retry-After`） | + +**401 的消息对所有失败原因都一样**（"用户名或密码不正确"）。这是有意设计——否则这个接口 +就成了账号枚举器。**前端不要试图从 401 的 message 里区分原因**，统一提示"账号或密码错误"。 + +```json +{"error": {"code": "AUTHENTICATION_REQUIRED", "message": "用户名或密码不正确", + "retryable": false, "field_errors": []}, + "meta": {"trace_id": "..."}} +``` + +--- + +## 4. 怎么加一个测试人员 + +### 一条命令 + +```powershell +# 先看现在有哪些账号 +python tools/create_test_user.py --list + +# 建一个客户账号 +python tools/create_test_user.py --id 9010 --username test_cust --role customer --password abc12345 + +# 建一个风控专员账号 +python tools/create_test_user.py --id 9011 --username test_risk --role risk_operator --password abc12345 +``` + +**角色只有三个可选值**：`customer` / `risk_operator` / `admin`（`sys_role` 里现成的三个）。 +要引入**新角色**得同时定义它的权限集合（`sys_role_permission`），那超出这个脚本的范围， +找平台侧。 + +### 脚本会做三件事，并**验证第四件** + +1. 写 `sys_user`（含 bcrypt 密码哈希）； +2. 绑 `sys_user_role`； +3. 打印**真实解析结果**： + +``` +[OK] id=9010 username=test_cust role=customer 已写入 + 解析结果：roles=('customer',) data_scope=self + 权限 10 项 +[OK] 可以用它登录：{"username": "test_cust", "password": "<你刚设的>"} +``` + +4. 第 3 步的验证很关键——它调的是 `IdentityService.resolve`，也就是请求进来时走的**同一条 + 链路**。**不要只看"插入成功"**：`sys_user_role` 有个静默陷阱（下节）。 + +### ⚠️ 一个静默陷阱（脚本已处理，但你手写 SQL 时要当心） + +`MySQL` 的 `DATETIME(0)` 会把微秒**四舍五入到秒**。如果 `sys_user_role.assigned_at` +用"当前时间"写入、进位后落在**未来**，而授权校验是 `assigned_at <= now`，那么： + +> 账号建好了、密码也对、**但一个角色都拿不到** —— 表现为 `roles=()`， +> 登录照样成功，**不报错**。 + +`tools/create_test_user.py` 统一把 `assigned_at` **往前留 5 秒**避开这个窗口。 +若你手写 SQL 或另写脚本，请照做。`tools/seed_test_rbac.py` 里也有这条注释。 + +### 重复执行同一个 `--id` + +是**覆盖**语义（更新用户名 / 密码 / 角色），不会堆出重复行。改角色也用它： + +```powershell +python tools/create_test_user.py --id 9010 --username test_cust --role admin --password abc12345 +``` + +### 单独改密码 + +```powershell +python tools/set_user_password.py --list # 看现状 +python tools/set_user_password.py --user 9010 --password newpass123 # 改一个 +python tools/set_user_password.py # 按演示规则设置 9001/9002/9003 +``` + +--- + +## 5. 前端怎么接 + +### 登录并保存令牌 + +```javascript +async function login(username, password) { + const res = await fetch("/api/v1/auth/tokens", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ username, password }), + }); + if (res.status === 401) throw new Error("账号或密码错误"); // 不要区分原因 + if (res.status === 429) throw new Error("尝试过于频繁，请稍后再试"); + if (!res.ok) throw new Error("登录失败"); + const { data } = await res.json(); + localStorage.setItem("token", data.access_token); // 演示够用； + return data; // 生产建议 httpOnly Cookie +} + +// 按角色分流界面 +const { roles } = await login("cust_t", "123456"); +if (roles.includes("admin")) router.push("/admin"); +else if (roles.includes("risk_operator")) router.push("/risk"); +else router.push("/chat"); +``` + +### 之后每个请求 + +```javascript +fetch("/api/v1/risk/alerts", { + headers: { Authorization: `Bearer ${localStorage.getItem("token")}` }, +}); +``` + +### 401 的处理 + +拿到 **401** 就清掉令牌、回到登录页——它涵盖"令牌缺失/无效/过期/被吊销/账号停用"， +不用也无法区分。**不要**在 401 里重试登录，那会撞上 429。 + +### SSE 接口注意 + +`GET /api/v1/agent-runs/{run_id}/events` 这类 SSE 端点**必须**带 `Authorization` 头， +所以**不能用原生 `EventSource`**（它设不了自定义头），要用支持自定义头的 `fetch` 流式实现 +或合规的 SSE 客户端（`docs/05` §3.2）。 + +--- + +## 6. 常见问题 + +| 现象 | 原因 | +|---|---| +| **登录一直 401** | ① 用错用户名——是 `cust_t` 不是 `9001`；② 密码没设过（`tools/set_user_password.py --list` 会显示"占位符，无法登录"）；③ 账号 `status` 不是 `正常` | +| **登录成功但接口全 403** | 角色的权限不够（不是登录问题）。看该接口要求的权限码，再核对 `sys_role_permission` | +| **登录成功但 `roles` 是 `()`** | 角色没绑上，多数是 `assigned_at` 落在未来——用 `create_test_user.py` 重跑，别手写 SQL | +| **429** | 60 秒内超过 10 次登录（含失败）。等 `Retry-After` 秒 | +| **令牌用一会儿就 401** | 有效期 30 分钟。**当前还没有刷新接口**，过期就重新登录 | +| **改了这个用户的角色，但界面没变** | 前端缓存了 `roles`。令牌里不含权限，重新调一次登录接口拿最新 `roles` 即可 | + +--- + +## 7. 当前边界（做之前先看这里） + +**已有**： +- 登录（账号密码换令牌）、限流、登录成功/失败审计（写在 `interaction_audit`）。 + +**还没有**（别假设它们存在）： +- **刷新令牌**、**注销/吊销**、**改密码接口**、**注册**、**找回密码**。 + 其中刷新与注销按 `docs/05` §11 归"统一身份认证模块"； + `app/core/security.py` 已留好 `RevocationStore` 协议，接上 Redis 即可，属于二期。 +- 令牌**过期只能重新登录**（30 分钟）。 + +**审计**：每次登录成功与失败都会写一条 `interaction_audit` +（`action_type` 为 `auth.login_succeeded` / `auth.login_failed`）。 +排查"谁登过"用 `GET /api/v1/admin/audit-records`。 + +--- + +## 8. 给平台侧的反馈 + +这个接口是平台侧新加的，边界是"**只做登录**"。如果你在联调中发现： + +- 需要**刷新令牌**（不想每 30 分钟重登）； +- 需要**注销**（切换账号、退出登录）； +- 需要更多**角色**（`operator`、`advisor`、`super_admin` 在 `bootstrap.py` 的 + `allowed_roles` 里已经被引用，但 `sys_role` 里还没有对应的角色行）； +- 需要**改密码**； + +直接提，别自己在业务分支里加路由——认证是共用的，两边各加一套会打架。 diff --git a/tools/create_test_user.py b/tools/create_test_user.py new file mode 100644 index 0000000..3aa3e46 --- /dev/null +++ b/tools/create_test_user.py @@ -0,0 +1,213 @@ +"""添加一个可登录的测试账号（用户 + 角色 + 密码），并**验证它真的能拿到权限**。 + +## 为什么需要它 + +`sys_user` / `sys_user_role` / `sys_role` 这些表**没有 ORM 模型**（全项目用裸 SQL 访问， +见 `tools/seed_test_rbac.py`），手写 INSERT 要凑齐 10 个字段、还要自己算密码哈希。 +更要紧的是这里有个**静默陷阱**： + + MySQL 的 DATETIME(0) 会把微秒**四舍五入到秒**。若 `sys_user_role.assigned_at` + 用"当前时间"写入，进位后可能落在未来，而授权校验是 `assigned_at <= now` —— + 于是刚建好的账号**一个角色都拿不到**，表现为 `roles=()`，**不报错也不失败**。 + +`seed_test_rbac.py` 已经踩过一次（它的注释里写着）。本脚本统一把 `assigned_at` +往前留 5 秒，并在最后**用 `IdentityService.resolve` 打印真实解析结果**而不是 +"插入成功" —— 后者根本不能说明这个账号能用。 + +## 用法 + + python tools/create_test_user.py --list + python tools/create_test_user.py --id 9010 --username test_cust \\ + --role customer --password abc12345 + +重复执行同一个 `--id` 是**覆盖**语义：更新用户名 / 密码 / 角色，不会产生重复行。 +""" + +from __future__ import annotations + +import argparse +import asyncio +import sys +from datetime import UTC, datetime, timedelta + +from sqlalchemy import text + +from app.core.contracts import RequestContext +from app.infrastructure.db import SessionFactory +from app.service.auth_service import hash_password +from app.service.identity_service import IdentityService + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] + +#: 库里现成的三个角色（`tools/seed_test_rbac.py` 建的）。新用户复用它们。 +#: 要引入**新角色**得同时定义它的权限集合（`sys_role_permission`），超出本脚本范围。 +ROLE_IDS: dict[str, int] = { + "customer": 9001, + "risk_operator": 9002, + "admin": 9003, +} + +#: 角色 → `sys_user.user_type`。注意这是 `user_type`，与 `employee_role` 不是一回事。 +ROLE_USER_TYPE: dict[str, str] = { + "customer": "customer", + "risk_operator": "employee", + "admin": "employee", +} + +#: 客户的开户状态。风控扫描等链路会读它，写成 `closed` 会让部分规则不成立。 +FUND_ACCOUNT_STATUS: dict[str, str] = { + "customer": "已开户", + "employee": "closed", +} + +#: `assigned_at` 往前留的秒数，见模块文档里的静默陷阱。 +ASSIGN_BACKDATE_SECONDS = 5 + + +async def list_users() -> None: + """列出所有账号、它的角色与密码状态。""" + async with SessionFactory() as session: + rows = ( + await session.execute( + text( + """ + SELECT u.id, u.username, u.user_type, u.status, + u.password_hash, + GROUP_CONCAT(r.role_code ORDER BY r.role_code) AS roles + FROM sys_user u + LEFT JOIN sys_user_role ur ON ur.user_id = u.id + LEFT JOIN sys_role r ON r.id = ur.role_id + GROUP BY u.id, u.username, u.user_type, u.status, u.password_hash + ORDER BY u.id + """ + ) + ) + ).mappings().all() + + print(f"{'id':<8}{'username':<18}{'user_type':<12}{'status':<8}{'roles':<28}密码") + for row in rows: + has_password = str(row["password_hash"] or "").startswith(("$2a$", "$2b$", "$2y$")) + print( + f"{row['id']:<8}{str(row['username']):<18}{str(row['user_type']):<12}" + f"{str(row['status']):<8}{str(row['roles'] or '(无角色)') or '(无角色)':<28}" + f"{'已设' if has_password else '占位符，无法登录'}" + ) + + +async def upsert_user( + *, user_id: int, username: str, role: str, password: str +) -> int: + """建/更新账号并绑定角色，最后验证权限能解析出来。""" + role_id = ROLE_IDS[role] + user_type = ROLE_USER_TYPE[role] + now = datetime.now(UTC).replace(tzinfo=None) + assigned_at = now - timedelta(seconds=ASSIGN_BACKDATE_SECONDS) + + async with SessionFactory() as session, session.begin(): + role_exists = await session.scalar( + text("SELECT id FROM sys_role WHERE id = :role_id"), {"role_id": role_id} + ) + if role_exists is None: + print(f"[失败] 角色 {role}（id={role_id}）不存在，先跑 tools/seed_test_rbac.py") + return 1 + + # 覆盖语义：同一个 id 重跑不会堆出第二行。 + await session.execute( + text( + """ + INSERT INTO sys_user + (id, user_no, username, password_hash, user_type, + professional_investor_status, fund_account_status, status, + created_at, updated_at) + VALUES + (:id, :user_no, :username, :password_hash, :user_type, + 'none', :fund_status, '正常', :now, :now) + AS new + ON DUPLICATE KEY UPDATE + username = new.username, + password_hash = new.password_hash, + user_type = new.user_type, + fund_account_status = new.fund_account_status, + status = '正常', + updated_at = new.updated_at + """ + ), + { + "id": user_id, + "user_no": f"T-{username.upper()[:20]}", + "username": username, + "password_hash": hash_password(password), + "user_type": user_type, + "fund_status": FUND_ACCOUNT_STATUS[user_type], + "now": now, + }, + ) + # 角色绑定先清后插：`sys_user_role` 没有唯一约束，直接插会堆重复行。 + await session.execute( + text("DELETE FROM sys_user_role WHERE user_id = :user_id"), {"user_id": user_id} + ) + await session.execute( + text( + "INSERT INTO sys_user_role (user_id, role_id, assigned_at)" + " VALUES (:user_id, :role_id, :assigned_at)" + ), + {"user_id": user_id, "role_id": role_id, "assigned_at": assigned_at}, + ) + + print(f"[OK] id={user_id} username={username} role={role} 已写入") + return await verify(user_id, username, role) + + +async def verify(user_id: int, username: str, role: str) -> int: + """用真实链路解析身份 —— 这是唯一能证明"这个账号能用的"方式。""" + context = await IdentityService().resolve( + RequestContext(user_id=str(user_id), trace_id="create-test-user") + ) + print(f" 解析结果：roles={context.roles} data_scope={context.data_scope}") + print(f" 权限 {len(context.permissions)} 项") + if role not in context.roles: + print( + "[失败] 角色没有解析出来。最可能的原因是 assigned_at 落在了未来" + "（DATETIME(0) 的毫秒舍入），请重跑本脚本。" + ) + return 1 + print(f"[OK] 可以用它登录：{{\"username\": \"{username}\", \"password\": \"<你刚设的>\"}}") + return 0 + + +async def main() -> int: + parser = argparse.ArgumentParser(description="添加可登录的测试账号") + parser.add_argument("--list", action="store_true", help="列出所有账号与角色") + parser.add_argument("--id", type=int, help="用户 id（9001-9003 已被演示账号占用）") + parser.add_argument("--username", help="登录用户名") + parser.add_argument("--role", choices=sorted(ROLE_IDS), help="角色") + parser.add_argument("--password", help="登录密码（仅限演示环境）") + args = parser.parse_args() + + if args.list: + await list_users() + return 0 + + missing = [ + name + for name, value in ( + ("--id", args.id), ("--username", args.username), + ("--role", args.role), ("--password", args.password), + ) + if value is None + ] + if missing: + print(f"[失败] 缺少参数：{' '.join(missing)}（或直接用 --list 看现有账号）") + return 1 + + return await upsert_user( + user_id=args.id, + username=args.username, + role=args.role, + password=args.password, + ) + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/tools/set_user_password.py b/tools/set_user_password.py index 9525428..cf4d768 100644 --- a/tools/set_user_password.py +++ b/tools/set_user_password.py @@ -29,10 +29,10 @@ import asyncio import sys from datetime import UTC, datetime -import bcrypt from sqlalchemy import text from app.infrastructure.db import SessionFactory +from app.service.auth_service import hash_password if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] @@ -49,10 +49,6 @@ DEMO_PASSWORDS: dict[str, str] = { BCRYPT_PREFIXES = ("$2a$", "$2b$", "$2y$") -def hash_password(password: str) -> str: - return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") - - def _is_real_hash(value: str | None) -> bool: return bool(value) and str(value).startswith(BCRYPT_PREFIXES)