# -*- coding: utf-8 -*- """ 第十八批: 登录与权限 (交易系统管理员 / 交易员) 的纯逻辑 —— 不联网、不连库 ===================================================================== 本批只测 app/web/auth.py 里的纯函数, 不导入 app.web.main (那会拉起 fastapi 与一堆 基础设施依赖)。覆盖四件事: 1. 角色判定 role_from_permissions: 管理员标记命中/未中、交易员准入闸放行与拦截; 2. 会话票 make_session / verify_session: 签发-校验往返, 篡改/过期/换密钥/空密钥一律拒; 3. bshop 返回解析 parse_bshop_login: 喂真实样例抽出手机号/用户名/权限/组织, code!=200 判失败; 4. 接口鉴权 authz_decision + is_public_path: 「写操作默认拒绝」——交易员白名单外的写接口 一律要管理员 (专门钉死 /api/macro/adopt、POST /api/proposals、.../archive 与 .../cancel 结尾段区分), GET 默认放行仅少数运维读接口要管理员; 公开白名单放行首页与登录接口。 运行: python scripts/test_batch18_units.py """ import os import sys import traceback sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.web import auth as A # noqa: E402 RESULTS = [] def case(name): def deco(fn): RESULTS.append((name, fn)) return fn return deco # 与用户给的 bshop 登录成功返回同构的样例 (裁剪了权限条数, 字段与层级一致) SAMPLE = { "code": 200, "msg": "登录成功", "data": { "access_token": "SMxzzyfoIJogItsmiR9Qv0NdnFAoKE-F5WdJPDhzZOA", "token_type": "bearer", "expires_in": 604800, "user": {"phone": "18795373439", "username": "18795373439", "wechat_bound": True}, "permissions": [ {"system_key": "band", "menu_key": "-", "system_name": "波段系统后端", "menu_name": "波段系统后端"}, {"system_key": "band", "menu_key": "bkld", "system_name": "波段系统后端", "menu_name": "板块轮动"}, {"system_key": "band", "menu_key": "yhyqx", "system_name": "波段系统后端", "menu_name": "用户与权限"}, {"system_key": "band_web", "menu_key": "traded", "system_name": "波段系统网页端", "menu_name": "波段监控"}, ], "orgs": [{"id": 277097889622, "org_name": "高姐"}, {"id": 489281497140, "org_name": "天盟资本"}], }, } # ---------------------------------------------------------------- 一、角色判定 @case("角色: 命中管理员标记 -> 交易系统管理员") def _role_admin(): r = A.role_from_permissions(["band:bkld", "pms:admin"], "pms:admin", "") assert r == A.ROLE_ADMIN, r @case("角色: 只有普通权限 -> 交易员") def _role_trader(): r = A.role_from_permissions(["band:bkld", "band:yhyqx"], "pms:admin", "") assert r == A.ROLE_TRADER, r @case("角色: 配了交易员准入且占其一 -> 交易员") def _role_trader_gate_pass(): r = A.role_from_permissions(["band:ccgl"], "pms:admin", "band:ccgl,pms:-") assert r == A.ROLE_TRADER, r @case("角色: 配了交易员准入却一条不占 -> None (无权进入)") def _role_denied(): r = A.role_from_permissions(["band:bkld"], "pms:admin", "band:ccgl,pms:-") assert r is None, r @case("角色: 管理员标记可多条, 命中任一即算") def _role_admin_multi(): r = A.role_from_permissions(["band:yhyqx"], "pms:admin, band:yhyqx", "") assert r == A.ROLE_ADMIN, r # ---------------------------------------------------------------- 二、会话票 @case("会话: 签发-校验往返, payload 字段齐全") def _sess_roundtrip(): tok = A.make_session("18795373439", "老王", A.ROLE_ADMIN, "s3cr3t", 12, now=1000.0) p = A.verify_session(tok, "s3cr3t", now=1001.0) assert p and p["phone"] == "18795373439" and p["username"] == "老王", p assert p["role"] == A.ROLE_ADMIN and p["exp"] == 1000 + 12 * 3600, p @case("会话: 篡改正文或签名 -> None") def _sess_tampered(): tok = A.make_session("186", "u", A.ROLE_TRADER, "s3cr3t", 12, now=1000.0) b, sig = tok.split(".", 1) assert A.verify_session(b + "x." + sig, "s3cr3t", now=1001.0) is None assert A.verify_session(b + "." + sig + "x", "s3cr3t", now=1001.0) is None @case("会话: 过期 / 换密钥 / 空密钥 / 空票 一律拒") def _sess_reject(): tok = A.make_session("186", "u", A.ROLE_TRADER, "s3cr3t", 1, now=1000.0) assert A.verify_session(tok, "s3cr3t", now=1000.0 + 3600 + 1) is None, "过期没拒" assert A.verify_session(tok, "other", now=1001.0) is None, "换密钥没拒" assert A.verify_session(tok, "", now=1001.0) is None, "空密钥没拒" assert A.verify_session("", "s3cr3t", now=1001.0) is None, "空票没拒" # ---------------------------------------------------------------- 三、bshop 解析 @case("解析: 真实样例 -> 手机号/用户名/权限/组织") def _parse_ok(): r = A.parse_bshop_login(SAMPLE) assert r["ok"] is True, r assert r["phone"] == "18795373439" and r["username"] == "18795373439", r assert "band:bkld" in r["permissions"] and "band:-" in r["permissions"], r["permissions"] assert "band:yhyqx" in r["permissions"], r["permissions"] assert len(r["orgs"]) == 2 and r["orgs"][0]["org_name"] == "高姐", r["orgs"] @case("解析: 样例接角色 —— 默认标记判交易员, 复用 band:yhyqx 判管理员") def _parse_then_role(): r = A.parse_bshop_login(SAMPLE) assert A.role_from_permissions(r["permissions"], "pms:admin", "") == A.ROLE_TRADER assert A.role_from_permissions(r["permissions"], "band:yhyqx", "") == A.ROLE_ADMIN @case("解析: code!=200 / 缺手机号 / 非字典 -> 失败") def _parse_fail(): assert A.parse_bshop_login({"code": 401, "msg": "密码错误"})["ok"] is False assert A.parse_bshop_login({"code": 200, "data": {"user": {}}})["ok"] is False assert A.parse_bshop_login("nope")["ok"] is False # ---------------------------------------------------------------- 四、接口鉴权 @case("鉴权: 交易员白名单 4 条写接口放行 (any)") def _authz_trader_ok(): for m, path in [("POST", "/api/commands"), ("POST", "/api/commands/CMD-1/cancel"), ("POST", "/api/commands/CMD-1/replan"), ("POST", "/api/proposals/PRP-1/decide")]: assert A.authz_decision(m, path) == "any", (m, path, A.authz_decision(m, path)) @case("鉴权: 写操作默认拒绝 —— 白名单外一律要管理员") def _authz_admin_default(): admin_paths = [ ("POST", "/api/params"), ("POST", "/api/ops/reconcile"), ("POST", "/api/strategies"), ("POST", "/api/strategies/S1/status"), ("POST", "/api/industry/import"), ("POST", "/api/ws-channel/clear-resync"), ("POST", "/api/instructions/INS-1/cancel"), ("POST", "/api/macro/adopt"), # 评审点出的漏网写接口, 兜住 ("POST", "/api/proposals"), # 人工补录=造数调试口, 归管理员 ("POST", "/api/commands/CMD-1/archive"), # 结尾段 archive != cancel ("POST", "/api/proposals/PRP-1/archive"), ] for m, path in admin_paths: assert A.authz_decision(m, path) == "admin", (m, path, A.authz_decision(m, path)) @case("鉴权: GET 默认放行, 仅三条运维读接口要管理员") def _authz_gets(): for path in ["/api/ops/rebuild-preflight", "/api/ops/rebuild-accept", "/api/ops/downstream-schema"]: assert A.authz_decision("GET", path) == "admin", path for path in ["/api/overview", "/api/positions", "/api/ledger", "/api/proposals"]: assert A.authz_decision("GET", path) == "any", path @case("公开白名单: 首页/静态/登录接口放行, 业务接口不放行") def _public(): for p in ["/", "/health", "/favicon.ico", "/static", "/static/vendor/vue.global.js", "/api/auth/login", "/api/auth/logout", "/api/me"]: assert A.is_public_path(p) is True, p for p in ["/api/params", "/api/overview", "/api/commands"]: assert A.is_public_path(p) is False, p def main(): passed, failed = 0, [] for name, fn in RESULTS: try: fn() passed += 1 print(f" ✓ {name}") except Exception as e: failed.append((name, e)) print(f" ✗ {name}: {type(e).__name__}: {e}") traceback.print_exc() print() if failed: print(f"FAILED {len(failed)}/{len(RESULTS)}") sys.exit(1) print(f"ALL PASS ({passed} cases)") if __name__ == "__main__": main()