zlt
/
tradingSystem
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

添加权限

This commit is contained in:
zlt 2026-08-27 10:16:45 +08:00
parent fcc39984a9
commit 5bb08fe5b6
6 changed files with 92 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
app/web/auth.py
View File

@ -7,14 +7,11 @@ PMS 登录与权限 —— 认证解析 / 角色判定 / 会话票签验 / 接
（scripts/test_batch18_units.py）。唯一碰网络的是 call_bshop_login 一个函数， （scripts/test_batch18_units.py）。唯一碰网络的是 call_bshop_login 一个函数，
它把真正的 HTTP 调用隔离出来，其余全部可离线测。 它把真正的 HTTP 调用隔离出来，其余全部可离线测。
三块职责： 一个人可以同时持有多个角色（既是系统管理员又是交易员），所以角色一律按「集合」处理，
一、认证：解析第三方登录系统 bshop 的登录返回（parse_bshop_login）。 不折叠成单一角色。
二、角色：从 bshop 返回的权限清单里认出系统管理员 / 交易员（role_from_permissions）。
三、会话与鉴权：自签会话票（make_session / verify_session），以及按「方法 + 路径」
判这个请求要什么角色（authz_decision）。
会话票格式：base64url( json({phone, username, role, exp}) ) + "." + base64url( hmac_sha256 ) 会话票格式：base64url( json({phone, username, roles, exp}) ) + "." + base64url( hmac_sha256 )
—— 只用标准库 hmac / hashlib，不引新依赖。exp 是 Unix 秒。 —— 只用标准库 hmac / hashlib，不引新依赖。exp 是 Unix 秒。roles 是角色字符串列表。
""" """
from __future__ import annotations from __future__ import annotations
@ -23,7 +20,7 @@ import hashlib
import hmac import hmac
import json import json
import time import time
from typing import Optional from typing import List, Optional
# 角色常量：既是写进会话票的字面值，也是回给前端、供前端判断的字面值。 # 角色常量：既是写进会话票的字面值，也是回给前端、供前端判断的字面值。
ROLE_ADMIN = "系统管理员" ROLE_ADMIN = "系统管理员"
@ -32,7 +29,7 @@ ROLE_TRADER = "交易员"
# ============================================================ 一、认证：解析 bshop 返回 # ============================================================ 一、认证：解析 bshop 返回
def parse_bshop_login(resp_json: dict) -> dict: def parse_bshop_login(resp_json: dict) -> dict:
"""把 bshop 登录返回解析成统一结构（不判角色，角色交给 role_from_permissions）。 """把 bshop 登录返回解析成统一结构（不判角色，角色交给 roles_from_permissions）。
入参是已经 json 解析好的字典（真正发 HTTP 的是 call_bshop_login）。 入参是已经 json 解析好的字典（真正发 HTTP 的是 call_bshop_login）。
成功: {"ok": True, "phone", "username", "permissions": [...], "orgs": [...]} 成功: {"ok": True, "phone", "username", "permissions": [...], "orgs": [...]}
@ -83,28 +80,35 @@ def call_bshop_login(login_url: str, phone: str, password: str, timeout: int = 1
return parse_bshop_login(body) return parse_bshop_login(body)
# ============================================================ 二、角色判定 # ============================================================ 二、角色判定（可多角色）
def _markers(csv: str) -> set: def _markers(csv: str) -> set:
"""把 "pms:admin, band:yhyqx" 这类逗号分隔配置切成集合。""" """把 "pms:admin, band:yhyqx" 这类逗号分隔配置切成集合。"""
return {t.strip() for t in (csv or "").split(",") if t.strip()} return {t.strip() for t in (csv or "").split(",") if t.strip()}
def role_from_permissions(permissions, admin_perms_csv: str, def roles_from_permissions(permissions, admin_perms_csv: str,
trader_perms_csv: str = "") -> Optional[str]: trader_perms_csv: str = "") -> List[str]:
"""按权限清单判角色。 """返回这个人持有的**全部**角色（一个人可以同时是系统管理员和交易员）。
- 命中任一「管理员标记」权限 -> 系统管理员 - 系统管理员：权限里含 PMS_ADMIN_PERMS 任一条。
- 否则若配置了「交易员准入」且一条都不占 -> None（无权进入） - 交易员：未配 PMS_TRADER_PERMS（留空=人人都是交易员），或权限里含其任一条。
- 否则 -> 交易员 两个角色都不占 -> 空列表，代表无权进入 PMS。
permissions 是 "system_key:menu_key" 字符串列表（parse_bshop_login 已归一）。 permissions 是 "system_key:menu_key" 字符串列表（parse_bshop_login 已归一）。
返回顺序固定为 [系统管理员, 交易员] 中此人占到的那些。
""" """
have = set(permissions or []) have = set(permissions or [])
admin = _markers(admin_perms_csv) admin = _markers(admin_perms_csv)
trader = _markers(trader_perms_csv) trader = _markers(trader_perms_csv)
roles: List[str] = []
if admin & have: if admin & have:
return ROLE_ADMIN roles.append(ROLE_ADMIN)
if trader and not (trader & have): if (not trader) or (trader & have):
return None roles.append(ROLE_TRADER)
return ROLE_TRADER return roles
def is_admin(roles) -> bool:
"""会话/前端统一用这个判管理员，避免各处各写一遍。"""
return ROLE_ADMIN in (roles or [])
# ============================================================ 三、会话票签验 # ============================================================ 三、会话票签验
@ -117,7 +121,7 @@ def _b64d(s: str) -> bytes:
def sign_session(payload: dict, secret: str) -> str: def sign_session(payload: dict, secret: str) -> str:
"""把 {phone, username, role, exp} 签成会话票字符串。""" """把 {phone, username, roles, exp} 签成会话票字符串。"""
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"), body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"),
sort_keys=True).encode("utf-8") sort_keys=True).encode("utf-8")
b = _b64e(body) b = _b64e(body)
@ -125,12 +129,13 @@ def sign_session(payload: dict, secret: str) -> str:
return b + "." + _b64e(sig) return b + "." + _b64e(sig)
def make_session(phone: str, username: str, role: str, secret: str, def make_session(phone: str, username: str, roles, secret: str,
ttl_hours: int, now: Optional[float] = None) -> str: ttl_hours: int, now: Optional[float] = None) -> str:
"""签发一张带过期时间的会话票。now 可注入，便于单测。""" """签发一张带过期时间的会话票。roles 是角色列表。now 可注入，便于单测。"""
base = time.time() if now is None else now base = time.time() if now is None else now
exp = int(base + int(ttl_hours) * 3600) exp = int(base + int(ttl_hours) * 3600)
return sign_session({"phone": phone, "username": username, "role": role, "exp": exp}, secret) return sign_session({"phone": phone, "username": username,
"roles": list(roles), "exp": exp}, secret)
def verify_session(cookie: str, secret: str, now: Optional[float] = None) -> Optional[dict]: def verify_session(cookie: str, secret: str, now: Optional[float] = None) -> Optional[dict]:

