From 01ee68034d0ad0728d0670fb4e883a139370fd9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 20:43:21 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E7=99=BB=E5=BD=95=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=EF=BC=9A=E8=B4=A6=E5=8F=B7=E5=AF=86=E7=A0=81=E6=8D=A2?= =?UTF-8?q?=E8=AE=BF=E9=97=AE=E4=BB=A4=E7=89=8C=EF=BC=88POST=20/api/v1/aut?= =?UTF-8?q?h/tokens=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景：客户 / 员工 / 管理员三种身份此前无法区分。但区分逻辑其实早就完备 —— 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。 --- app/api/controllers/auth.py | 49 +++++ app/api/dependencies/rate_limit.py | 46 ++++ app/api/schemas/auth.py | 15 ++ app/main.py | 2 + app/service/auth_service.py | 232 +++++++++++++++++++++ docs/05-接口文档.md | 12 ++ pyproject.toml | 1 + requirements.txt | 5 + tests/integration/test_auth_login_mysql.py | 201 ++++++++++++++++++ tools/set_user_password.py | 127 +++++++++++ 10 files changed, 690 insertions(+) create mode 100644 app/api/controllers/auth.py create mode 100644 app/api/schemas/auth.py create mode 100644 app/service/auth_service.py create mode 100644 tests/integration/test_auth_login_mysql.py create mode 100644 tools/set_user_password.py diff --git a/app/api/controllers/auth.py b/app/api/controllers/auth.py new file mode 100644 index 0000000..28ef09c --- /dev/null +++ b/app/api/controllers/auth.py @@ -0,0 +1,49 @@ +"""登录接口：账号密码换访问令牌。 + +`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}} diff --git a/app/api/dependencies/rate_limit.py b/app/api/dependencies/rate_limit.py index 017d2e1..5c06f78 100644 --- a/app/api/dependencies/rate_limit.py +++ b/app/api/dependencies/rate_limit.py @@ -69,3 +69,49 @@ async def enforce_rate_limit( f"请在 {retry_after_seconds} 秒后重试", retry_after_seconds, ) + + +#: 登录端点的限流参数。比普通接口严得多：普通接口的 `policy.max_requests` 是按"已登录用户 +#: 的操作频率"定的，而这里是**密码爆破**的入口，必须独立收紧。 +LOGIN_WINDOW_SECONDS = 60 +LOGIN_MAX_ATTEMPTS = 10 +LOGIN_COUNTER_PREFIX = "login" + + +async def enforce_login_rate_limit(request: Request) -> None: + """登录端点专用的限流闸门：按客户端 IP，**不依赖认证上下文**。 + + 为什么不能复用 `enforce_rate_limit`：它声明依赖 `build_request_context` + （见本模块文档"顺序保证"），挂到登录端点就变成"要登录先登录"——登录请求本来 + 就不带令牌。而爆破恰恰发生在**没有令牌**的时候，所以这里必须另立一个闸门。 + + 维度取客户端 IP + 路由模板：拿不到 `RequestContext.user_id`（那时还没有身份）， + 用 IP 是唯一可用的稳定维度；本地部署里所有客户端可能共用一个出口 IP，但登录 + 端点的价值在于**挡住自动化爆破**，IP 维度足够，且不引入第二套鉴权解析。 + + 降级与 `enforce_rate_limit` 一致：后端返回 `None`（Redis 不可用）时**放行**并告警， + 不因为限流组件故障把所有人挡在门外。 + """ + policy = RateLimitPolicy.from_settings(get_settings()) + if not policy.enabled: + return + client = request.client.host if request.client is not None else "unknown" + template = route_template(request) + result = await get_counter_backend().increment( + f"{LOGIN_COUNTER_PREFIX}:{client}:{request.method}:{template}", + LOGIN_WINDOW_SECONDS, + ) + if result is None: + logger.warning("限流后端不可用，降级放行 route=%s", template) + return + count, retry_after_seconds = result + if count > LOGIN_MAX_ATTEMPTS: + logger.warning( + "登录限流 route=%s ip=%s count=%s limit=%s", + template, client, count, LOGIN_MAX_ATTEMPTS, + ) + raise RateLimitExceededError( + f"登录尝试过于频繁：每 {LOGIN_WINDOW_SECONDS} 秒最多 {LOGIN_MAX_ATTEMPTS} 次，" + f"请在 {retry_after_seconds} 秒后重试", + retry_after_seconds, + ) diff --git a/app/api/schemas/auth.py b/app/api/schemas/auth.py new file mode 100644 index 0000000..778c264 --- /dev/null +++ b/app/api/schemas/auth.py @@ -0,0 +1,15 @@ +from pydantic import BaseModel, ConfigDict, Field + + +class LoginRequest(BaseModel): + """账号密码登录。 + + `extra="forbid"`：多传字段直接 422，避免调用方以为可以塞 `roles`/`user_id` + 之类的越权参数（身份只能由服务端按 RBAC 解析）。 + 长度上限是防滥用，不是密码策略——真正的校验在 `AuthService`。 + """ + + model_config = ConfigDict(extra="forbid") + + username: str = Field(min_length=1, max_length=64) + password: str = Field(min_length=1, max_length=128) diff --git a/app/main.py b/app/main.py index bc793eb..167c41d 100644 --- a/app/main.py +++ b/app/main.py @@ -5,6 +5,7 @@ from fastapi.responses import JSONResponse from app.api.controllers.admin import router as admin_router from app.api.controllers.agent_runs import router as agent_runs_router +from app.api.controllers.auth import router as auth_router from app.api.controllers.conversations import router as conversations_router from app.api.controllers.health import router as health_router from app.api.controllers.knowledge import router as knowledge_router @@ -100,6 +101,7 @@ def create_app() -> FastAPI: }, "meta": {"trace_id": _trace_id(request)}, }) + application.include_router(auth_router) application.include_router(agent_runs_router) application.include_router(conversations_router) application.include_router(public_platform_router) diff --git a/app/service/auth_service.py b/app/service/auth_service.py new file mode 100644 index 0000000..e9ef4fe --- /dev/null +++ b/app/service/auth_service.py @@ -0,0 +1,232 @@ +"""账号密码登录：校验密码、签发访问令牌、留痕审计。 + +## 为什么放在平台侧 + +认证是**平台级能力**：所有业务域共用同一套 RBAC（`sys_user_role` → `sys_role_permission` +→ `sys_permission`），`docs/05` §11 把"JWT 签发、刷新、注销"划给统一身份认证模块。 +本模块只做**登录这一步**（账号密码换令牌）；刷新与注销留给后续迭代 —— +`app/core/security.py` 已经留好 `RevocationStore` 协议，接上 Redis 即可。 + +放在 Service 层而不是业务 Agent 里，是因为它不属于任何一个业务域：让业务分支自己加登录 +路由，等于又开一条绕过公共鉴权的路径（`AGENTS.md` 规则 7）。 + +## 令牌里为什么只放 `sub` + +`JwtAuthenticator.authenticate` 只从令牌取 `sub`（用户 id），角色/权限/数据范围由 +`IdentityService.resolve` **每次请求查库**解析（`identity_repository.load_context`： +`Fresh RBAC reads make revocation immediate`）。这是有意设计——权限变更立即生效、不受 +令牌有效期拖累。所以登录只要签一个含 `sub` 的 JWT，**现有鉴权链路一行都不用改**。 + +三个角色的区分（客户 / 员工 / 管理员）因此已经完备：`bootstrap.py` 里各 Agent 的 +`allowed_roles` 早就分开了（`CustomerServiceAgent` 只要 `customer`、`RiskAgent` 要 +`risk_operator`/`admin`、`PlatformProbeAgent` 只要 `admin`），此前唯独缺"怎么证明你是谁"。 + +## 安全约定（金融场景，逐条对应下面的实现） + +1. **不区分失败原因**。用户不存在、密码错、账号停用、密码未初始化 —— 对外**同一条** 401 + 消息。`docs/05` §3.6 只给了一个 `AUTHENTICATION_REQUIRED`，客户端本来也不该据 message + 区分。否则这个接口就成了账号枚举器。 +2. **防时序枚举**。用户不存在时**照样跑一次 bcrypt 比对**（`_DUMMY_HASH`）。否则 + "查无此人"会明显快于"密码错"，同样能枚举出哪些账号存在。 +3. **成功与失败都审计**。金融场景必须能回答"谁、什么时候、从哪、试图登录哪个账号、成没成"。 + `interaction_audit.actor_id` 可空，正是为失败场景准备的。 +4. **绝不记录密码**。`detail` 里只有用户名与失败原因，没有任何形式的 password 字段。 +5. **密码哈希用 bcrypt**。`cryptography` 是给 JWT（RS256）用的，它不提供密码哈希。 +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta +from functools import lru_cache +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import bcrypt +import jwt +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.config import get_settings +from app.core.contracts import RequestContext +from app.core.errors import UnauthorizedAgentError +from app.model.audit import InteractionAudit +from app.service.identity_service import IdentityService + +logger = logging.getLogger(__name__) + +#: 访问令牌有效期。与 `publish_*` 脚本长期使用的 30 分钟一致；权限不放在令牌里， +#: 所以这个值只影响"要不要重新登录"，不影响权限变更的生效速度。 +ACCESS_TOKEN_TTL_SECONDS = 1800 + +#: 对外统一的失败消息。刻意不区分原因，见模块文档第 1 条。 +INVALID_CREDENTIALS_MESSAGE = "用户名或密码不正确" + +#: 用户不存在时用来比对的固定哈希，见模块文档第 2 条。 +#: 用 `bcrypt.hashpw` 现算一次即可，不需要是"某个真实用户的密码"。 +_DUMMY_HASH = bcrypt.hashpw(b"not-a-real-password", bcrypt.gensalt()) + + +@lru_cache(maxsize=1) +def _private_key() -> str: + """签发私钥。只在本进程内缓存，不落任何地方、不进日志。""" + settings = get_settings() + path = Path(settings.jwt_private_key_path) + if not path.is_absolute(): + path = Path.cwd() / path + return path.read_text(encoding="utf-8") + + +def issue_access_token(user_id: int) -> tuple[str, int]: + """签一个只含 `sub` 的访问令牌，返回 (token, expires_in 秒)。 + + `security.py` 的 `options={"require": [...]}` 要求 + `sub/iss/aud/exp/nbf/jti` 齐全，缺任何一个都会被判非法令牌。 + """ + settings = get_settings() + now = datetime.now(UTC) + expires_in = ACCESS_TOKEN_TTL_SECONDS + token = jwt.encode( + { + "sub": str(user_id), + "iss": settings.jwt_issuer, + "aud": settings.jwt_audience, + "iat": now, + "nbf": now - timedelta(seconds=5), + "exp": now + timedelta(seconds=expires_in), + "jti": str(uuid4()), + }, + _private_key(), + algorithm=settings.jwt_algorithm, + ) + return token, expires_in + + +def verify_password(password: str, stored_hash: str | None) -> bool: + """常数时间的密码校验；任何异常都当校验失败。 + + `stored_hash` 在本项目的现状是**占位符**（种子写 `'x'`、worker 身份写 + `!worker-only-no-password-login!`），它们都不是合法 bcrypt 格式， + `bcrypt.checkpw` 会抛 `ValueError` —— 必须吞掉并返回 False， + 否则"没设过密码的账号"会变成 500 而不是 401。 + """ + if not stored_hash: + return False + try: + return bcrypt.checkpw(password.encode("utf-8"), stored_hash.encode("utf-8")) + except (ValueError, TypeError): + return False + + +class AuthService: + """登录入口。只依赖一个数据库会话，不持有请求上下文（登录时还没有身份）。""" + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def login(self, username: str, password: str, *, trace_id: str) -> dict[str, Any]: + """校验账号密码并签发令牌。 + + 失败一律抛 `UnauthorizedAgentError`（401 `AUTHENTICATION_REQUIRED`）， + 由 `app/main.py` 的 `AgentError` 处理器输出 `docs/05` §3.4 的统一错误信封。 + """ + row = ( + await self.session.execute( + text( + "SELECT id, username, password_hash, status " + "FROM sys_user WHERE username = :username LIMIT 1" + ), + {"username": username}, + ) + ).mappings().first() + + # 不存在时也跑一次 bcrypt，让"查无此人"与"密码错"的耗时一致（见模块文档第 2 条）。 + # 注意不能指望 `verify_password(password, None)` 代劳 —— 它对空哈希直接返回 False， + # 那就等于"查无此人"立刻返回，时序差异照样能用来枚举账号。 + stored_hash = str(row["password_hash"]) if row is not None else None + matched = verify_password(password, stored_hash) + + if row is None: + verify_password(password, _DUMMY_HASH.decode("utf-8")) + await self._audit( + actor_id=None, + action_type="auth.login_failed", + detail={"username": username, "reason": "user_not_found", "trace_id": trace_id}, + ) + raise UnauthorizedAgentError(INVALID_CREDENTIALS_MESSAGE) + + user_id = int(row["id"]) + if not matched: + await self._audit( + actor_id=user_id, + action_type="auth.login_failed", + detail={"username": username, "reason": "bad_password", "trace_id": trace_id}, + ) + raise UnauthorizedAgentError(INVALID_CREDENTIALS_MESSAGE) + + # 账号停用、角色读取失败等一律归到同一条 401：身份解析走的就是请求期那条链路， + # 保证"能登录"与"登录后能用"用的是同一套判断。 + try: + resolved = await self._resolve(user_id, trace_id) + except Exception as exc: + await self._audit( + actor_id=user_id, + action_type="auth.login_failed", + detail={ + "username": username, + "reason": f"identity_unavailable:{type(exc).__name__}", + "trace_id": trace_id, + }, + ) + raise UnauthorizedAgentError(INVALID_CREDENTIALS_MESSAGE) from exc + + token, expires_in = issue_access_token(user_id) + await self._audit( + actor_id=user_id, + action_type="auth.login_succeeded", + detail={ + "username": username, + "roles": list(resolved.roles), + "trace_id": trace_id, + }, + ) + return { + "access_token": token, + "token_type": "Bearer", + "expires_in": expires_in, + "user_id": str(user_id), + # 前端据此决定进哪个界面；**鉴权仍以库里实时数据为准**，不看这两个字段。 + "roles": list(resolved.roles), + "data_scope": resolved.data_scope, + } + + async def _resolve(self, user_id: int, trace_id: str) -> RequestContext: + """复用请求期的身份解析，保证登录与后续调用看到的是同一套 RBAC。""" + identity = RequestContext(user_id=str(user_id), trace_id=trace_id) + return await IdentityService().resolve(identity) + + async def _audit( + self, *, actor_id: int | None, action_type: str, detail: dict[str, Any] + ) -> None: + """登录审计。 + + 与业务写入分开提交：登录失败时**也要**留下记录，不能因为随后抛异常而被回滚掉。 + 审计写失败不阻断登录流程（只告警）——否则审计表的问题会变成"谁都登不进来"。 + """ + self.session.add( + InteractionAudit( + actor_type="user", + actor_id=actor_id, + portal=None, + session_id=None, + action_type=action_type, + detail=detail, + created_at=datetime.now(UTC).replace(tzinfo=None), + ) + ) + try: + await self.session.commit() + except Exception: + logger.warning("login audit write failed", exc_info=True) + await self.session.rollback() diff --git a/docs/05-接口文档.md b/docs/05-接口文档.md index f1973fc..516512c 100644 --- a/docs/05-接口文档.md +++ b/docs/05-接口文档.md @@ -970,6 +970,17 @@ GET /internal/metrics 这些接口不使用业务 JSON 信封，不暴露数据库地址、模型密钥、Token、完整客户资料或异常堆栈，只允许内网和监控系统访问。JWT 签发、刷新、注销由统一身份认证模块负责，Agent 平台不重复实现。 +> **实现现状（2026-09-11 更新）**：上面这句原本是"平台不做签发"的依据，实际落地时确认了 +> 平台**必须**有一个登录入口 —— 否则客户 / 员工 / 管理员三种身份无法区分（各 Agent 的 +> `allowed_roles` 早就分开了，缺的只是"怎么证明你是谁"）。因此平台现在提供 +> **`POST /api/v1/auth/tokens`**（账号密码换访问令牌，见 §19 的 A034）， +> 这是本文档 §11 那句的**唯一例外**。 +> +> 边界仍然守住：平台**只做登录**，**刷新与注销仍归统一身份认证模块** +> （`app/core/security.py` 已留好 `RevocationStore` 协议，接上 Redis 即可）。 +> 令牌里只放 `sub`，角色 / 权限 / 数据范围一律由 `IdentityService` 每次请求查库解析， +> 所以权限变更立即生效，不受令牌有效期影响。 + ## 16. 验收与契约测试 ### 16.1 HTTP 通用测试 @@ -1087,6 +1098,7 @@ GET /internal/metrics | A031 | `GET /api/v1/admin/negative-word-rules` | `config:read` | 否 | `200` | 否 | | A032 | `PUT /api/v1/admin/negative-word-rules/{rule_id}` | `config:write` | 必须 | `200` | 禁止表达 | | A033 | `GET /api/v1/admin/audit-records` | `audit:read` | 否 | `200` | 否 | +| A034 | `POST /api/v1/auth/tokens` | 公开（登录前无身份） | 否 | `200` | 登录成功/失败 | | O001 | `GET /internal/health/live` | 内网 | 否 | `200` | 否 | | O002 | `GET /internal/health/ready` | 内网 | 否 | `200/503` | 否 | | O003 | `GET /internal/metrics` | 监控系统 | 否 | `200` | 否 | diff --git a/pyproject.toml b/pyproject.toml index 1fecb58..61bb733 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ dependencies = [ "pymilvus>=2.5,<3", "PyJWT>=2.10,<3", "cryptography>=44,<51", + "bcrypt>=4.0,<5", "httpx>=0.28,<1", "tzdata>=2025.1,<2027", "alibabacloud_docmind_api20220711==1.4.14", diff --git a/requirements.txt b/requirements.txt index ee54e8b..147173e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -42,6 +42,11 @@ python-multipart>=0.0.20,<1 # 原先漏声明：别人的环境跑知识入库会直接 ModuleNotFoundError: No module named 'docx' python-docx>=1.1,<2 +# 登录（账号密码换令牌） +# - bcrypt：密码哈希。注意 `cryptography` 只用于 JWT(RS256)，它不提供密码哈希。 +# python 自带的 hashlib 也不适合存密码（无可调工作因子、无盐管理）。 +bcrypt>=4.0,<5 + # Development and test dependencies pytest>=8.3,<9 pytest-asyncio>=0.25,<1 diff --git a/tests/integration/test_auth_login_mysql.py b/tests/integration/test_auth_login_mysql.py new file mode 100644 index 0000000..21c7699 --- /dev/null +++ b/tests/integration/test_auth_login_mysql.py @@ -0,0 +1,201 @@ +"""登录接口的端到端验证（真实 MySQL + 真实 HTTP 栈）。 + +这里刻意**不用替身**：登录的价值就在于"签出来的令牌能不能真的用"， +用 mock 验证等于只测了自己写的桩。所以每个用例都走 `app.main.app` 的 ASGI 栈， +并且至少有一个用例拿令牌去调**另一个真实接口**。 + +前置：`python tools/seed_test_rbac.py`（用户与角色）与 +`python tools/set_user_password.py`（演示口令）。 + +覆盖的安全约定（与 `app/service/auth_service.py` 的模块文档一一对应）： +1. 三个角色各自能登录，且拿到的 `roles` 正确 —— 这正是"区分客户/员工/管理员"的落点； +2. 密码错与外挂账号**返回完全相同的响应**，接口不能当账号枚举器； +3. 从没设过密码的账号（占位符哈希）不能登录，且不能变成 500。 +""" + +from typing import Any + +import httpx +import pytest + +from app.api.dependencies.rate_limit import LOGIN_MAX_ATTEMPTS +from app.main import app + +pytestmark = pytest.mark.integration + +LOGIN_PATH = "/api/v1/auth/tokens" + +#: 演示账号（tools/set_user_password.py 设置）。 +DEMO_ACCOUNTS = ( + ("cust_t", "123456", "customer"), + ("risk_t", "666666", "risk_operator"), + ("admin_t", "88888888", "admin"), +) + +#: `sys_user.password_hash` 仍是占位符的账号（没设过密码，不该能登录）。 +PLACEHOLDER_ACCOUNTS = ("review_t", "offsite_worker") + + +def client() -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=30 + ) + + +class _AlwaysAllowBackend: + """恒放行：计数 1，远低于上限。""" + + async def increment(self, key: str, window_seconds: int) -> tuple[int, int] | None: + del key, window_seconds + return (1, 0) + + +class _AlwaysDenyBackend: + """恒超限：用来验证登录闸门确实会拦。""" + + async def increment(self, key: str, window_seconds: int) -> tuple[int, int] | None: + del key, window_seconds + return (LOGIN_MAX_ATTEMPTS + 1, 30) + + +@pytest.fixture(autouse=True) +def _replace_rate_limit_backend(monkeypatch: pytest.MonkeyPatch) -> None: + """把限流后端换成恒放行替身，只作用于本文件。 + + 为什么必须换：本文件所有用例加起来要发十几次登录请求，而登录闸门是 60 秒 10 次。 + 限流对所有请求生效（包括测试自己发的），Redis 里的计数还会**跨测试累积** —— + 于是后面的用例拿到 429 而不是想断言的 200/401。那是用例互相污染，不是产品缺陷。 + + `get_counter_backend` 正是为此留的替换点（见它的文档字符串："模块级函数是唯一的 + 替换点（测试注入替身，不连 Redis）"）。限流本身由下面那个用例单独验证， + 不会被这个替身掩盖掉。 + """ + monkeypatch.setattr( + "app.api.dependencies.rate_limit.get_counter_backend", + lambda: _AlwaysAllowBackend(), + ) + + +@pytest.mark.asyncio +async def test_login_is_actually_rate_limited(monkeypatch: pytest.MonkeyPatch) -> None: + """登录闸门必须真的会拦 —— 它是密码爆破的唯一防线。 + + 用一个恒超限的替身后端验证"接了闸门且会抛 429"，与上面那些替身用例互补： + 那些证明认证逻辑对，这个证明防线在。 + """ + monkeypatch.setattr( + "app.api.dependencies.rate_limit.get_counter_backend", + lambda: _AlwaysDenyBackend(), + ) + async with client() as http: + response = await login(http, "cust_t", "123456") + + assert response.status_code == 429 + assert response.json()["error"]["code"] == "RATE_LIMITED" + assert response.json()["error"]["retryable"] is True + + +async def login( + http: httpx.AsyncClient, username: str, password: str +) -> httpx.Response: + return await http.post(LOGIN_PATH, json={"username": username, "password": password}) + + +@pytest.mark.parametrize(("username", "password", "expected_role"), DEMO_ACCOUNTS) +@pytest.mark.asyncio +async def test_each_role_can_login_with_its_own_role( + username: str, password: str, expected_role: str +) -> None: + """客户、员工、管理员各自登录，拿到的 `roles` 就是区分三种登录的落点。""" + async with client() as http: + response = await login(http, username, password) + + assert response.status_code == 200, response.text + body: dict[str, Any] = response.json() + # docs/05 §3.3：业务字段全在 data 里，meta 只有 trace_id。 + assert set(body) == {"data", "meta"} + assert set(body["meta"]) == {"trace_id"} + data = body["data"] + assert data["token_type"] == "Bearer" + assert data["expires_in"] == 1800 + assert expected_role in data["roles"], f"{username} 的角色里没有 {expected_role}" + assert data["access_token"] + + +@pytest.mark.asyncio +async def test_issued_token_actually_works_on_a_real_endpoint() -> None: + """签出来的令牌必须能真的用 —— 这是本文件不用替身的理由。 + + `GET /api/v1/users/me/memory-profile` 需要 `memory:read:self`（客户角色有）， + 走的是 `build_request_context` → `JwtAuthenticator` → `IdentityService.resolve` + 这条真实链路：令牌只带 `sub`，角色与权限全部查库解析。 + """ + async with client() as http: + response = await login(http, "cust_t", "123456") + assert response.status_code == 200, response.text + token = response.json()["data"]["access_token"] + + authorized = await http.get( + "/api/v1/users/me/memory-profile", + headers={"Authorization": f"Bearer {token}"}, + ) + # 200=有画像，404=该客户还没有画像行；两者都说明**令牌被接受并通过了 RBAC**。 + # 401/403 则说明令牌或身份解析链有问题。 + assert authorized.status_code in (200, 404), authorized.text + + +@pytest.mark.asyncio +async def test_missing_and_malformed_token_are_rejected() -> None: + async with client() as http: + missing = await http.get("/api/v1/users/me/memory-profile") + malformed = await http.get( + "/api/v1/users/me/memory-profile", + headers={"Authorization": "Bearer not-a-jwt"}, + ) + + assert missing.status_code == 401 + assert malformed.status_code == 401 + # auth.py 的约定：令牌缺失/非法/吊销不区分，都不泄露内部原因。 + assert missing.json()["error"]["code"] == "AUTHENTICATION_REQUIRED" + assert malformed.json()["error"]["code"] == "AUTHENTICATION_REQUIRED" + + +@pytest.mark.asyncio +async def test_wrong_password_and_unknown_user_are_indistinguishable() -> None: + """接口不能当账号枚举器：两种失败的**状态码与消息**必须完全一致。""" + async with client() as http: + wrong_password = await login(http, "cust_t", "definitely-wrong") + unknown_user = await login(http, "no-such-user-at-all", "whatever") + + assert wrong_password.status_code == 401 + assert unknown_user.status_code == 401 + assert wrong_password.json()["error"]["message"] == unknown_user.json()["error"]["message"] + assert wrong_password.json()["error"]["code"] == unknown_user.json()["error"]["code"] + # 也不该回显是哪个字段错了。 + assert wrong_password.json()["error"]["field_errors"] == [] + + +@pytest.mark.parametrize("username", PLACEHOLDER_ACCOUNTS) +@pytest.mark.asyncio +async def test_account_without_real_password_cannot_login(username: str) -> None: + """没设过密码的账号（`password_hash` 是占位符）必须 401，而不是 500。 + + `'x'` 与 `!worker-only-no-password-login!` 都不是合法 bcrypt 格式， + `bcrypt.checkpw` 会抛 `ValueError` —— `verify_password` 吞掉它并返回 False。 + """ + async with client() as http: + response = await login(http, username, "123456") + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_extra_fields_in_login_body_are_rejected() -> None: + """`extra="forbid"`：调用方不能借登录接口塞身份字段。""" + async with client() as http: + response = await http.post( + LOGIN_PATH, + json={"username": "cust_t", "password": "123456", "roles": ["admin"]}, + ) + + assert response.status_code == 422 diff --git a/tools/set_user_password.py b/tools/set_user_password.py new file mode 100644 index 0000000..9525428 --- /dev/null +++ b/tools/set_user_password.py @@ -0,0 +1,127 @@ +"""设置用户登录密码（bcrypt 哈希写入 `sys_user.password_hash`）。 + +## ⚠️ 仅限演示环境 + +本脚本把演示口令写在源码里、也允许明文命令行传入，目的是让演示与联调**当天可用**。 +`123456` / `666666` / `88888888` 这类弱口令**在生产环境等于没有密码**： +上线前必须全部更换，并由运维走单独的改密流程（本脚本只服务演示）。 + +## 为什么需要它 + +`sys_user.password_hash` 此前**全是占位符** —— 种子写 `'x'`、worker 身份写 +`!worker-only-no-password-login!`，即"这个字段从来没过真实密码"。登录接口上线后， +不设密码就没人能登进来；这个脚本补的正是这一步。 + +## 用法 + + python tools/set_user_password.py --list # 只列现状，不改任何数据 + python tools/set_user_password.py # 按内置演示规则设置 + python tools/set_user_password.py --user 9002 --password 'xxx' + +注意：bcrypt 每次加盐不同，**重复执行等于重设密码**（不是"已存在就跳过"）。这是有意的 +——改密本来就该覆盖，但要清楚它不是幂等操作。 +""" + +from __future__ import annotations + +import argparse +import asyncio +import sys +from datetime import UTC, datetime + +import bcrypt +from sqlalchemy import text + +from app.infrastructure.db import SessionFactory + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] + +#: 演示口令（用户指定）。键是 `sys_user.id`。 +#: 9001 客户 / 9002 员工（风控专员）/ 9003 管理员。 +DEMO_PASSWORDS: dict[str, str] = { + "9001": "123456", + "9002": "666666", + "9003": "88888888", +} + +#: 与 `AuthService.verify_password` 保持一致的识别方式：只有 bcrypt 格式才算"已设真密码"。 +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) + + +async def list_users() -> None: + async with SessionFactory() as session: + rows = ( + await session.execute( + text( + "SELECT id, username, user_type, status, password_hash " + "FROM sys_user ORDER BY id" + ) + ) + ).mappings().all() + print(f"{'id':<8}{'username':<18}{'user_type':<12}{'status':<8}密码状态") + for row in rows: + state = "已设（bcrypt）" if _is_real_hash(row["password_hash"]) else "占位符，无法登录" + print( + f"{row['id']:<8}{str(row['username']):<18}{str(row['user_type']):<12}" + f"{str(row['status']):<8}{state}" + ) + + +async def set_password(user_id: str, password: str) -> int: + now = datetime.now(UTC).replace(tzinfo=None) + async with SessionFactory() as session, session.begin(): + result = await session.execute( + text( + "UPDATE sys_user SET password_hash = :hash, updated_at = :now " + "WHERE id = :user_id" + ), + {"hash": hash_password(password), "now": now, "user_id": int(user_id)}, + ) + if result.rowcount == 0: + print(f"[失败] sys_user 里没有 id={user_id} 的用户") + return 1 + print(f"[OK] id={user_id} 密码已设置（bcrypt）") + return 0 + + +async def main() -> int: + parser = argparse.ArgumentParser(description="设置用户登录密码（bcrypt）") + parser.add_argument("--list", action="store_true", help="只列现状，不改数据") + parser.add_argument("--user", help="单个用户 id（配合 --password 使用）") + parser.add_argument("--password", help="要设置的明文密码") + args = parser.parse_args() + + if args.list: + await list_users() + return 0 + + if args.user or args.password: + if not (args.user and args.password): + print("[失败] --user 与 --password 必须成对给出") + return 1 + return await set_password(args.user, args.password) + + print("按内置演示规则设置密码（生产环境必须更换）：") + failures = 0 + for user_id, password in DEMO_PASSWORDS.items(): + print(f" · id={user_id} → {len(password)} 位口令") + failures += await set_password(user_id, password) + print("\n设置后的现状：") + await list_users() + if failures: + return 1 + print("\n可以登录了。接口：POST /api/v1/auth/tokens {\"username\": \"\", \"password\": \"...\"}") + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()))