"""只读账号 / 管理员的权限边界对账。 做三件事： 1. 用 viewer 登录，把「读接口」逐个打一遍 —— 必须全部 code=0。 2. 用 admin 登录，同样的读接口再打一遍 —— 结果必须与 viewer 逐项一致， 以此证明只读账号没有被削掉任何读取能力。 （只读 bug 的典型形态是「某人少了一个口子」，只有拿两个角色横向对比才看得出来。） 3. 用 viewer 打所有写接口 —— 必须全部被拒，且拒绝理由必须是 403（权限）， 不能是 404/405 这种「路由压根不存在」的巧合。 设计上的两个讲究： - 所有业务 ID 都在运行时从列表接口里现取，不写死 1 号。硬编码 ID 一旦数据库 重新播种就会全线 404，届时看起来像权限炸了，实际只是测试自己过期了。 - 写接口的 body 一律填足必要字段。若因缺参数返回 422，虽然也是「被拒」， 但那证明不了权限生效 —— 所以额外断言「必须是 403」。 用法：python verify/perm_check.py [--base http://127.0.0.1:8010] """ from __future__ import annotations import argparse import json import sys import urllib.error import urllib.request from pathlib import Path BASE = "http://127.0.0.1:8010" PROBE_NAME = "权限对账临时顾问" passed = 0 failed: list[str] = [] def purge_probe() -> None: """收尾：把本脚本造的探针尸体物理删掉。 第 4 节的写权限验证真的建了一条顾问记录（否则证明不了 admin 没被门禁拦住）， 然后走接口把它删了 —— 但接口删的是逻辑删除，每跑一次就多留一行 is_del=1。 演示库是给人看的，不能越跑越脏，所以这里复用 purge_selftest 的清理逻辑硬删。 """ try: sys.path.insert(0, str(Path(__file__).resolve().parent)) import purge_selftest # noqa: PLC0415 findings = purge_selftest.collect() deleted = purge_selftest.purge(findings) n = sum(deleted.values()) print(f" [清理] 探针数据已物理删除 {n} 行" if n else " [清理] 无探针残留") except Exception as exc: # noqa: BLE001 print(f" [WARN] 探针清理失败（不影响权限结论）：{exc}") def call(method: str, path: str, token: str, body: dict | None = None) -> tuple[int, dict]: url = BASE + path data = None headers = {"Authorization": f"Bearer {token}"} if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=20) as resp: return resp.status, json.loads(resp.read().decode()) except urllib.error.HTTPError as e: raw = e.read().decode() try: return e.code, json.loads(raw) except Exception: return e.code, {"raw": raw[:200]} except Exception as exc: # noqa: BLE001 return -1, {"raw": f"{type(exc).__name__}: {exc}"} def check(cond: bool, label: str, extra: str = "") -> None: global passed if cond: passed += 1 print(f" [OK] {label}") else: failed.append(label) print(f" [FAIL] {label} {extra}") def call_raw(method: str, path: str, token: str) -> tuple[int, bytes, str]: """打二进制接口（比如 xlsx 模板下载），返回 (状态码, 原始字节, content-type)。""" req = urllib.request.Request(BASE + path, headers={"Authorization": f"Bearer {token}"}, method=method) try: with urllib.request.urlopen(req, timeout=20) as resp: return resp.status, resp.read(), resp.headers.get("content-type", "") except urllib.error.HTTPError as e: return e.code, e.read(), e.headers.get("content-type", "") except Exception as exc: # noqa: BLE001 return -1, str(exc).encode(), "" def login(username: str, password: str) -> str: status, body = call("POST", "/auth/login", "", {"username": username, "password": password}) if status != 200 or body.get("code") != 0: print(f"登录失败 {username}: HTTP {status} {body}") sys.exit(1) return body["data"]["access_token"] def first_id(token: str, path: str, id_key: str = "id") -> int | None: status, body = call("GET", path, token) data = body.get("data") or {} items = data.get("items") if isinstance(data, dict) else data if not items: return None row = items[0] return row.get(id_key) def main() -> int: global BASE ap = argparse.ArgumentParser() ap.add_argument("--base", default=BASE) args = ap.parse_args() BASE = args.base print("=" * 74) print(f"权限边界对账 @ {BASE}") print("=" * 74) vtoken = login("viewer", "viewer123") atoken = login("admin", "admin123") # ---------- 运行时取真实 ID（用 admin，它权限最全）---------- sid = first_id(atoken, "/students?page=1&page_size=1") cid = first_id(atoken, "/classes?page=1&page_size=1") tid = first_id(atoken, "/teachers?page=1&page_size=1") aid = first_id(atoken, "/advisors?page=1&page_size=1") eid = first_id(atoken, "/employments?page=1&page_size=1") scid = first_id(atoken, "/scores?page=1&page_size=1") print(f"\n运行时取样 ID：student={sid} class={cid} teacher={tid} advisor={aid} employment={eid} score={scid}") missing = [n for n, v in [("student", sid), ("class", cid), ("teacher", tid), ("advisor", aid), ("employment", eid), ("score", scid)] if v is None] if missing: print(f"数据库缺少样本数据：{missing} —— 请先跑 app/scripts/seed_data.py 再对账。") return 2 # ---------- 读接口清单（按真实路由表）---------- read_targets: list[tuple[str, str, str]] = [ ("学生列表", "GET", "/students?page=1&page_size=5"), ("学生下拉", "GET", "/students/meta/options"), ("学生详情", "GET", f"/students/{sid}"), ("学生成绩明细", "GET", f"/students/{sid}/scores"), ("成绩列表", "GET", "/scores?page=1&page_size=5"), ("成绩口径", "GET", "/scores/meta"), ("某生成绩汇总", "GET", f"/scores/student/{sid}"), ("红线名单", "GET", "/scores/warnings"), ("班级列表", "GET", "/classes?page=1&page_size=5"), ("班级下拉", "GET", "/classes/meta/options"), ("班级详情", "GET", f"/classes/{cid}"), ("老师列表", "GET", "/teachers?page=1&page_size=5"), ("老师详情", "GET", f"/teachers/{tid}"), ("顾问列表", "GET", "/advisors?page=1&page_size=5"), ("顾问详情", "GET", f"/advisors/{aid}"), ("就业列表", "GET", "/employments?page=1&page_size=5"), ("某生就业", "GET", f"/employments/student/{sid}"), ("某班就业", "GET", f"/employments/class/{cid}"), ("总览", "GET", "/statistics/overview"), ("班级概览", "GET", "/statistics/classes/overview"), ("年龄分布", "GET", "/statistics/students/age-distribution"), ("按年龄筛", "GET", "/statistics/students/by-age?op=gt&value=20"), ("班级均分", "GET", "/statistics/scores/class-average"), ("班级均分排序", "GET", "/statistics/scores/class-average?order=desc"), ("全科达标", "GET", "/statistics/scores/all-above"), ("多次不及格", "GET", "/statistics/scores/failures?min_times=2"), ("成绩波动", "GET", "/statistics/scores/volatility?top=5"), ("就业时长", "GET", "/statistics/employment/durations"), ("班级均时长", "GET", "/statistics/employment/class-avg-duration"), ("薪资 TopN", "GET", "/statistics/employment/top-salary?top=5"), ("就业漏斗", "GET", "/statistics/employment/funnel"), ("规则字段", "GET", "/advanced/meta"), ("规则示例", "GET", "/advanced/examples"), ("健康检查", "GET", "/api/health"), ("[兼容]某生就业", "GET", f"/employment/students/{sid}"), ("[兼容]某班就业", "GET", f"/employment/class/{cid}"), ] # POST 但语义是「读」的接口，必须与上表同等对待（viewer 要能用） read_post: list[tuple[str, str, dict]] = [ ("高级查询", "/advanced/query", {"model": "student", "rules": [], "page": 1, "page_size": 5}), ("高级聚合", "/advanced/aggregate", {"model": "student", "group_by": ["class_id"], "metrics": [{"func": "count", "alias": "人数"}]}), ] write_targets: list[tuple[str, str, str, dict | None]] = [ ("新增学生", "POST", "/students", {"name": "权限探针", "gender": 1}), ("改学生", "PUT", f"/students/{sid}", {"name": "权限探针"}), ("删学生", "DELETE", f"/students/{sid}", None), ("恢复学生", "POST", f"/students/{sid}/restore", None), ("批量删学生", "POST", "/students/batch/delete", {"ids": [sid]}), ("导入学生", "POST", "/students/import", None), ("录成绩", "POST", "/scores", {"stu_id": sid, "exam_seq": 99, "score": 60}), ("批量录成绩", "POST", "/scores/batch", {"items": [{"stu_id": sid, "exam_seq": 99, "score": 60}]}), ("改成绩", "PUT", f"/scores/{scid}", {"score": 61}), ("删成绩", "DELETE", f"/scores/{scid}", None), ("清空成绩", "POST", f"/scores/student/{sid}/clear", None), ("新增班级", "POST", "/classes", {"name": "权限探针班"}), ("改班级", "PUT", f"/classes/{cid}", {"name": "权限探针班"}), ("删班级", "DELETE", f"/classes/{cid}", None), ("新增老师", "POST", "/teachers", {"name": "权限探针师"}), ("改老师", "PUT", f"/teachers/{tid}", {"name": "权限探针师"}), ("删老师", "DELETE", f"/teachers/{tid}", None), ("新增顾问", "POST", "/advisors", {"name": "权限探针顾"}), ("改顾问", "PUT", f"/advisors/{aid}", {"name": "权限探针顾"}), ("删顾问", "DELETE", f"/advisors/{aid}", None), ("登记就业", "POST", "/employments", {"stu_id": sid, "open_date": "2024-01-01"}), ("改就业", "PUT", f"/employments/{eid}", {"city": "深圳"}), ("删就业", "DELETE", f"/employments/{eid}", None), ("[兼容]加就业", "POST", f"/employment/students/{sid}", {"city": "深圳"}), ("新增账号", "POST", "/auth/accounts", {"username": "permprobe", "password": "permprobe1", "real_name": "权限探针"}), ] # ---------- 1. viewer 读 ---------- print(f"\n--- 1. viewer 读权限（{len(read_targets)} 个 GET + {len(read_post)} 个读语义 POST，应全部 code=0）---") v_read_ok: dict[str, bool] = {} for name, method, path in read_targets: status, body = call(method, path, vtoken) good = status == 200 and body.get("code") == 0 v_read_ok[name] = good check(good, f"viewer 读 {name}", f"→ HTTP {status} {str(body)[:130]}") for name, path, payload in read_post: status, body = call("POST", path, vtoken, payload) good = status == 200 and body.get("code") == 0 v_read_ok[name] = good check(good, f"viewer 读 {name}", f"→ HTTP {status} {str(body)[:150]}") # ---------- 2. admin 读（逐项一致性）---------- print(f"\n--- 2. admin 读权限（与 viewer 逐项一致）---") for name, method, path in read_targets: status, body = call(method, path, atoken) good = status == 200 and body.get("code") == 0 check(good, f"admin 读 {name}", f"→ HTTP {status} {str(body)[:130]}") check(good == v_read_ok[name], f"admin/viewer 一致 · {name}", f"→ admin={good} viewer={v_read_ok[name]}") for name, path, payload in read_post: status, body = call("POST", path, atoken, payload) good = status == 200 and body.get("code") == 0 check(good, f"admin 读 {name}", f"→ HTTP {status} {str(body)[:150]}") check(good == v_read_ok[name], f"admin/viewer 一致 · {name}", f"→ admin={good} viewer={v_read_ok[name]}") # ---------- 2b. 二进制下载接口（xlsx 模板）---------- print("\n--- 2b. 导入模板下载（xlsx 二进制，避开 JSON 解析）---") for who, tok in [("viewer", vtoken), ("admin", atoken)]: status, raw, ctype = call_raw("GET", "/students/import/template", tok) is_xlsx = raw[:2] == b"PK" # xlsx 本质是 zip，魔数 PK check(status == 200 and is_xlsx, f"{who} 能下载导入模板（xlsx）", f"→ HTTP {status} type={ctype} len={len(raw)}") # ---------- 3. viewer 写（必须 403）---------- print(f"\n--- 3. viewer 写权限（{len(write_targets)} 个写接口，必须 403）---") for name, method, path, payload in write_targets: status, body = call(method, path, vtoken, payload) code = body.get("code") is403 = status == 403 or code == 403 check(is403, f"viewer 写 {name} → 403", f"→ HTTP {status} code={code} {str(body)[:110]}") # ---------- 4. admin 写（抽样放行 + 回滚）---------- print("\n--- 4. admin 写权限（抽样放行，跑完回滚）---") status, body = call("POST", "/advisors", atoken, {"name": PROBE_NAME}) made = body.get("code") == 0 check(made, "admin 能新增顾问", f"→ HTTP {status} {str(body)[:150]}") if made: probe_id = body["data"]["id"] status, body = call("DELETE", f"/advisors/{probe_id}", atoken) check(body.get("code") == 0, "admin 能删除刚建的顾问（逻辑删除）", f"→ {str(body)[:150]}") # ---------- 5. 管理员专属 ---------- print("\n--- 5. 管理员专属接口边界 ---") status, body = call("GET", "/auth/accounts", vtoken) check(body.get("code") == 403, "viewer 读账号列表 → 403", f"→ HTTP {status} {str(body)[:120]}") status, body = call("GET", "/auth/accounts", atoken) check(body.get("code") == 0, "admin 读账号列表 → 放行", f"→ HTTP {status} {str(body)[:120]}") # 未登录也不该漏 print("\n--- 6. 无 token 访问 ---") for name, method, path in [("学生列表", "GET", "/students?page=1&page_size=1"), ("总览", "GET", "/statistics/overview")]: status, body = call(method, path, "") check(body.get("code") == 401, f"无 token 读 {name} → 401", f"→ HTTP {status} {str(body)[:120]}") print("\n--- 7. 收尾清理 ---") purge_probe() print("\n" + "=" * 74) total = passed + len(failed) print(f"结果：{passed}/{total} 通过") if failed: print("失败项：") for f in failed: print(" -", f) print("=" * 74) return 0 if not failed else 1 if __name__ == "__main__": sys.exit(main())