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/controllers/rbac.py b/app/api/controllers/rbac.py new file mode 100644 index 0000000..275b7d7 --- /dev/null +++ b/app/api/controllers/rbac.py @@ -0,0 +1,84 @@ +"""角色与权限的只读接口（B1：先让管理员看得见）。 + +前缀与 `admin.py` 相同（`/api/v1/admin`），但**独立成文件**：`admin.py` 是表驱动的 +配置面 CRUD 工厂（`RESOURCES` 单表增删改），而角色/权限是**多对多关系查询**， +塞进那个工厂既不自然、也会让"配置管理"和"身份管理"两件事混在一个文件里。 + +四个接口全部只读，且**不写审计**——它们返回的就是审计材料本身（"谁能访问什么"， +合规检查要看的正是这份清单）。`docs/05` §19 登记时"审计"列标"否"。 + +所有响应都走 `docs/05` §3.3 的统一信封（复用 `app/api/views/envelope.py`）： +列表的 `data` 是纯数组、分页元数据在 `meta`；单对象的 `meta` 只有 `trace_id`。 +""" + +from fastapi import APIRouter, Depends, Path +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.dependencies.auth import build_request_context +from app.api.dependencies.database import get_session +from app.api.dependencies.rate_limit import enforce_rate_limit +from app.api.views.envelope import envelope, list_envelope +from app.core.contracts import RequestContext +from app.service.rbac_query_service import RbacQueryService + +router = APIRouter( + prefix="/api/v1/admin", + tags=["platform-admin"], + dependencies=[Depends(enforce_rate_limit)], +) + +#: 角色码在 `sys_role` 里形如 `customer` / `risk_operator` / `admin`。 +ROLE_CODE = Path(min_length=1, max_length=64, pattern=r"^[a-z0-9_]+$") +#: `sys_user.id` 是 BIGINT UNSIGNED。 +USER_ID = Path(pattern=r"^[0-9]{1,20}$") + + +@router.get("/roles") +async def list_roles( + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """所有角色 + 每个角色的权限数与在用人数。""" + page = await RbacQueryService(session).list_roles(context) + return list_envelope(page, context) + + +@router.get("/roles/{role_code}") +async def get_role( + role_code: str = ROLE_CODE, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """单个角色的详情。 + + 与 `/permissions` 分开是必要的：权限为空的角色（例如刚建好还没授权的）在一个合并 + 接口里很容易被"查不到权限"误判成"角色不存在"。 + """ + data = await RbacQueryService(session).get_role_detail(context, role_code) + return envelope(data, context) + + +@router.get("/roles/{role_code}/permissions") +async def list_role_permissions( + role_code: str = ROLE_CODE, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """该角色的权限清单，按权限码排序，便于与代码里各 Service 的 `require(...)` 对照。""" + page = await RbacQueryService(session).list_role_permissions(context, role_code) + return list_envelope(page, context) + + +@router.get("/users/{user_id}/roles") +async def get_user_roles( + user_id: str = USER_ID, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """某个用户**实际解析出来**的角色 / 权限 / 数据范围 / 可见客户。 + + 这是四个接口里最有用的一个：它直接回答"这个人为什么 403"—— + 走的是 `IdentityService.resolve`，与请求期完全同一条链路。 + """ + data = await RbacQueryService(session).get_user_identity(context, user_id) + return envelope(data, context) 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 ad874ee..76edecf 100644 --- a/app/main.py +++ b/app/main.py @@ -9,6 +9,7 @@ from fastapi.staticfiles import StaticFiles 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.asset_allocation import router as asset_allocation_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.investment_goals import router as investment_goals_router @@ -20,6 +21,7 @@ from app.api.controllers.onboarding import router as onboarding_router from app.api.controllers.portfolio_analysis import router as portfolio_analysis_router from app.api.controllers.promotion_material import router as promotion_material_router from app.api.controllers.public_platform import router as public_platform_router +from app.api.controllers.rbac import router as rbac_router from app.api.controllers.recommendations import ( admin_router as recommendation_admin_router, ) @@ -114,9 +116,11 @@ 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) + application.include_router(rbac_router) application.include_router(risk_router) application.include_router(visitor_tokens_router) application.include_router(offsite_fund_router) diff --git a/app/service/auth_service.py b/app/service/auth_service.py new file mode 100644 index 0000000..185bbaa --- /dev/null +++ b/app/service/auth_service.py @@ -0,0 +1,242 @@ +"""账号密码登录：校验密码、签发访问令牌、留痕审计。 + +## 为什么放在平台侧 + +认证是**平台级能力**：所有业务域共用同一套 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 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: + """常数时间的密码校验；任何异常都当校验失败。 + + `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/app/service/rbac_query_service.py b/app/service/rbac_query_service.py new file mode 100644 index 0000000..c03adb0 --- /dev/null +++ b/app/service/rbac_query_service.py @@ -0,0 +1,202 @@ +"""角色与权限的只读查询（B1：先让管理员**看得见**）。 + +## 为什么先做只读 + +权限变更（提权 / 降权）必须带三条红线：**审计留痕**、**禁止自我提权**、 +**保护内置角色**（否则可能把自己锁在门外）。那是一条需要单独评审的写路径。 +而运维的大半诉求其实是"这个角色到底有哪些权限""这个人为什么 403"——只读就能回答。 + +## 权限码为什么复用 `audit:read`，而不新增 `rbac:read` + +1. `audit:read` 的 `data_scope` 已经是 `all`，且只发给管理员； +2. "谁能访问什么"本身就是**审计材料**——合规检查要看的正是这份清单； +3. 复用是**零数据改动、立刻可用**：新增权限码要先往 `sys_permission` 插行再授权， + 而 `sys_role_permission` 目前由 `seed_test_rbac.py` 以 **DELETE 重建**语义管理 + （见该脚本 95-98 行），为一个只读接口去动权限表不划算。 + 将来要细分时再加 `rbac:read`，与现在不冲突。 + +## 为什么 `user_identity()` 复用 `IdentityService.resolve` + +那是请求进来时走的**同一条链路**：含 `status` 检查、`assigned_at` / `expires_at` +时间窗、`data_scope` 取最高、客户分配（`identity_repository.load_context`）。 +自己再拼一遍 SQL 必然与它漂移，而"这里查出来的权限"与"实际能用的权限"不一致， +比没有这个接口更糟——排障时会被引到错的方向。 +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.contracts import RequestContext +from app.core.errors import GenericResourceNotFoundError, UnauthorizedAgentError +from app.service.authorization_service import AuthorizationService +from app.service.identity_service import IdentityService + +#: 这三个接口共用的权限闸门，理由见模块文档。 +READ_PERMISSION = "audit:read" + + +class RbacQueryService: + """RBAC 只读查询。不写任何表，也不改任何状态。""" + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def list_roles(self, context: RequestContext) -> dict[str, Any]: + """列出所有角色，带权限数与在用人数。 + + 角色数量是个位数量级，不分页；返回 `{items, next_cursor, has_more}` 是为了 + 与其它列表接口共用 `list_envelope`（`docs/05` §3.3），而不是真会翻页。 + """ + await AuthorizationService.require(context, READ_PERMISSION) + rows = ( + await self.session.execute( + text( + """ + SELECT r.id, r.role_code, r.role_name, r.status, + (SELECT COUNT(*) FROM sys_role_permission rp + WHERE rp.role_id = r.id) AS permission_count, + (SELECT COUNT(*) FROM sys_user_role ur + WHERE ur.role_id = r.id) AS user_count + FROM sys_role r + ORDER BY r.id + """ + ) + ) + ).mappings().all() + return { + "items": [ + { + "role_id": str(row["id"]), + "role_code": str(row["role_code"]), + "role_name": str(row["role_name"]), + "status": str(row["status"]), + "permission_count": int(row["permission_count"]), + "user_count": int(row["user_count"]), + } + for row in rows + ], + "next_cursor": None, + "has_more": False, + } + + async def list_role_permissions( + self, context: RequestContext, role_code: str + ) -> dict[str, Any]: + """某个角色拥有的全部权限（按权限码排序，便于与代码里 `require(...)` 对照）。""" + await AuthorizationService.require(context, READ_PERMISSION) + role = ( + await self.session.execute( + text( + "SELECT id, role_code, role_name, status FROM sys_role " + "WHERE role_code = :code" + ), + {"code": role_code}, + ) + ).mappings().first() + if role is None: + raise GenericResourceNotFoundError("角色不存在") + rows = ( + await self.session.execute( + text( + """ + SELECT p.permission_code, p.resource, p.action, p.data_scope + FROM sys_role_permission rp + JOIN sys_permission p ON p.id = rp.permission_id + WHERE rp.role_id = :role_id + ORDER BY p.permission_code + """ + ), + {"role_id": int(role["id"])}, + ) + ).mappings().all() + return { + "items": [ + { + "permission_code": str(row["permission_code"]), + "resource": str(row["resource"]), + "action": str(row["action"]), + "data_scope": str(row["data_scope"]), + } + for row in rows + ], + "next_cursor": None, + "has_more": False, + } + + async def get_role_detail( + self, context: RequestContext, role_code: str + ) -> dict[str, Any]: + """角色本身的信息（与权限清单分开：详情接口不应因权限为空就 404）。""" + await AuthorizationService.require(context, READ_PERMISSION) + row = ( + await self.session.execute( + text( + """ + SELECT r.id, r.role_code, r.role_name, r.status, r.created_at, r.updated_at, + (SELECT COUNT(*) FROM sys_role_permission rp + WHERE rp.role_id = r.id) AS permission_count, + (SELECT COUNT(*) FROM sys_user_role ur + WHERE ur.role_id = r.id) AS user_count + FROM sys_role r WHERE r.role_code = :code + """ + ), + {"code": role_code}, + ) + ).mappings().first() + if row is None: + raise GenericResourceNotFoundError("角色不存在") + return { + "role_id": str(row["id"]), + "role_code": str(row["role_code"]), + "role_name": str(row["role_name"]), + "status": str(row["status"]), + "permission_count": int(row["permission_count"]), + "user_count": int(row["user_count"]), + } + + async def get_user_identity( + self, context: RequestContext, user_id: str + ) -> dict[str, Any]: + """某个用户**实际解析出来**的身份：角色、权限、数据范围、可见客户。 + + 直接复用 `IdentityService.resolve`，见模块文档。它抛 `UnauthorizedAgentError` + 表示"账号不存在或未启用"，对这个只读接口来说是**查不到**（404）， + 不是调用方鉴权失败——所以在这里转成 404，避免排障时误以为是权限问题。 + """ + await AuthorizationService.require(context, READ_PERMISSION) + row = ( + await self.session.execute( + text( + "SELECT id, user_no, username, user_type, status " + "FROM sys_user WHERE id = :user_id" + ), + {"user_id": int(user_id)}, + ) + ).mappings().first() + if row is None: + raise GenericResourceNotFoundError("用户不存在") + + try: + resolved = await IdentityService().resolve( + RequestContext(user_id=user_id, trace_id=context.trace_id) + ) + except UnauthorizedAgentError: + # 账号被停用（status != '正常'）——用户存在但拿不到任何权限， + # 如实返回空权限集，比 404 更有助于排障。 + resolved = None + + return { + "user_id": str(row["id"]), + "user_no": str(row["user_no"]), + "username": str(row["username"]), + "user_type": str(row["user_type"]), + "status": str(row["status"]), + "roles": list(resolved.roles) if resolved is not None else [], + "permissions": list(resolved.permissions) if resolved is not None else [], + "data_scope": resolved.data_scope if resolved is not None else None, + "customer_ids": list(resolved.customer_ids) if resolved is not None else [], + } diff --git a/docs/05-接口文档.md b/docs/05-接口文档.md index aac6010..ade4de4 100644 --- a/docs/05-接口文档.md +++ b/docs/05-接口文档.md @@ -1002,6 +1002,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 通用测试 @@ -1123,10 +1134,26 @@ 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` | 登录成功/失败 | +| A035 | `GET /api/v1/admin/roles` | `audit:read` | 否 | `200` | 否 | +| A036 | `GET /api/v1/admin/roles/{role_code}` | `audit:read` | 否 | `200` | 否 | +| A037 | `GET /api/v1/admin/roles/{role_code}/permissions` | `audit:read` | 否 | `200` | 否 | +| A038 | `GET /api/v1/admin/users/{user_id}/roles` | `audit:read` | 否 | `200` | 否 | | O001 | `GET /internal/health/live` | 内网 | 否 | `200` | 否 | | O002 | `GET /internal/health/ready` | 内网 | 否 | `200/503` | 否 | | O003 | `GET /internal/metrics` | 监控系统 | 否 | `200` | 否 | +> **A034 – A038 的两点说明**： +> +> - `POST /api/v1/auth/tokens` 是平台内**唯一的登录入口**（§11 已注明这是"平台不重复实现 +> 签发"的唯一例外；**刷新与注销仍归统一身份认证模块**）。 +> - A035 – A038 是 RBAC 的**只读**查询，供管理员回答"谁能访问什么""这个人为什么 403"。 +> 它们复用 `audit:read` 而**不新增** `rbac:read`：这份清单本身就是审计材料，且复用是 +> 零数据改动、立刻可用（新增权限码得先改 `sys_permission`，而它目前由 +> `seed_test_rbac.py` 以 DELETE 重建语义管理）。 +> **权限变更（提权 / 降权）尚无接口** —— 那条写路径必须带三条红线 +> （审计留痕、禁止自我提权、保护内置角色），需要单独评审，不是遗漏。 + 业务域接口 `/customer-service/handover-tickets/**`、`/advisory-plans/**`、`/sim-orders/**`、`/risk-scans/**` 和 `/risk-alerts/**` 的具体方法、请求体、领域状态机和错误码分别由对应业务文档登记；它们仍必须遵守本文第 3-5、11 和 12 节。 ## 20. 变更流程 diff --git a/docs/29-Agent组员登录接口使用说明.md b/docs/29-Agent组员登录接口使用说明.md new file mode 100644 index 0000000..c44b8e2 --- /dev/null +++ b/docs/29-Agent组员登录接口使用说明.md @@ -0,0 +1,274 @@ +# 登录接口使用说明（给 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. 怎么加一个测试人员 + +### ⚠️ 第 0 步：先确认**你们自己那台**有角色和密码 + +**环境数据不随代码合并**——`config_release`、`sys_role`、`sys_user` 都是各环境自己的。 +换句话说：**别人机器上能登录的账号，你们那台不一定有**。先跑这两条（都幂等，可重复执行）： + +```powershell +python tools/seed_test_rbac.py # 建三个角色（customer/risk_operator/admin）与它们的权限 +python tools/set_user_password.py # 给 9001/9002/9003 设演示密码 +``` + +然后确认角色确实绑上了（最后一列不该是"(无角色)"）： + +```powershell +python tools/create_test_user.py --list +``` + +``` +id username user_type status roles 密码 +9001 cust_t customer 正常 customer 已设 +9002 risk_t employee 正常 risk_operator 已设 +9003 admin_t employee 正常 admin 已设 +``` + +`create_test_user.py` 也会替你检查：如果 `sys_role` 里没有对应角色，它会直接告诉你 +"先跑 tools/seed_test_rbac.py"，而不是建出一个登不进任何地方的账号。 + +### 一条命令 + +```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/docs/evidence/auth-state.json b/docs/evidence/auth-state.json new file mode 100644 index 0000000..ac35093 --- /dev/null +++ b/docs/evidence/auth-state.json @@ -0,0 +1,27 @@ +{ + "user_count": 5, + "by_status": { + "正常": 4, + "禁用": 1 + }, + "by_user_type": { + "employee": 3, + "customer": 1, + "员工": 1 + }, + "password_hash_shape": { + "length_distribution": { + "1": 4, + "31": 1 + }, + "prefix_distribution": { + "<未知格式，首字符 'x'>": 4, + "<未知格式，首字符 '!'>": 1 + } + }, + "has_real_password_hash": false, + "placeholder_examples": [ + "!worker-only-no-password-login!", + "x" + ] +} \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 085ab3b..789ea59 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,7 @@ dependencies = [ "milvus-lite>=3.2,<4", "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 acfb07c..defd557 100644 --- a/requirements.txt +++ b/requirements.txt @@ -43,6 +43,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/tests/integration/test_rbac_read_mysql.py b/tests/integration/test_rbac_read_mysql.py new file mode 100644 index 0000000..66fe99d --- /dev/null +++ b/tests/integration/test_rbac_read_mysql.py @@ -0,0 +1,184 @@ +"""RBAC 只读接口的端到端验证（真实 MySQL + 真实 HTTP 栈）。 + +重点不在"能不能查出数据"，而在三件事： + +1. **门槛对不对** —— 这三个接口暴露的是"谁能访问什么"，属管理员级只读， + 客户令牌必须 403。权限码复用 `audit:read`（理由见 `rbac_query_service` 的模块文档）。 +2. **两个接口说的是不是同一件事** —— `/admin/users/{id}/roles` 走 + `IdentityService.resolve`，而登录响应的 `roles` 也走同一条链路；两者必须一致， + 否则排障时会被引到错方向。 +3. **信封形状** —— 列表的 `data` 是纯数组、分页元数据在 `meta`（`docs/05` §3.3）。 + +前置：`tools/seed_test_rbac.py` + `tools/set_user_password.py`。 +""" + +from typing import Any + +import httpx +import pytest + +from app.main import app + +pytestmark = pytest.mark.integration + +LOGIN_PATH = "/api/v1/auth/tokens" + + +class _AlwaysAllowBackend: + """恒放行，隔离跨用例的限流计数累积（同 test_auth_login_mysql.py）。""" + + async def increment(self, key: str, window_seconds: int) -> tuple[int, int] | None: + del key, window_seconds + return (1, 0) + + +@pytest.fixture(autouse=True) +def _replace_rate_limit_backend(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "app.api.dependencies.rate_limit.get_counter_backend", + lambda: _AlwaysAllowBackend(), + ) + + +def client() -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=30 + ) + + +async def token_for(http: httpx.AsyncClient, username: str, password: str) -> str: + response = await http.post( + LOGIN_PATH, json={"username": username, "password": password} + ) + assert response.status_code == 200, response.text + return str(response.json()["data"]["access_token"]) + + +def auth(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +@pytest.mark.asyncio +async def test_admin_can_list_roles_with_shape() -> None: + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + response = await http.get("/api/v1/admin/roles", headers=auth(token)) + + assert response.status_code == 200, response.text + body: dict[str, Any] = response.json() + # §3.3：列表的 data 是纯数组，分页元数据在 meta。 + assert isinstance(body["data"], list) + assert set(body["meta"]) == {"trace_id", "next_cursor", "has_more"} + codes = {row["role_code"] for row in body["data"]} + assert {"customer", "risk_operator", "admin"} <= codes, f"角色清单缺内置角色：{codes}" + for row in body["data"]: + assert isinstance(row["permission_count"], int) + assert isinstance(row["user_count"], int) + assert row["status"] + + +@pytest.mark.asyncio +async def test_role_permissions_are_listed_and_sorted() -> None: + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + response = await http.get( + "/api/v1/admin/roles/customer/permissions", headers=auth(token) + ) + + assert response.status_code == 200, response.text + body = response.json() + assert isinstance(body["data"], list) and body["data"], "客户角色不该没有任何权限" + codes = [row["permission_code"] for row in body["data"]] + assert codes == sorted(codes), "权限清单应按权限码排序，便于与代码里的 require() 对照" + assert {"agent:run", "suitability:read"} <= set(codes) + + +@pytest.mark.asyncio +async def test_role_detail_is_separate_from_permissions() -> None: + """权限为空的角色也必须能查到详情，不能被当成"角色不存在"。""" + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + detail = await http.get("/api/v1/admin/roles/operator", headers=auth(token)) + + assert detail.status_code == 200, detail.text + assert detail.json()["data"]["role_code"] == "operator" + + +@pytest.mark.asyncio +async def test_customer_token_is_denied() -> None: + """门槛验证：客户不能读"谁能访问什么"。""" + async with client() as http: + token = await token_for(http, "cust_t", "123456") + for path in ( + "/api/v1/admin/roles", + "/api/v1/admin/roles/customer/permissions", + "/api/v1/admin/users/9001/roles", + ): + response = await http.get(path, headers=auth(token)) + assert response.status_code == 403, f"{path} 不该让客户访问（{response.status_code}）" + assert response.json()["error"]["code"] == "AGENT_PERMISSION_DENIED" + + +@pytest.mark.asyncio +async def test_anonymous_is_unauthorized() -> None: + async with client() as http: + response = await http.get("/api/v1/admin/roles") + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "AUTHENTICATION_REQUIRED" + + +@pytest.mark.asyncio +async def test_unknown_role_and_user_are_not_found() -> None: + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + role = await http.get("/api/v1/admin/roles/no_such_role", headers=auth(token)) + user = await http.get("/api/v1/admin/users/99999999/roles", headers=auth(token)) + + assert role.status_code == 404 + assert user.status_code == 404 + + +@pytest.mark.asyncio +async def test_user_identity_agrees_with_login_response() -> None: + """交叉验证：两个接口必须说同一件事。 + + 登录响应的 `roles` 与 `/admin/users/{id}/roles` 的 `roles` 都来自 + `IdentityService.resolve`；若哪天有人给其中一条路径加了缓存或另写一份 SQL， + 这个断言会立刻发现。 + """ + async with client() as http: + login = await http.post( + LOGIN_PATH, json={"username": "risk_t", "password": "666666"} + ) + assert login.status_code == 200, login.text + login_data = login.json()["data"] + + admin_token = await token_for(http, "admin_t", "88888888") + identity = await http.get( + f"/api/v1/admin/users/{login_data['user_id']}/roles", headers=auth(admin_token) + ) + + assert identity.status_code == 200, identity.text + data = identity.json()["data"] + assert data["roles"] == login_data["roles"], "两个接口解析出的角色不一致" + assert data["data_scope"] == login_data["data_scope"], "两个接口解析出的数据范围不一致" + assert data["username"] == "risk_t" + assert "audit:read" in data["permissions"] + + +@pytest.mark.asyncio +async def test_deactivated_account_reports_empty_permissions_not_404() -> None: + """被停用的账号：**存在**但没有权限。返回空权限集比 404 更有助于排障。""" + async with client() as http: + admin_token = await token_for(http, "admin_t", "88888888") + # 9004(review_t) 是种子里的账号：仅绑了角色但没设密码，且此处不依赖密码。 + response = await http.get( + "/api/v1/admin/users/9004/roles", headers=auth(admin_token) + ) + + assert response.status_code == 200, response.text + data = response.json()["data"] + assert data["username"] == "review_t" + # 它没有 sys_user_role 绑定，因此 roles 为空 —— 但不该是 404。 + assert data["roles"] == [] diff --git a/tools/create_test_user.py b/tools/create_test_user.py new file mode 100644 index 0000000..4dd99f4 --- /dev/null +++ b/tools/create_test_user.py @@ -0,0 +1,216 @@ +"""添加一个可登录的测试账号（用户 + 角色 + 密码），并**验证它真的能拿到权限**。 + +## 为什么需要它 + +`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, + }, + ) + # 角色绑定先清后插。表上其实**有** `uk_sys_user_role (user_id, role_id)` 唯一键 + # （`tools/generate_baseline_sql.py:128`），所以直接 INSERT 不会堆重复行 —— + # 先清的理由是**支持改角色**：不清的话换角色会在表里留下两条绑定， + # `IdentityService` 解析出的 roles 就变成两个（既是 customer 又是 admin）。 + 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/login_console.py b/tools/login_console.py new file mode 100644 index 0000000..347959c --- /dev/null +++ b/tools/login_console.py @@ -0,0 +1,267 @@ +"""登录测试台：一个记事本级别的前端，用来在浏览器里验证登录接口。 + +浏览器打开 **http://127.0.0.1:8099**（可用 `--port` 改）。 + +## 它做什么 / 不做什么 + +- **走真实登录接口**（`POST /api/v1/auth/tokens`）拿令牌，和前端将来要做的一模一样。 + 与 `tools/chat_console.py` 不同：那个是当初还没有登录接口时自己签令牌的权宜做法， + 这个测的就是真链路。 +- 拿到令牌后可以按角色调几个**只读**接口，直接把状态码与响应 JSON 摊在页面上 —— + 方便确认"登录给的 roles 与接口实际放行的权限是否一致"。 +- **不动 `app/` 一个字节**：内层用 `create_app()`，本进程只做两件事 —— 提供一个静态页面、 + 把 `/api/*` 同源转发给它。同源转发是为了避开 CORS，也让浏览器不必直连主服务。 +- 只监听 `127.0.0.1`，不出本机。私钥始终在服务端（登录是后端做的），页面只拿到令牌。 + +## 为什么现在可以做这个了 + +`docs/24` 当初拒绝"给底座加 dev token 端点"，理由是那种端点等于把**任意身份**开放给 +任何能访问服务的人。现在不同了：登录接口要**校验密码**，所以浏览器拿令牌这件事 +不再等于"谁都能冒充任何人"。这条顾虑已经消除。 + +用法： + + python tools/login_console.py # 默认 8099 + python tools/login_console.py --port 9000 +""" + +from __future__ import annotations + +import argparse + +import httpx +import uvicorn +from fastapi import FastAPI, Request +from fastapi.responses import HTMLResponse, Response + +from app.main import create_app + +PAGE = """ + + + + +登录测试台 + + + +