16
app/web/main.py
View File

@ -71,8 +71,8 @@ async def _auth_guard(request: Request, call_next):
return JSONResponse({"ok": False, "error": "未登录或登录已过期", "auth": "login_required"}, return JSONResponse({"ok": False, "error": "未登录或登录已过期", "auth": "login_required"},
status_code=401) status_code=401)
if (authmod.authz_decision(request.method, path) == "admin" if (authmod.authz_decision(request.method, path) == "admin"
and sess.get("role") != authmod.ROLE_ADMIN): and not authmod.is_admin(sess.get("roles"))):
return JSONResponse({"ok": False, "error": "无权限: 该操作需要交易系统管理员", "auth": "forbidden"}, return JSONResponse({"ok": False, "error": "无权限: 该操作需要系统管理员", "auth": "forbidden"},
status_code=403) status_code=403)
request.state.user = sess request.state.user = sess
return await call_next(request) return await call_next(request)
@ -173,16 +173,16 @@ def api_auth_login(payload: dict = Body(default=None)):
settings.PMS_BSHOP_TIMEOUT) settings.PMS_BSHOP_TIMEOUT)
if not res.get("ok"): if not res.get("ok"):
return JSONResponse({"ok": False, "error": res.get("error") or "登录失败"}, status_code=401) return JSONResponse({"ok": False, "error": res.get("error") or "登录失败"}, status_code=401)
role = authmod.role_from_permissions(res["permissions"], settings.PMS_ADMIN_PERMS, roles = authmod.roles_from_permissions(res["permissions"], settings.PMS_ADMIN_PERMS,
settings.PMS_TRADER_PERMS) settings.PMS_TRADER_PERMS)
if role is None: if not roles:
return JSONResponse({"ok": False, "error": "这个账号没有持仓管理系统的使用权限"}, status_code=403) return JSONResponse({"ok": False, "error": "这个账号没有持仓管理系统的使用权限"}, status_code=403)
if not settings.PMS_SESSION_SECRET: if not settings.PMS_SESSION_SECRET:
return JSONResponse({"ok": False, "error": "服务未配置会话密钥 PMS_SESSION_SECRET"}, status_code=503) return JSONResponse({"ok": False, "error": "服务未配置会话密钥 PMS_SESSION_SECRET"}, status_code=503)
token = authmod.make_session(res["phone"], res["username"], role, token = authmod.make_session(res["phone"], res["username"], roles,
settings.PMS_SESSION_SECRET, settings.PMS_SESSION_TTL_HOURS) settings.PMS_SESSION_SECRET, settings.PMS_SESSION_TTL_HOURS)
resp = JSONResponse({"ok": True, "user": {"phone": res["phone"], "username": res["username"]}, resp = JSONResponse({"ok": True, "user": {"phone": res["phone"], "username": res["username"]},
"role": role, "orgs": res.get("orgs") or []}) "roles": roles, "orgs": res.get("orgs") or []})
resp.set_cookie(settings.PMS_SESSION_COOKIE, token, resp.set_cookie(settings.PMS_SESSION_COOKIE, token,
max_age=settings.PMS_SESSION_TTL_HOURS * 3600, max_age=settings.PMS_SESSION_TTL_HOURS * 3600,
httponly=True, samesite="lax", path="/") httponly=True, samesite="lax", path="/")
@ -202,7 +202,7 @@ def api_me(request: Request):
"""当前登录人与角色。关掉登录时回一个合成的管理员, 前端就不弹登录框也不藏东西。""" """当前登录人与角色。关掉登录时回一个合成的管理员, 前端就不弹登录框也不藏东西。"""
if not settings.PMS_AUTH_ENABLED: if not settings.PMS_AUTH_ENABLED:
return {"ok": True, "auth_enabled": False, return {"ok": True, "auth_enabled": False,
"user": {"phone": "-", "username": "本地(未开登录)"}, "role": authmod.ROLE_ADMIN} "user": {"phone": "-", "username": "本地(未开登录)"}, "roles": [authmod.ROLE_ADMIN, authmod.ROLE_TRADER]}
secret = settings.PMS_SESSION_SECRET secret = settings.PMS_SESSION_SECRET
sess = (authmod.verify_session(request.cookies.get(settings.PMS_SESSION_COOKIE), secret) sess = (authmod.verify_session(request.cookies.get(settings.PMS_SESSION_COOKIE), secret)
if secret else None) if secret else None)
@ -210,7 +210,7 @@ def api_me(request: Request):
return JSONResponse({"ok": False, "auth": "login_required"}, status_code=401) return JSONResponse({"ok": False, "auth": "login_required"}, status_code=401)
return {"ok": True, "auth_enabled": True, return {"ok": True, "auth_enabled": True,
"user": {"phone": sess.get("phone"), "username": sess.get("username")}, "user": {"phone": sess.get("phone"), "username": sess.get("username")},
"role": sess.get("role")} "roles": sess.get("roles") or []}
@app.get("/api/overview") @app.get("/api/overview")

