lzf_0626
bfd622b592
- tools/portal.py：客户/风控/运营/管理员/投顾五个工作台，走真实登录与真实接口 - 新增 6 个此前从未建过的权限码（promotion:* 4 个 / financial:nl2sql:read / probe:read）， 它们让推广材料与金融 NL2SQL 两条线对所有角色都是 403 - 投顾权限 10 → 25 项：补投资目标创建确认、agent:run、行情、知识检索、客户画像、推广材料 - 运营补 financial:nl2sql:read；create_test_user.py 不再硬编码角色 id（改按 role_code 查库） - portal 的写请求补 Idempotency-Key 头（漏了会被 AGENT_INPUT_INVALID 拒） - 新增 tools/check_permission_coverage.py 做权限对账
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())
|