登录测试台

+
+ 走真实接口 POST /api/v1/auth/tokens。密码仅用于本地演示。 +
+ +
+
+ +
+
+ +
+
+ + + 快捷填充： + + + +
+
+ +
+
当前身份
+
未登录
+
+ +
+
+ 用这个令牌调接口 + （下面按钮按角色给，方便核对权限） +
+
+ 
（还没有请求）
+
+ + + + +""" + + +def build_console() -> FastAPI: + base_app = create_app() + console = FastAPI(title="登录测试台", docs_url=None, redoc_url=None) + + @console.get("/", response_class=HTMLResponse) + async def index() -> HTMLResponse: + return HTMLResponse(PAGE) + + @console.api_route( + "/api/{path:path}", + methods=["GET", "POST", "PUT", "PATCH", "DELETE"], + ) + async def proxy(path: str, request: Request) -> Response: + """把 `/api/*` 同源转发给主应用（进程内 ASGI，不起第二个服务）。 + + 同源是为了避开 CORS，也让浏览器不必直连主服务；`Authorization` 等请求头 + 原样透传。响应连状态码一起回传 —— 这个页面的用处之一就是看真实状态码。 + """ + headers = { + key: value + for key, value in request.headers.items() + if key.lower() not in {"host", "content-length"} + } + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=base_app), + base_url="http://base", + timeout=60, + ) as client: + upstream = await client.request( + request.method, + f"/api/{path}", + headers=headers, + content=await request.body(), + params=request.query_params, + ) + passthrough = { + key: value + for key, value in upstream.headers.items() + if key.lower() in {"content-type", "retry-after", "x-trace-id"} + } + return Response( + content=upstream.content, + status_code=upstream.status_code, + headers=passthrough, + ) + + return console + + +def main() -> None: + parser = argparse.ArgumentParser(description="登录测试台（浏览器验证登录接口）") + parser.add_argument("--port", type=int, default=8099) + args = parser.parse_args() + print(f"登录测试台：http://127.0.0.1:{args.port}") + print(" 演示账号：cust_t/123456　risk_t/666666　admin_t/88888888") + uvicorn.run(build_console(), host="127.0.0.1", port=args.port, log_level="warning") + + +if __name__ == "__main__": + main() diff --git a/tools/probe_auth_state.py b/tools/probe_auth_state.py new file mode 100644 index 0000000..6b22e04 --- /dev/null +++ b/tools/probe_auth_state.py @@ -0,0 +1,96 @@ +"""只读探查：`sys_user` 的认证字段现状 —— 判断"账号密码换令牌"这条路通不通。 + +只做 SELECT；`password_hash` **只统计前缀特征与长度，不输出完整值**（金融项目的基本习惯， +即使是测试库）。 + + python tools/probe_auth_state.py +""" + +from __future__ import annotations + +import asyncio +import json +from collections import Counter +from pathlib import Path +from typing import Any + +from sqlalchemy import text + +from app.infrastructure.db import SessionFactory + +OUTPUT = Path("docs/evidence/auth-state.json") + +#: 常见密码哈希方案的特征前缀。任何一项命中，说明库里存在**真实可校验**的密码。 +KNOWN_HASH_PREFIXES = ( + "$2a$", "$2b$", "$2y$", # bcrypt + "$argon2", # argon2 + "$pbkdf2", "$pbkdf2-sha256", # passlib pbkdf2 + "$5$", "$6$", "$1$", # crypt sha256/sha512/md5 + "$scrypt", +) + + +def _classify(values: list[str]) -> dict[str, Any]: + prefixes: Counter[str] = Counter() + lengths: Counter[int] = Counter() + for value in values: + lengths[len(value)] += 1 + matched = next( + (prefix for prefix in KNOWN_HASH_PREFIXES if value.startswith(prefix)), None + ) + prefixes[matched or f"<未知格式，首字符 {value[:1]!r}>"] += 1 + return { + "length_distribution": dict(sorted(lengths.items())), + "prefix_distribution": dict(prefixes.most_common()), + } + + +async def collect() -> dict[str, Any]: + report: dict[str, Any] = {} + 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() + + report["user_count"] = len(rows) + report["by_status"] = dict( + Counter(str(row["status"]) for row in rows).most_common() + ) + report["by_user_type"] = dict( + Counter(str(row["user_type"]) for row in rows).most_common() + ) + report["password_hash_shape"] = _classify( + [str(row["password_hash"] or "") for row in rows] + ) + # 判断"能不能校验密码"的关键结论，直接给出来。 + hashes = [str(row["password_hash"] or "") for row in rows] + report["has_real_password_hash"] = any( + value.startswith(KNOWN_HASH_PREFIXES) for value in hashes + ) + report["placeholder_examples"] = sorted( + { + value + for value in hashes + if not value.startswith(KNOWN_HASH_PREFIXES) + } + )[:5] + return report + + +async def main() -> None: + report = await collect() + OUTPUT.parent.mkdir(parents=True, exist_ok=True) + OUTPUT.write_text( + json.dumps(report, ensure_ascii=False, indent=2, default=str), + encoding="utf-8", + ) + print(f"wrote {OUTPUT}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tools/set_user_password.py b/tools/set_user_password.py new file mode 100644 index 0000000..cf4d768 --- /dev/null +++ b/tools/set_user_password.py @@ -0,0 +1,123 @@ +"""设置用户登录密码（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 + +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] + +#: 演示口令（用户指定）。键是 `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 _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()))