2026-09-12 12:29:20 +08:00
|
|
|
|
"""校验 RBAC 权限号段的一致性(只读,不连数据库)。
|
|
|
|
|
|
|
|
|
|
|
|
## 为什么需要它
|
|
|
|
|
|
|
|
|
|
|
|
权限码的**定义源**是 `tools/seed_test_rbac.py` 的 `PERMISSIONS` 常量 —— 因为那个脚本是
|
|
|
|
|
|
**DELETE 重建**语义:
|
|
|
|
|
|
|
|
|
|
|
|
```
|
|
|
|
|
|
DELETE FROM sys_permission WHERE id BETWEEN 9001 AND 9099
|
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
|
|
任何没并进 `PERMISSIONS` 的权限码,重建一次就没了,表现是"接口突然 403",
|
|
|
|
|
|
而且**没有任何报错线索**。
|
|
|
|
|
|
|
|
|
|
|
|
2026-09-12 出过一次真实事故:库里有一批 `9020-9035` 是 `grant_advisor_role.py` 用**旧号段**
|
|
|
|
|
|
建的,而投顾线把 `9020-9034` 写进了种子 —— 两套 **id→code 映射不同**。种子的清理只清
|
|
|
|
|
|
`role_id 9001-9003` 的角色绑定,`advisor`(9004) 的绑定**不在清理范围内**,于是重建后
|
|
|
|
|
|
advisor 会静默拿到**语义完全错误**的权限组合(例如 id 9020 从 `asset-allocation:generate:self`
|
|
|
|
|
|
变成 `investment-goal:write:self`),全程不报错。
|
|
|
|
|
|
|
|
|
|
|
|
本脚本把这条约束变成可自动检查的四条:
|
|
|
|
|
|
|
|
|
|
|
|
1. 种子内 id 不重复;
|
|
|
|
|
|
2. 每个 `grant_*.py` 声明的 `(id, code)` 都能在种子里找到**完全一致**的一条;
|
|
|
|
|
|
3. `CUSTOMER_PERMISSIONS` 引用的 id 都存在;
|
|
|
|
|
|
4. 各 `grant_*.py` 之间不抢同一个 id。
|
|
|
|
|
|
|
|
|
|
|
|
`tests/unit/tools/test_rbac_seed_consistency.py` 会调用它,所以这四条是纳入门禁的。
|
|
|
|
|
|
|
|
|
|
|
|
用法:python tools/check_rbac_seed_consistency.py
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
|
|
import importlib.util
|
|
|
|
|
|
import sys
|
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
from types import ModuleType
|
|
|
|
|
|
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
2026-09-12 13:03:16 +08:00
|
|
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
|
|
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
2026-09-12 12:29:20 +08:00
|
|
|
|
|
|
|
|
|
|
#: 参与校验的 grant 脚本(都只做「按 code 判重」的幂等补齐)。
|
|
|
|
|
|
GRANT_SCRIPTS: tuple[tuple[str, str, str], ...] = (
|
|
|
|
|
|
("tools/grant_advisor_role.py", "ADVISOR_PERMISSIONS", "投顾"),
|
|
|
|
|
|
("tools/grant_customer_service_phase2_permissions.py", "PHASE2_PERMISSIONS", "客服二期"),
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _load(name: str, relative: str) -> ModuleType:
|
|
|
|
|
|
spec = importlib.util.spec_from_file_location(name, PROJECT_ROOT / relative)
|
|
|
|
|
|
if spec is None or spec.loader is None:
|
|
|
|
|
|
raise RuntimeError(f"无法加载 {relative}")
|
|
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
|
|
|
|
spec.loader.exec_module(module)
|
|
|
|
|
|
return module
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def collect_findings() -> list[str]:
|
|
|
|
|
|
"""返回问题列表;空列表表示一致。"""
|
|
|
|
|
|
problems: list[str] = []
|
|
|
|
|
|
|
|
|
|
|
|
seed = _load("_rbac_seed", "tools/seed_test_rbac.py")
|
|
|
|
|
|
seed_map: dict[int, str] = {}
|
|
|
|
|
|
for row in seed.PERMISSIONS:
|
|
|
|
|
|
permission_id, code = int(row[0]), str(row[1])
|
|
|
|
|
|
if permission_id in seed_map:
|
|
|
|
|
|
problems.append(
|
|
|
|
|
|
f"种子里 id {permission_id} 重复:{seed_map[permission_id]} 与 {code}"
|
|
|
|
|
|
)
|
|
|
|
|
|
seed_map[permission_id] = code
|
|
|
|
|
|
|
|
|
|
|
|
claimed: dict[int, str] = {}
|
|
|
|
|
|
for relative, attribute, label in GRANT_SCRIPTS:
|
|
|
|
|
|
module = _load(f"_rbac_{attribute.lower()}", relative)
|
|
|
|
|
|
for row in getattr(module, attribute):
|
|
|
|
|
|
permission_id, code = int(row[0]), str(row[1])
|
|
|
|
|
|
seeded = seed_map.get(permission_id)
|
|
|
|
|
|
if seeded is None:
|
|
|
|
|
|
problems.append(f"{label}({relative}):id {permission_id}({code})不在种子里")
|
|
|
|
|
|
elif seeded != code:
|
|
|
|
|
|
problems.append(
|
|
|
|
|
|
f"{label}({relative}):id {permission_id} 与种子冲突 —— "
|
|
|
|
|
|
f"种子={seeded},脚本={code}"
|
|
|
|
|
|
)
|
|
|
|
|
|
if permission_id in claimed and claimed[permission_id] != code:
|
|
|
|
|
|
problems.append(
|
|
|
|
|
|
f"id {permission_id} 被两个脚本抢用:{claimed[permission_id]} 与 {code}"
|
|
|
|
|
|
)
|
|
|
|
|
|
claimed[permission_id] = code
|
|
|
|
|
|
|
|
|
|
|
|
for permission_id in seed.CUSTOMER_PERMISSIONS:
|
|
|
|
|
|
if int(permission_id) not in seed_map:
|
|
|
|
|
|
problems.append(f"CUSTOMER_PERMISSIONS 引用了不存在的 id {permission_id}")
|
|
|
|
|
|
|
|
|
|
|
|
return problems
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def main() -> int:
|
|
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
|
|
|
|
sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr]
|
|
|
|
|
|
|
|
|
|
|
|
seed = _load("_rbac_seed_report", "tools/seed_test_rbac.py")
|
|
|
|
|
|
print(f"种子权限 {len(seed.PERMISSIONS)} 条;grant 脚本 {len(GRANT_SCRIPTS)} 个")
|
|
|
|
|
|
|
|
|
|
|
|
problems = collect_findings()
|
|
|
|
|
|
if problems:
|
|
|
|
|
|
print("\n[失败] 号段不一致:")
|
|
|
|
|
|
for item in problems:
|
|
|
|
|
|
print(f" - {item}")
|
|
|
|
|
|
print(
|
|
|
|
|
|
"\n处置:权限码的定义以 tools/seed_test_rbac.py 的 PERMISSIONS 为准;"
|
|
|
|
|
|
"grant 脚本只补种子里缺的,且 id 必须与种子一致。"
|
|
|
|
|
|
)
|
|
|
|
|
|
return 1
|
|
|
|
|
|
|
|
|
|
|
|
print("一致性检查通过:种子内 id 唯一,各 grant 脚本与种子逐条一致。")
|
|
|
|
|
|
return 0
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
|
|
sys.exit(main())
|