15
app/web/static/index.html
View File

@ -432,7 +432,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
<span class="muted hd-refresh-at" style="font-size:11.5px;cursor:pointer" @click="autoRefresh=!autoRefresh" <span class="muted hd-refresh-at" style="font-size:11.5px;cursor:pointer" @click="autoRefresh=!autoRefresh"
:title="autoRefresh?'自动刷新开，点此关闭':'自动刷新关，点此开启'">{{ autoRefresh ? '自动' : '手动' }}<span v-if="lastRefresh"> · {{ lastRefresh }}</span></span> :title="autoRefresh?'自动刷新开，点此关闭':'自动刷新关，点此开启'">{{ autoRefresh ? '自动' : '手动' }}<span v-if="lastRefresh"> · {{ lastRefresh }}</span></span>
<button class="tbtn pri" :disabled="loading" @click="loadAll">{{ loading ? '刷新中…' : '刷新' }}</button> <button class="tbtn pri" :disabled="loading" @click="loadAll">{{ loading ? '刷新中…' : '刷新' }}</button>
<span class="chip hd-user" v-if="authed" :title="me.phone"><span class="dot"></span>{{ me.username || me.phone }} · {{ me.role }}</span> <span class="chip hd-user" v-if="authed" :title="me.phone"><span class="dot"></span>{{ me.username || me.phone }} · {{ rolesText }}</span>
<button class="tbtn sm" v-if="authed" @click="doLogout">退出</button> <button class="tbtn sm" v-if="authed" @click="doLogout">退出</button>
</div> </div>
@ -1965,8 +1965,9 @@ createApp({
const mode = ref('trader'); const mode = ref('trader');
// ---- 登录与角色 (2026-08 加权限) // ---- 登录与角色 (2026-08 加权限)
const authed = ref(false), authReady = ref(false); const authed = ref(false), authReady = ref(false);
const me = ref({ phone: '', username: '', role: '' }); const me = ref({ phone: '', username: '', roles: [] });
const isAdmin = computed(() => me.value.role === '交易系统管理员'); const isAdmin = computed(() => (me.value.roles || []).includes('系统管理员'));
const rolesText = computed(() => (me.value.roles || []).join(' · '));
const loginForm = reactive({ phone: '', password: '' }); const loginForm = reactive({ phone: '', password: '' });
const loginBusy = ref(false), loginErr = ref(''); const loginBusy = ref(false), loginErr = ref('');
_onUnauth = () => { authed.value = false; }; _onUnauth = () => { authed.value = false; };
@ -2404,7 +2405,7 @@ createApp({
const st = e && e.response && e.response.status; const st = e && e.response && e.response.status;
if (st === 401) { authed.value = false; return { ok: false, error: '未登录' }; } if (st === 401) { authed.value = false; return { ok: false, error: '未登录' }; }
if (st === 403) { if (st === 403) {
const m = (e.response && e.response.data && e.response.data.error) || '无权限: 需要交易系统管理员'; const m = (e.response && e.response.data && e.response.data.error) || '无权限: 需要系统管理员';
ElementPlus.ElMessage.error(m); return { ok: false, error: m }; ElementPlus.ElMessage.error(m); return { ok: false, error: m };
} }
err.value = url + ' → ' + e.message; return { ok: false, error: e.message }; err.value = url + ' → ' + e.message; return { ok: false, error: e.message };
@ -2416,7 +2417,7 @@ createApp({
try { try {
const d = (await api.get('/api/me')).data || {}; const d = (await api.get('/api/me')).data || {};
if (d.ok) { if (d.ok) {
me.value = { phone: d.user.phone, username: d.user.username, role: d.role }; me.value = { phone: d.user.phone, username: d.user.username, roles: d.roles || [] };
authed.value = true; authed.value = true;
if (!isAdmin.value && mode.value === 'ops') mode.value = 'trader'; if (!isAdmin.value && mode.value === 'ops') mode.value = 'trader';
} else { authed.value = false; } } else { authed.value = false; }
@ -2440,7 +2441,7 @@ createApp({
async function doLogout() { async function doLogout() {
try { await api.post('/api/auth/logout', {}); } catch (e) {} try { await api.post('/api/auth/logout', {}); } catch (e) {}
authed.value = false; authed.value = false;
me.value = { phone: '', username: '', role: '' }; me.value = { phone: '', username: '', roles: [] };
} }
async function loadParams() { async function loadParams() {
@ -3190,7 +3191,7 @@ createApp({
msgLatest, msgLevelClass, setMsgKind, msgGo, msgLatest, msgLevelClass, setMsgKind, msgGo,
sigTab, srcOpen, toggleSrc, sigBriefHeld, sigBriefOther, sigTab, srcOpen, toggleSrc, sigBriefHeld, sigBriefOther,
lFold, rFold, railOpen, foldRail, closeRail, openRailSide, lFold, rFold, railOpen, foldRail, closeRail, openRailSide,
authed, authReady, me, isAdmin, loginForm, loginBusy, loginErr, doLogin, doLogout, authed, authReady, me, isAdmin, rolesText, loginForm, loginBusy, loginErr, doLogin, doLogout,
sectClosed, toggleSect }; sectClosed, toggleSect };
} }
}).use(ElementPlus).mount('#app'); }).use(ElementPlus).mount('#app');

4
config/settings.py
View File

@ -50,7 +50,7 @@ class Settings(BaseSettings):
PMS_WEB_PORT: int = 38100 PMS_WEB_PORT: int = 38100
# ========================================================= # =========================================================
# 一之二、登录与权限 (PMS 自带登录 + 两个角色: 交易系统管理员 / 交易员) # 一之二、登录与权限 (PMS 自带登录 + 两个角色: 系统管理员 / 交易员)
# ========================================================= # =========================================================
# 认证接第三方登录系统 bshop: 手机号+密码, 登录成功的返回里直接带这个人的权限清单。 # 认证接第三方登录系统 bshop: 手机号+密码, 登录成功的返回里直接带这个人的权限清单。
# 角色从返回的 permissions 里认「管理员标记」权限得出; 会话用 PMS 自签票 (HMAC) 维持, # 角色从返回的 permissions 里认「管理员标记」权限得出; 会话用 PMS 自签票 (HMAC) 维持,
@ -62,7 +62,7 @@ class Settings(BaseSettings):
PMS_SESSION_COOKIE: str = "pms_session" # 会话 cookie 名 PMS_SESSION_COOKIE: str = "pms_session" # 会话 cookie 名
PMS_BSHOP_LOGIN_URL: str = "https://bshop.bmbs.tech/authProxy/api/auth/login" # 第三方登录接口 PMS_BSHOP_LOGIN_URL: str = "https://bshop.bmbs.tech/authProxy/api/auth/login" # 第三方登录接口
PMS_BSHOP_TIMEOUT: int = 10 # 调 bshop 登录的超时 (秒) PMS_BSHOP_TIMEOUT: int = 10 # 调 bshop 登录的超时 (秒)
PMS_ADMIN_PERMS: str = "pms:admin" # 「交易系统管理员」标记权限, "system_key:menu_key", 逗号分隔可多条 PMS_ADMIN_PERMS: str = "pms:admin" # 「系统管理员」标记权限, "system_key:menu_key", 逗号分隔可多条
PMS_TRADER_PERMS: str = "" # 进入 PMS 至少要占的权限; 留空=任何能登录的人都算交易员 PMS_TRADER_PERMS: str = "" # 进入 PMS 至少要占的权限; 留空=任何能登录的人都算交易员
# ========================================================= # =========================================================

2
scripts/run_tests.py
View File

@ -33,7 +33,7 @@
名额/暂停买入/接力全链)/真跑写一跳摘要而预演不写/ 名额/暂停买入/接力全链)/真跑写一跳摘要而预演不写/
判分脚本聚合与对照分组 (41 例) 判分脚本聚合与对照分组 (41 例)
test_batch18_units.py 登录与权限: 角色判定/会话票签验/bshop 返回解析/ test_batch18_units.py 登录与权限: 角色判定/会话票签验/bshop 返回解析/
接口鉴权(写操作默认拒绝) (14 例) 接口鉴权(维护类归管理员) (16 例)
test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (58 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (58 例)
共 560 例 共 560 例
任一子集失败即整体失败 (退出码 1)。 任一子集失败即整体失败 (退出码 1)。

83
scripts/test_batch18_units.py
View File

@ -1,15 +1,12 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
第十八批: 登录与权限 (系统管理员 / 交易员) 的纯逻辑 —— 不联网、不连库 第十八批: 登录与权限 (系统管理员 / 交易员, 可同时持有) 的纯逻辑 —— 不联网、不连库
===================================================================== =====================================================================
本批只测 app/web/auth.py 里的纯函数, 不导入 app.web.main (那会拉起 fastapi 与一堆 只测 app/web/auth.py 里的纯函数, 不导入 app.web.main。覆盖四件事:
基础设施依赖)。覆盖四件事: 1. 角色判定 roles_from_permissions: 返回**角色集合**, 一个人可同时是管理员和交易员;
1. 角色判定 role_from_permissions: 管理员标记命中/未中、交易员准入闸放行与拦截; 2. 会话票 make_session / verify_session: roles 列表原样签发-校验, 篡改/过期/换密钥/空票一律拒;
2. 会话票 make_session / verify_session: 签发-校验往返, 篡改/过期/换密钥/空密钥一律拒; 3. bshop 返回解析 parse_bshop_login: 喂真实样例抽字段, code!=200 判失败;
3. bshop 返回解析 parse_bshop_login: 喂真实样例抽出手机号/用户名/权限/组织, code!=200 判失败; 4. 接口鉴权 authz_decision + is_public_path: 只有运维维护类接口归管理员, 其余登录人都放行。
4. 接口鉴权 authz_decision + is_public_path: 分工按前端两块视图划分 —— 交易员工作台上能点到
的都归交易员 (下单/撤单/拍板/策略/参数微调/大盘冷热采纳), 只有运维视图里的系统维护
操作 (ops/* 、ws-channel、industry 导入、人工补录提议) 归管理员; 公开白名单放行首页与登录接口。
运行: python scripts/test_batch18_units.py 运行: python scripts/test_batch18_units.py
""" """
import os import os
@ -30,7 +27,6 @@ def case(name):
return deco return deco
# 与用户给的 bshop 登录成功返回同构的样例 (裁剪了权限条数, 字段与层级一致)
SAMPLE = { SAMPLE = {
"code": 200, "msg": "登录成功", "code": 200, "msg": "登录成功",
"data": { "data": {
@ -49,49 +45,56 @@ SAMPLE = {
} }
# ---------------------------------------------------------------- 一、角色判定 # ---------------------------------------------------------------- 一、角色判定（可多角色）
@case("角色: 命中管理员标记 => 管理员") @case("角色: 管理员标记 + 交易员开放 => 同时是管理员和交易员")
def _role_admin(): def _roles_both_open():
r = A.role_from_permissions(["band:bkld", "pms:admin"], "pms:admin", "") r = A.roles_from_permissions(["band:bkld", "pms:admin"], "pms:admin", "")
assert r == A.ROLE_ADMIN, r assert r == [A.ROLE_ADMIN, A.ROLE_TRADER], r
@case("角色: 只有普通权限 => 交易员") @case("角色: 只有普通权限 (交易员开放) => 只交易员")
def _role_trader(): def _roles_trader_only():
r = A.role_from_permissions(["band:bkld", "band:yhyqx"], "pms:admin", "") r = A.roles_from_permissions(["band:bkld"], "pms:admin", "")
assert r == A.ROLE_TRADER, r assert r == [A.ROLE_TRADER], r
@case("角色: 配了交易员准入且占其一 => 交易员") @case("角色: 配了交易员准入, 有管理员无交易员权限 => 只管理员")
def _role_trader_gate_pass(): def _roles_admin_only():
r = A.role_from_permissions(["band:ccgl"], "pms:admin", "band:ccgl,pms:-") r = A.roles_from_permissions(["pms:admin"], "pms:admin", "band:ccgl")
assert r == A.ROLE_TRADER, r assert r == [A.ROLE_ADMIN], r
@case("角色: 配了交易员准入却一条不占 => None (无权进入)") @case("角色: 配了交易员准入, 两者都占 => 同时两个角色 (用户场景)")
def _role_denied(): def _roles_both_gated():
r = A.role_from_permissions(["band:bkld"], "pms:admin", "band:ccgl,pms:-") r = A.roles_from_permissions(["pms:admin", "band:ccgl"], "pms:admin", "band:ccgl")
assert r is None, r assert r == [A.ROLE_ADMIN, A.ROLE_TRADER], r
@case("角色: 管理员标记可多条, 命中任一即算") @case("角色: 配了交易员准入却一条不占 => 空 (无权进入)")
def _role_admin_multi(): def _roles_denied():
r = A.role_from_permissions(["band:yhyqx"], "pms:admin, band:yhyqx", "") r = A.roles_from_permissions(["band:x"], "pms:admin", "band:ccgl")
assert r == A.ROLE_ADMIN, r assert r == [], r
@case("角色: is_admin 辅助判定")
def _is_admin_helper():
assert A.is_admin([A.ROLE_ADMIN, A.ROLE_TRADER]) is True
assert A.is_admin([A.ROLE_TRADER]) is False
assert A.is_admin([]) is False
# ---------------------------------------------------------------- 二、会话票 # ---------------------------------------------------------------- 二、会话票
@case("会话: 签发-校验往返, payload 字段齐全") @case("会话: roles 列表签发-校验往返, 字段齐全")
def _sess_roundtrip(): def _sess_roundtrip():
tok = A.make_session("18795373439", "老王", A.ROLE_ADMIN, "s3cr3t", 12, now=1000.0) tok = A.make_session("18795373439", "老王", [A.ROLE_ADMIN, A.ROLE_TRADER], "s3cr3t", 12, now=1000.0)
p = A.verify_session(tok, "s3cr3t", now=1001.0) p = A.verify_session(tok, "s3cr3t", now=1001.0)
assert p and p["phone"] == "18795373439" and p["username"] == "老王", p assert p and p["phone"] == "18795373439" and p["username"] == "老王", p
assert p["role"] == A.ROLE_ADMIN and p["exp"] == 1000 + 12 * 3600, p assert p["roles"] == [A.ROLE_ADMIN, A.ROLE_TRADER] and p["exp"] == 1000 + 12 * 3600, p
@case("会话: 篡改正文或签名 => None") @case("会话: 篡改正文或签名 => None")
def _sess_tampered(): def _sess_tampered():
tok = A.make_session("186", "u", A.ROLE_TRADER, "s3cr3t", 12, now=1000.0) tok = A.make_session("186", "u", [A.ROLE_TRADER], "s3cr3t", 12, now=1000.0)
b, sig = tok.split(".", 1) b, sig = tok.split(".", 1)
assert A.verify_session(b + "x." + sig, "s3cr3t", now=1001.0) is None 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 assert A.verify_session(b + "." + sig + "x", "s3cr3t", now=1001.0) is None
@ -99,7 +102,7 @@ def _sess_tampered():
@case("会话: 过期 / 换密钥 / 空密钥 / 空票 一律拒") @case("会话: 过期 / 换密钥 / 空密钥 / 空票 一律拒")
def _sess_reject(): def _sess_reject():
tok = A.make_session("186", "u", A.ROLE_TRADER, "s3cr3t", 1, now=1000.0) 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, "s3cr3t", now=1000.0 + 3600 + 1) is None, "过期没拒"
assert A.verify_session(tok, "other", now=1001.0) 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(tok, "", now=1001.0) is None, "空密钥没拒"
@ -117,11 +120,11 @@ def _parse_ok():
assert len(r["orgs"]) == 2 and r["orgs"][0]["org_name"] == "高姐", r["orgs"] assert len(r["orgs"]) == 2 and r["orgs"][0]["org_name"] == "高姐", r["orgs"]
@case("解析: 样例接角色 —— 默认标记判交易员, 复用 band:yhyqx 判管理员") @case("解析: 样例接角色 —— 默认标记只交易员, band:yhyqx 标记则管理员+交易员")
def _parse_then_role(): def _parse_then_roles():
r = A.parse_bshop_login(SAMPLE) r = A.parse_bshop_login(SAMPLE)
assert A.role_from_permissions(r["permissions"], "pms:admin", "") == A.ROLE_TRADER assert A.roles_from_permissions(r["permissions"], "pms:admin", "") == [A.ROLE_TRADER]
assert A.role_from_permissions(r["permissions"], "band:yhyqx", "") == A.ROLE_ADMIN assert A.roles_from_permissions(r["permissions"], "band:yhyqx", "") == [A.ROLE_ADMIN, A.ROLE_TRADER]
@case("解析: code!=200 / 缺手机号 / 非字典 => 失败") @case("解析: code!=200 / 缺手机号 / 非字典 => 失败")