299 lines
14 KiB
Python
299 lines
14 KiB
Python
"""只读账号 / 管理员的权限边界对账。
|
||
|
||
做三件事：
|
||
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())
|