# -*- coding: utf-8 -*- """角色与授权 E2E(二期批次 1~4)。 这个脚本验的不是"接口返回 200",而是**"改完之后,系统真的按新授权做事"** —— 权限系统最容易出的错是"看起来配好了、其实没生效",而这个脚本专门盯这一类: | # | 要证明的事 | 怎么证明 | |---|---|---| | 1 | 自定义角色的**权限码真的在挡** | 给自定义角色只授 `task:view` → 该账号 GET /tasks 通、POST /tasks 得 40301 | | 2 | 改授权**立即生效**(不必重新登录) | 用**同一个 token** 改完授权马上再请求,权限随之变化 | | 3 | 停用角色**立即生效** | 停用后同一个 token 立刻失去该角色的权限 | | 4 | 菜单跟着角色走 | `/auth/me` 的 menus 随角色配置增减(登录响应里也必须有,否则"刷新一次才看得见") | | 5 | 内置角色**只读** | 改/删/授权/停用内置角色一律 40001 | | 6 | 越权防线在**服务端** | 学校租户授平台级权限码 → 40301;建"平台运营"类角色 → 40301;发 PLATFORM_ADMIN → 40301 | | 7 | 内置角色的权限**以代码为准** | 人为删掉库里的一行授权 → 接口返回的仍是代码那一套;跑一次同步任务 → 库里那行回来 | | 8 | 防自锁 | 管理员把自己唯一一条 `role:manage` / `role` 菜单去掉 → 40001(而不是真的锁死) | | 9 | 删除守卫 | 还有账号在用的角色删不掉(40045);解绑后能删,且账号立刻失去权限 | 跑法:python scripts/e2e-m10-role.py 前置:后端 8080 在跑、xt-pg / xt-redis 在跑(与其它 e2e 相同)。 """ from __future__ import annotations import argparse import importlib.util import json import os import sys sys.stdout.reconfigure(encoding="utf-8") ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) BASE = "http://localhost:8080" PASSED: list[str] = [] FAILED: list[tuple[str, str]] = [] CREATED_USERS: list[int] = [] CREATED_ROLES: list[int] = [] def _load(module_name: str, filename: str): path = os.path.join(ROOT, "scripts", filename) spec = importlib.util.spec_from_file_location(module_name, path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod R = _load("e2e_m2_roster", "e2e-m2-roster.py") def ok(name: str) -> None: PASSED.append(name) print(f" [OK] {name}") def fail(name: str, detail: str) -> None: FAILED.append((name, detail)) print(f" [FAIL] {name}\n {detail}") def check(name: str, cond: bool, detail: str = "") -> bool: if cond: ok(name) else: fail(name, detail or "断言不成立") return cond def sql(stmt: str) -> str: _, out = R.sql(stmt) return out def body_of(method: str, path: str, token: str | None = None, payload: dict | None = None) -> dict: _, b = R.call(method, f"{BASE}{path}", token, payload) return b def clear_login_fail(*accounts: str) -> None: """清掉登录失败计数:脚本反复跑会把账号锁掉,而锁定看起来像"密码错了"。""" for a in accounts: os.system('docker exec xt-redis redis-cli -a redis_dev_pwd --no-auth-warning ' 'DEL xt:login:fail:%s > NUL 2>&1' % a) def login(username: str, password: str, tenant: str | None = "demo"): _, b = R.login(BASE, username, password, tenant) if b.get("code") != 0: return None, b return b["data"]["token"], b["data"] def role_by_code(token: str, code: str) -> dict: for r in (body_of("GET", "/api/v1/roles", token).get("data") or []): if r.get("roleCode") == code: return r return {} def create_role(token: str, code: str, name: str, user_type: int, perms: list[str], menus: list[str]) -> dict: b = body_of("POST", "/api/v1/roles", token, { "roleCode": code, "roleName": name, "userType": user_type, "permCodes": perms, "menus": menus, }) if b.get("code") == 0 and b.get("data"): CREATED_ROLES.append(b["data"]) return b def create_user(token: str, username: str, real_name: str, phone: str, role_codes: list[str], user_type: int) -> dict: b = body_of("POST", "/api/v1/users", token, { "username": username, "realName": real_name, "phone": phone, "roleCodes": role_codes, "userType": user_type, }) if b.get("code") == 0 and (b.get("data") or {}).get("id"): CREATED_USERS.append(b["data"]["id"]) return b def cleanup() -> None: """只删本脚本造的东西(演示租户里别人的数据一律不碰)。""" for uid in CREATED_USERS: R.sql(f"delete from sys_user_role where user_id={uid};") R.sql(f"delete from sys_user where id={uid};") for rid in CREATED_ROLES: R.sql(f"delete from sys_role_permission where role_id={rid};") R.sql(f"delete from sys_role_menu where role_id={rid};") R.sql(f"delete from sys_role where id={rid};") def main() -> int: ap = argparse.ArgumentParser(description="角色与授权 E2E") ap.parse_args() print("=" * 82) print(" 角色与授权 E2E(自定义角色真的生效 / 内置角色只读 / 越权与自锁防线)") print("=" * 82) suffix = os.urandom(3).hex() role_a_code = f"E2E_ROLE_A_{suffix.upper()}" role_b_code = f"E2E_ROLE_B_{suffix.upper()}" user_a = f"e2e_role_a_{suffix}" user_b = f"e2e_role_b_{suffix}" clear_login_fail("admin", "platform", user_a, user_b) at, adm = login("admin", "Admin@123") if not at: print(f" x 管理员登录失败:{adm}") return 2 admin_uid = adm["user"]["id"] # ══════════════ 1. 权限码目录 ══════════════ print("\n[1] 权限码目录(角色授权界面的勾选项来源)") b = body_of("GET", "/api/v1/permissions", at) groups = b.get("data") or [] check("权限码目录可读且按分组返回", b.get("code") == 0 and len(groups) >= 5, json.dumps(b, ensure_ascii=False)[:200]) items = [i for g in groups for i in (g.get("items") or [])] check("目录里 28 个权限码一个不少(删掉 join:audit / cleanup:run 之后)", len(items) == 28, f"实际 {len(items)}") by_code = {i.get("code"): i for i in items} check("每条权限码都带中文名与适用端", all(i.get("label") and i.get("platform") for i in items)) check("⭐ 平台级权限被标出来(tenant:manage → platformOnly=true)", by_code.get("tenant:manage", {}).get("platformOnly") is True) check("role:manage 不是平台级(学校管理员可以授予)", by_code.get("role:manage", {}).get("platformOnly") is False) # ══════════════ 2. 内置角色:只读 + 权限来自代码 ══════════════ print("\n[2] 内置角色:看得见、改不了") roles = body_of("GET", "/api/v1/roles", at).get("data") or [] check("角色列表可读(内置 4 个 + 可能的自定义角色)", len(roles) >= 4, f"实际 {len(roles)} 个") check("⭐ 内置角色一律 editable=false(权限由代码决定)", all(r.get("editable") is False for r in roles if r.get("roleType") == 1), json.dumps([(r.get("roleCode"), r.get("editable")) for r in roles], ensure_ascii=False)[:220]) sa = role_by_code(at, "SCHOOL_ADMIN") check("内置角色的权限码取代码定义(含 role:manage / task:manage)", "role:manage" in (sa.get("permCodes") or []) and "task:manage" in (sa.get("permCodes") or []), json.dumps(sa, ensure_ascii=False)[:220]) check("内置角色的菜单取代码定义(含 role 菜单)", "role" in (sa.get("menus") or [])) check("内置角色带 userCount(删除守卫与界面都要用)", isinstance(sa.get("userCount"), int) and sa.get("userCount") >= 1, f"userCount={sa.get('userCount')}") check("⭐ 可分配的内置角色 assignable=true(学校管理员能发给账号)", sa.get("assignable") is True, f"assignable={sa.get('assignable')}") # ⚠️ demo 租户里有一行历史遗留的 PLATFORM_ADMIN(早于"平台角色只发平台租户"那次修正)。 # 它正好是这条断言的样本:**看得见、发不出去** —— 界面据此不显示它。 pa = role_by_code(at, "PLATFORM_ADMIN") if pa: check("⭐ 学校租户里的平台运营角色 assignable=false(看得见但发不出去)", pa.get("assignable") is False, f"assignable={pa.get('assignable')}") else: check("学校租户里没有平台运营角色(也没问题)", True) # ══════════════ 3. 建自定义角色:服务端校验 ══════════════ print("\n[3] 建自定义角色(含服务端白名单校验)") b = create_role(at, role_a_code, "E2E 年级审核员", 3, ["task:view", "material:manage", "progress:viewOwn"], ["dashboard", "task"]) check("自定义角色创建成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) role_a = role_by_code(at, role_a_code) check("列表里能看到它,且 editable=true", role_a.get("editable") is True) check("权限码原样保存", sorted(role_a.get("permCodes") or []) == sorted(["task:view", "material:manage", "progress:viewOwn"])) check("菜单原样保存", sorted(role_a.get("menus") or []) == sorted(["dashboard", "task"])) check("userType 落库(账号的用户类型由它推导)", role_a.get("userType") == 3) check("新建角色默认 dataScope=4 仅本人(最保守)", role_a.get("dataScope") == 4) check("新建角色 userCount=0", role_a.get("userCount") == 0) check("角色码重复 → 40001", create_role(at, role_a_code, "重名", 3, ["task:view"], ["task"]).get("code") == 40001) check("⭐ 占用内置角色码 → 40001(不能造一个假 SCHOOL_ADMIN)", create_role(at, "SCHOOL_ADMIN", "假管理员", 2, ["task:view"], ["task"]).get("code") == 40001) check("角色码格式不合法(小写)→ 40001", create_role(at, f"e2e_bad_{suffix}", "小写码", 3, ["task:view"], ["task"]).get("code") == 40001) check("⭐ 学校租户授平台级权限 → 40301", create_role(at, f"E2E_ESC_{suffix.upper()}", "越权角色", 2, ["tenant:manage"], ["tenant"]).get("code") == 40301, json.dumps(create_role(at, f"E2E_ESC2_{suffix.upper()}", "越权角色2", 2, ["tenant:manage"], ["tenant"]), ensure_ascii=False)[:200]) check("不存在的权限码 → 40001", create_role(at, f"E2E_P_{suffix.upper()}", "坏权限", 3, ["no:such:perm"], ["task"]).get("code") == 40001) check("不存在的菜单键 → 40001", create_role(at, f"E2E_M_{suffix.upper()}", "坏菜单", 3, ["task:view"], ["no-such-menu"]).get("code") == 40001) check("⭐ 学校租户配平台专用菜单(租户管理)→ 40301", create_role(at, f"E2E_MT_{suffix.upper()}", "平台菜单", 3, ["task:view"], ["tenant"]).get("code") == 40301) check("⭐ 学校租户建「平台运营」类角色 → 40301", create_role(at, f"E2E_U1_{suffix.upper()}", "平台类角色", 1, ["task:view"], ["task"]).get("code") == 40301) # ══════════════ 4. 发出去 + 权限真的在挡 ══════════════ print("\n[4] 把自定义角色发给账号,验证权限码真的在挡") phone_a = "139" + "".join(c for c in suffix if c.isdigit()).ljust(8, "0")[:8] b = create_user(at, user_a, "E2E角色甲", phone_a, [role_a_code], 3) check("给账号发自定义角色成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) uid_a = (b.get("data") or {}).get("id") pwd_a = (b.get("data") or {}).get("initialPassword") check("新增账号返回初始密码", bool(pwd_a)) check("⭐ 用户类型与角色不符 → 40001(角色说 3,传 4)", create_user(at, f"{user_a}_x", "E2E角色甲x", "13900000001", [role_a_code], 4).get("code") == 40001) check("⭐ 学校租户发 PLATFORM_ADMIN → 40301(原有防线未回归)", create_user(at, f"{user_a}_p", "E2E越权", "13900000002", ["PLATFORM_ADMIN"], 1).get("code") == 40301) ta, prof_a = login(user_a, pwd_a) check("自定义角色账号能登录", bool(ta), json.dumps(prof_a, ensure_ascii=False)[:200]) check("⭐ 登录响应里就带着自定义角色的权限(否则要刷新一次才看得见按钮)", "task:view" in (prof_a.get("permissions") or []), json.dumps(prof_a.get("permissions"), ensure_ascii=False)[:220]) check("登录响应里带着自定义角色的菜单", "task" in (prof_a.get("menus") or []) and "role" not in (prof_a.get("menus") or []), json.dumps(prof_a.get("menus"), ensure_ascii=False)[:220]) b = body_of("GET", "/api/v1/auth/me", ta) check("/auth/me 的 permissions 与登录响应一致", "task:view" in (b.get("data", {}).get("permissions") or []), json.dumps(b, ensure_ascii=False)[:220]) b = body_of("GET", "/api/v1/tasks", ta) check("有 task:view → 任务列表可读", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:160]) b = body_of("POST", "/api/v1/tasks", ta, { "taskName": "不该建出来", "batchNo": "202610", "majorName": "数控加工", "collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31"}) check("⭐ 没有 task:manage → 建任务被 40301 拦住", b.get("code") == 40301, f"code={b.get('code')} msg={b.get('msg')}") b = body_of("GET", "/api/v1/roles", ta) check("⭐ 没有 role:manage / user:view → 读角色列表 40301", b.get("code") == 40301, f"code={b.get('code')}") # ══════════════ 5. 改授权:立即生效(同一个 token)══════════════ print("\n[5] 改授权立即生效(不需要重新登录)") b = body_of("PUT", f"/api/v1/roles/{role_a['id']}/permissions", at, {"permCodes": ["roster:view"]}) check("管理员改角色权限成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:160]) b = body_of("GET", "/api/v1/tasks", ta) check("⭐ 用**同一个 token** 立刻失去 task:view(40301,不必重新登录)", b.get("code") == 40301, f"code={b.get('code')}") b = body_of("GET", "/api/v1/auth/me", ta) check("同一 token 的 /auth/me 权限同步变化", "task:view" not in (b.get("data", {}).get("permissions") or []), json.dumps(b.get("data", {}).get("permissions"), ensure_ascii=False)[:200]) body_of("PUT", f"/api/v1/roles/{role_a['id']}/permissions", at, {"permCodes": ["task:view", "material:manage"]}) b = body_of("GET", "/api/v1/tasks", ta) check("权限加回来 → 立刻又能读(缓存双向都生效)", b.get("code") == 0, f"code={b.get('code')}") # ══════════════ 6. 菜单随角色走 ══════════════ print("\n[6] 菜单跟着角色配置走") body_of("PUT", f"/api/v1/roles/{role_a['id']}/menus", at, {"menus": ["dashboard", "export"]}) menus = (body_of("GET", "/api/v1/auth/me", ta).get("data", {}) or {}).get("menus") or [] check("⭐ 菜单改成 export → /auth/me 里出现 export、task 消失", "export" in menus and "task" not in menus, json.dumps(menus, ensure_ascii=False)[:200]) body_of("PUT", f"/api/v1/roles/{role_a['id']}/menus", at, {"menus": ["dashboard", "task"]}) menus = (body_of("GET", "/api/v1/auth/me", ta).get("data", {}) or {}).get("menus") or [] check("菜单改回来 → task 回来", "task" in menus) # ══════════════ 7. 停用角色立即生效 ══════════════ print("\n[7] 停用角色:立刻不再给权限") b = body_of("PUT", f"/api/v1/roles/{role_a['id']}/status?status=0", at) check("停用成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:160]) b = body_of("GET", "/api/v1/tasks", ta) check("⭐ 同一个 token 立刻失去权限(40301)", b.get("code") == 40301, f"code={b.get('code')}") body_of("PUT", f"/api/v1/roles/{role_a['id']}/status?status=1", at) check("重新启用 → 权限回来", body_of("GET", "/api/v1/tasks", ta).get("code") == 0) check("非法状态值 → 40001", body_of("PUT", f"/api/v1/roles/{role_a['id']}/status?status=9", at).get("code") == 40001) # ══════════════ 8. 内置角色不可改 ══════════════ print("\n[8] 内置角色:改 / 删 / 授权 / 停用 一律拒绝") sid = sa["id"] check("改内置角色 → 40001", body_of("PUT", f"/api/v1/roles/{sid}", at, {"roleName": "改个名", "userType": 2}).get("code") == 40001) check("改内置角色授权 → 40001", body_of("PUT", f"/api/v1/roles/{sid}/permissions", at, {"permCodes": ["task:view"]}).get("code") == 40001) check("改内置角色菜单 → 40001", body_of("PUT", f"/api/v1/roles/{sid}/menus", at, {"menus": ["task"]}).get("code") == 40001) check("停用内置角色 → 40001", body_of("PUT", f"/api/v1/roles/{sid}/status?status=0", at).get("code") == 40001) check("删内置角色 → 40001", body_of("DELETE", f"/api/v1/roles/{sid}", at).get("code") == 40001) check("改一个不存在的角色 → 40001", body_of("PUT", "/api/v1/roles/99999999", at, {"roleName": "x", "userType": 3}).get("code") == 40001) # ══════════════ 9. 账号列表里的角色名(读库)══════════════ print("\n[9] 账号列表的角色名读库(自定义角色的名字不在代码里)") b = body_of("GET", f"/api/v1/users?keyword={user_a}", at) row = next((u for u in (b.get("data", {}).get("list") or []) if u.get("id") == uid_a), {}) check("⭐ 自定义角色显示的是库里的角色名", row.get("roleNames") == ["E2E 年级审核员"], json.dumps(row.get("roleNames"), ensure_ascii=False)[:200]) # ══════════════ 10. 删除守卫 ══════════════ print("\n[10] 删除守卫:还有账号在用就删不掉") b = body_of("DELETE", f"/api/v1/roles/{role_a['id']}", at) check("⭐ 还有账号在用 → 40045", b.get("code") == 40045, f"code={b.get('code')} msg={b.get('msg')}") b = body_of("PUT", f"/api/v1/users/{uid_a}", at, { "realName": "E2E角色甲", "phone": phone_a, "roleCodes": ["HEAD_TEACHER"], "userType": 3}) check("把该账号换成内置角色成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) b = body_of("DELETE", f"/api/v1/roles/{role_a['id']}", at) check("解绑后可以删除", b.get("code") == 0, f"code={b.get('code')} msg={b.get('msg')}") check("删除后列表里不再有它", not role_by_code(at, role_a_code)) ta2, _ = login(user_a, pwd_a) perms = (body_of("GET", "/api/v1/auth/me", ta2).get("data", {}) or {}).get("permissions") or [] # ⚠️ 判据要选**只有自定义角色才有**的那条(material:manage)。 # 第一版断言的是 roster:view / audit:first —— 而 HEAD_TEACHER 本来就有这两条, # 于是"换角色成功"被读成了"权限没清掉"(断言写错,不是功能错)。 check("⭐ 换角色后重新登录:自定义角色的权限没了、新角色的权限在", "material:manage" not in perms and "task:view" in perms, json.dumps(perms, ensure_ascii=False)[:220]) # ══════════════ 11. 防自锁 ══════════════ print("\n[11] 防自锁:不能把自己管理角色的最后一条路去掉") b = create_role(at, role_b_code, "E2E 角色管理员", 2, ["role:manage", "user:manage", "user:view"], ["dashboard", "role", "user"]) check("建「角色管理员」角色成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) role_b = role_by_code(at, role_b_code) b = create_user(at, user_b, "E2E角色乙", "13900000003", [role_b_code], 2) check("发给账号成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) uid_b = (b.get("data") or {}).get("id") pwd_b = (b.get("data") or {}).get("initialPassword") tb, prof_b = login(user_b, pwd_b) check("该账号能登录,且带着 role:manage", "role:manage" in (prof_b.get("permissions") or []), json.dumps(prof_b.get("permissions"), ensure_ascii=False)[:220]) b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/permissions", tb, {"permCodes": ["user:manage", "user:view"]}) check("⭐ 去掉自己唯一的 role:manage → 40001(防自锁)", b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}") b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/menus", tb, {"menus": ["dashboard", "user"]}) check("⭐ 去掉自己唯一的 role 菜单 → 40001(否则菜单里再也点不到本页)", b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}") b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/status?status=0", tb) check("⭐ 停用自己唯一的角色 → 40001", b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}") b = body_of("PUT", f"/api/v1/users/{uid_b}", tb, { "realName": "E2E角色乙", "phone": "13900000003", "roleCodes": ["HEAD_TEACHER"], "userType": 3}) check("⭐ 把自己换成普通角色(会丢掉账号管理)→ 40001", b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}") b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/permissions", tb, {"permCodes": ["role:manage", "user:manage", "user:view", "task:view"]}) check("加权限(不减少)不受防自锁影响", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) b = body_of("PUT", f"/api/v1/users/{uid_b}", tb, { "realName": "E2E角色乙", "phone": "13900000003", "roleCodes": [role_b_code], "userType": 2}) check("改自己的其它字段(角色不变)不受影响", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) # ══════════════ 12. 内置角色授权同步(跨租户定时任务)══════════════ print("\n[12] 内置角色的权限以代码为准 + 每日同步回填") role_id_sa = sa["id"] R.sql(f"delete from sys_role_permission where role_id={role_id_sa} and perm_code='task:clone';") check("夹具:库里那一行已删掉", sql(f"select count(*) from sys_role_permission where role_id={role_id_sa} " f"and perm_code='task:clone' and deleted=0;").strip() == "0") sa2 = role_by_code(at, "SCHOOL_ADMIN") check("⭐ 接口返回的仍是代码那一套(库里缺了也不影响显示与判定)", "task:clone" in (sa2.get("permCodes") or [])) _, ops = R.login(BASE, "platform", "Platform@123", None) ops_token = (ops.get("data") or {}).get("token") check("平台运营账号可登录(同步任务是运维动作)", bool(ops_token)) b = body_of("POST", "/api/v1/schedules/run", ops_token, {"job": "role"}) check("手工触发「内置角色授权同步」", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200]) check("⭐ 同步后库里那行回来了", sql(f"select count(*) from sys_role_permission where role_id={role_id_sa} " f"and perm_code='task:clone' and deleted=0;").strip() == "1") body_of("POST", "/api/v1/schedules/run", ops_token, {"job": "role"}) check("⭐ 再跑一次不产生重复行(幂等)", sql(f"select count(*) from sys_role_permission where role_id={role_id_sa} " f"and perm_code='task:clone' and deleted=0;").strip() == "1") # ══════════════ 13. 越权:跨租户拿不到别人的角色 ══════════════ print("\n[13] 跨租户防线") b = body_of("PUT", f"/api/v1/roles/{role_b['id']}", ops_token, {"roleName": "平台改名", "userType": 2}) check("⭐ 平台运营不选学校就改学校的角色 → 40001(必须先选定学校)", b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}") b = body_of("GET", "/api/v1/roles", None) check("未登录读角色列表 → 40101", b.get("code") == 40101, f"code={b.get('code')}") cleanup() print("\n" + "=" * 82) print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}") print("=" * 82) for n, det in FAILED: print(f" ✗ {n}\n {det}") sys.stdout.flush() return 1 if FAILED else 0 if __name__ == "__main__": raise SystemExit(main())