B1:RBAC 只读查询接口(4 个)+ docs/05 登记 A035-A038

在此之前,权限只能靠脚本改,平台里**没有任何地方能看"谁能访问什么"**。
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 个)。
This commit is contained in:
2026-09-11 21:22:56 +08:00
parent edd8c53c3a
commit 6812fbe317
5 changed files with 487 additions and 0 deletions
+184
View File
@@ -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"] == []