lzf_0626
6812fbe317
在此之前，权限只能靠脚本改，平台里**没有任何地方能看"谁能访问什么"**。
B1 先把"看得见"做出来：
- GET /api/v1/admin/roles 角色清单 + 权限数 + 在用人数
- GET /api/v1/admin/roles/{role_code} 角色详情
- GET /api/v1/admin/roles/{role_code}/permissions 权限清单（按权限码排序，
便于与各 Service 的 require() 对照）
- GET /api/v1/admin/users/{user_id}/roles 某人**实际解析出来**的角色/权限/数据范围
几个刻意的决定：
1. 权限码复用 `audit:read` 而不新增 `rbac:read`：这份清单本身就是审计材料，且复用是零数据
改动、立刻可用（新增权限码要先改 sys_permission，而它目前由 seed_test_rbac.py 以
DELETE 重建语义管理）。将来要细分再加，不冲突。
2. `/users/{id}/roles` 直接复用 IdentityService.resolve，不自己拼 SQL —— 那正是请求进来时
走的链路（status 检查、assigned_at/expires_at 时间窗、data_scope 取最高、客户分配）。
自己写一遍必然漂移，而"这里查出的权限"与"实际能用的权限"不一致比没有这个接口更糟。
测试里加了一条交叉验证：该接口的 roles/data_scope 必须与登录响应完全一致。
3. 停用账号返回**空权限集**而不是 404 —— 用户存在但拿不到权限，如实呈现比 404 更利于排障。
4. 角色详情与权限清单分开：权限为空的角色不该被误判成"角色不存在"。
5. 全部只读、不写审计（它们返回的就是审计材料本身），docs/05 §19 登记时审计列标"否"。
权限变更（提权/降权）仍无接口 —— docs/05 §19 已注明那不是遗漏，而是需要单独评审
（审计留痕 + 禁止自我提权 + 保护内置角色三条红线）。
另：确认了 load_context 对 sys_user_role.expires_at 是有过滤的（assigned_at<=now AND
(expires_at IS NULL OR expires_at>now)），我上一轮只读了半段 SQL 差点误报。
验证：ruff 干净 / mypy 185 文件 0 错 / 文档守卫 38 份无编号冲突 /
unit+contract 1140 passed / integration 98 passed（本批新增 8 个）。
203 lines
8.5 KiB
Python
203 lines
8.5 KiB
Python
"""角色与权限的只读查询（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 [],
|
||
}
|