150 lines
5.6 KiB
Python
150 lines
5.6 KiB
Python
|
"""对账：代码声明的权限码 vs 库里存在的 vs 各角色实际拥有的（只读）。
|
|||
|
|
|
|||
|
|
## 为什么需要它
|
|||
|
|
|
|||
|
|
权限缺失的表现一律是 `AGENT_PERMISSION_DENIED: 缺少操作权限`，但底下有**三种完全不同的原因**，
|
|||
|
|
报错区分不出来，排查时极易往错的方向找：
|
|||
|
|
|
|||
|
|
1. **权限码在库里根本不存在**（代码写了、没人建）→ 任何角色都过不去，
|
|||
|
|
看起来却像"这个角色没授权"；
|
|||
|
|
2. 权限码存在，但**没授给这个角色**（例如投顾缺 `investment-goal:write:self`）；
|
|||
|
|
3. 角色本身没有对应的**接口范围或界面**（例如运营只有 `offsite:write`，
|
|||
|
|
而它要用的 `financial:nl2sql:read` 从没建过）。
|
|||
|
|
|
|||
|
|
这类问题**服务层测试测不出来** —— 那层直接构造 `RequestContext`，权限是测试自己塞的。
|
|||
|
|
只有把"代码声明"和"库里数据"放一起比才看得见。2026-09-12 用它在主干上查出
|
|||
|
|
6 个不存在的权限码（`promotion:*` 四个导致推广材料整条线 403、`financial:nl2sql:read`、
|
|||
|
|
`probe:read`）。
|
|||
|
|
|
|||
|
|
只读：不连业务逻辑、不改任何数据。
|
|||
|
|
|
|||
|
|
用法：
|
|||
|
|
|
|||
|
|
python tools/check_permission_coverage.py
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import re
|
|||
|
|
import sys
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|||
|
|
|
|||
|
|
from app.core.config import get_settings # noqa: E402
|
|||
|
|
|
|||
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|||
|
|
|
|||
|
|
#: 代码里声明权限码的几种写法。
|
|||
|
|
PATTERNS: tuple[str, ...] = (
|
|||
|
|
r'require\(\s*context\s*,\s*"([^"]+)"',
|
|||
|
|
r'^\s*permission\s*[:=]\s*"([^"]+)"',
|
|||
|
|
r'^[A-Z_]*PERMISSION[A-Z_]*\s*[:=]\s*"([^"]+)"',
|
|||
|
|
r'required_permission\s*=\s*"([^"]+)"',
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
CODE_SHAPE = re.compile(r"^[a-z][a-z0-9_-]*(?::[a-z0-9_-]+)+$")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def collect_required(root: Path = PROJECT_ROOT) -> dict[str, set[str]]:
|
|||
|
|
found: dict[str, set[str]] = {}
|
|||
|
|
for path in (root / "app").rglob("*.py"):
|
|||
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|||
|
|
for pattern in PATTERNS:
|
|||
|
|
for match in re.finditer(pattern, text, re.MULTILINE):
|
|||
|
|
code = match.group(1)
|
|||
|
|
if CODE_SHAPE.match(code):
|
|||
|
|
found.setdefault(code, set()).add(path.relative_to(root).as_posix())
|
|||
|
|
return found
|
|||
|
|
|
|||
|
|
|
|||
|
|
def load_from_db() -> tuple[dict[str, int], dict[str, set[str]]]:
|
|||
|
|
"""返回 `(权限码 → id, 角色 → 权限码集合)`。"""
|
|||
|
|
settings = get_settings()
|
|||
|
|
dsn = settings.mysql_dsn.split("://", 1)[1]
|
|||
|
|
credentials, location = dsn.split("@", 1)
|
|||
|
|
user, password = credentials.split(":", 1)
|
|||
|
|
host_port, database = location.split("/", 1)
|
|||
|
|
host, _, port = host_port.partition(":")
|
|||
|
|
return _query(host, int(port or 3306), user, password, database)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _query(host: str, port: int, user: str, password: str, database: str):
|
|||
|
|
import asyncio
|
|||
|
|
|
|||
|
|
import asyncmy
|
|||
|
|
|
|||
|
|
async def run():
|
|||
|
|
connection = await asyncmy.connect(
|
|||
|
|
host=host, port=port, user=user, password=password, db=database
|
|||
|
|
)
|
|||
|
|
try:
|
|||
|
|
cursor = connection.cursor()
|
|||
|
|
await cursor.execute("SELECT id, permission_code FROM sys_permission")
|
|||
|
|
permissions = {str(code): int(pid) for pid, code in await cursor.fetchall()}
|
|||
|
|
|
|||
|
|
await cursor.execute(
|
|||
|
|
"""
|
|||
|
|
SELECT r.role_code, p.permission_code
|
|||
|
|
FROM sys_role r
|
|||
|
|
LEFT JOIN sys_role_permission rp ON rp.role_id = r.id
|
|||
|
|
LEFT JOIN sys_permission p ON p.id = rp.permission_id
|
|||
|
|
"""
|
|||
|
|
)
|
|||
|
|
roles: dict[str, set[str]] = {}
|
|||
|
|
for role_code, permission_code in await cursor.fetchall():
|
|||
|
|
roles.setdefault(str(role_code), set())
|
|||
|
|
if permission_code:
|
|||
|
|
roles[str(role_code)].add(str(permission_code))
|
|||
|
|
return permissions, roles
|
|||
|
|
finally:
|
|||
|
|
connection.close()
|
|||
|
|
|
|||
|
|
return asyncio.run(run())
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
required = collect_required()
|
|||
|
|
permissions, roles = load_from_db()
|
|||
|
|
|
|||
|
|
print(f"代码里声明的权限码：{len(required)} 个")
|
|||
|
|
print(f"库里已有的权限码：{len(permissions)} 个")
|
|||
|
|
|
|||
|
|
missing = sorted(code for code in required if code not in permissions)
|
|||
|
|
print()
|
|||
|
|
print(f"【一】代码要求、但库里不存在（任何角色都过不去）：{len(missing)} 个")
|
|||
|
|
for code in missing:
|
|||
|
|
where = sorted(required[code])[:3]
|
|||
|
|
print(f" ✗ {code:<44} 用于 {', '.join(where)}")
|
|||
|
|
|
|||
|
|
unused = sorted(code for code in permissions if code not in required)
|
|||
|
|
if unused:
|
|||
|
|
print()
|
|||
|
|
print(f"【二】库里有、代码里没直接搜到（可能是变量拼接或历史遗留）：{len(unused)} 个")
|
|||
|
|
for code in unused:
|
|||
|
|
holders = sorted(r for r, s in roles.items() if code in s)
|
|||
|
|
print(f" · {code:<44} 持有角色 {holders or '无'}")
|
|||
|
|
|
|||
|
|
print()
|
|||
|
|
print("【三】各角色的权限数，以及「代码要求却没拿到」的码：")
|
|||
|
|
for role, codes in sorted(roles.items()):
|
|||
|
|
lacks = sorted(c for c in required if c in permissions and c not in codes)
|
|||
|
|
print(f" {role:<16} 共 {len(codes):>2} 项；代码要求但未授予 {len(lacks)} 个")
|
|||
|
|
for code in lacks[:12]:
|
|||
|
|
print(f" - {code}")
|
|||
|
|
if len(lacks) > 12:
|
|||
|
|
print(f" … 另有 {len(lacks) - 12} 个")
|
|||
|
|
|
|||
|
|
if missing:
|
|||
|
|
print()
|
|||
|
|
print("处置：把这几个码并进 `tools/seed_test_rbac.py` 的 `PERMISSIONS`（那是唯一定义源），")
|
|||
|
|
print(" 再按角色在对应的 `grant_*.py` 里授权。")
|
|||
|
|
return 1
|
|||
|
|
print()
|
|||
|
|
print("结论：代码要求的权限码在库里都存在。")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(main())
|