216 lines
9.2 KiB
Python
216 lines
9.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
PMS 登录与权限 —— 认证解析 / 角色判定 / 会话票签验 / 接口鉴权判定（纯函数）
|
||
=======================================================================
|
||
方案见「PMS 加登录与两个角色（系统管理员 / 交易员）」。本模块刻意只放纯函数：
|
||
不连库、几乎不联网，喂进去字典或字符串就能算，便于单测
|
||
（scripts/test_batch18_units.py）。唯一碰网络的是 call_bshop_login 一个函数，
|
||
它把真正的 HTTP 调用隔离出来，其余全部可离线测。
|
||
|
||
三块职责：
|
||
一、认证：解析第三方登录系统 bshop 的登录返回（parse_bshop_login）。
|
||
二、角色：从 bshop 返回的权限清单里认出系统管理员 / 交易员（role_from_permissions）。
|
||
三、会话与鉴权：自签会话票（make_session / verify_session），以及按「方法 + 路径」
|
||
判这个请求要什么角色（authz_decision）。
|
||
|
||
会话票格式：base64url( json({phone, username, role, exp}) ) + "." + base64url( hmac_sha256 )
|
||
—— 只用标准库 hmac / hashlib，不引新依赖。exp 是 Unix 秒。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import time
|
||
from typing import Optional
|
||
|
||
# 角色常量：既是写进会话票的字面值，也是回给前端、供前端判断的字面值。
|
||
ROLE_ADMIN = "系统管理员"
|
||
ROLE_TRADER = "交易员"
|
||
|
||
|
||
# ============================================================ 一、认证：解析 bshop 返回
|
||
def parse_bshop_login(resp_json: dict) -> dict:
|
||
"""把 bshop 登录返回解析成统一结构（不判角色，角色交给 role_from_permissions）。
|
||
|
||
入参是已经 json 解析好的字典（真正发 HTTP 的是 call_bshop_login）。
|
||
成功: {"ok": True, "phone", "username", "permissions": [...], "orgs": [...]}
|
||
失败: {"ok": False, "error": "..."}
|
||
permissions 每条归一成 "system_key:menu_key" 字符串，方便后面比对。
|
||
"""
|
||
if not isinstance(resp_json, dict):
|
||
return {"ok": False, "error": "登录服务返回的不是 JSON 对象"}
|
||
if resp_json.get("code") != 200:
|
||
return {"ok": False, "error": str(resp_json.get("msg") or "登录失败")}
|
||
data = resp_json.get("data") or {}
|
||
user = data.get("user") or {}
|
||
phone = str(user.get("phone") or "").strip()
|
||
username = str(user.get("username") or phone or "").strip()
|
||
if not phone:
|
||
return {"ok": False, "error": "登录返回里没有手机号"}
|
||
perms = []
|
||
for p in (data.get("permissions") or []):
|
||
sk = str((p or {}).get("system_key") or "").strip()
|
||
mk = str((p or {}).get("menu_key") or "").strip()
|
||
if sk:
|
||
perms.append(sk + ":" + mk)
|
||
orgs = []
|
||
for o in (data.get("orgs") or []):
|
||
orgs.append({"id": (o or {}).get("id"), "org_name": (o or {}).get("org_name")})
|
||
return {"ok": True, "phone": phone, "username": username,
|
||
"permissions": perms, "orgs": orgs}
|
||
|
||
|
||
def call_bshop_login(login_url: str, phone: str, password: str, timeout: int = 10) -> dict:
|
||
"""真正发 HTTP 调 bshop 登录。**本模块里唯一碰网络的函数。**
|
||
用 multipart/form-data 提交（与对方 curl 一致）。网络或解析失败也回统一的
|
||
{"ok": False, "error"} 结构，绝不抛出去。
|
||
"""
|
||
import requests # 局部导入：纯函数单测既不装也不连任何东西
|
||
try:
|
||
r = requests.post(
|
||
login_url,
|
||
files={"phone": (None, str(phone)), "password": (None, str(password))},
|
||
timeout=timeout,
|
||
)
|
||
except Exception as e:
|
||
return {"ok": False, "error": "连不上登录服务: " + type(e).__name__ + ": " + str(e)}
|
||
try:
|
||
body = r.json()
|
||
except Exception:
|
||
return {"ok": False, "error": "登录服务返回非 JSON (HTTP " + str(r.status_code) + ")"}
|
||
return parse_bshop_login(body)
|
||
|
||
|
||
# ============================================================ 二、角色判定
|
||
def _markers(csv: str) -> set:
|
||
"""把 "pms:admin, band:yhyqx" 这类逗号分隔配置切成集合。"""
|
||
return {t.strip() for t in (csv or "").split(",") if t.strip()}
|
||
|
||
|
||
def role_from_permissions(permissions, admin_perms_csv: str,
|
||
trader_perms_csv: str = "") -> Optional[str]:
|
||
"""按权限清单判角色。
|
||
- 命中任一「管理员标记」权限 -> 系统管理员
|
||
- 否则若配置了「交易员准入」且一条都不占 -> None（无权进入）
|
||
- 否则 -> 交易员
|
||
permissions 是 "system_key:menu_key" 字符串列表（parse_bshop_login 已归一）。
|
||
"""
|
||
have = set(permissions or [])
|
||
admin = _markers(admin_perms_csv)
|
||
trader = _markers(trader_perms_csv)
|
||
if admin & have:
|
||
return ROLE_ADMIN
|
||
if trader and not (trader & have):
|
||
return None
|
||
return ROLE_TRADER
|
||
|
||
|
||
# ============================================================ 三、会话票签验
|
||
def _b64e(raw: bytes) -> str:
|
||
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
|
||
|
||
|
||
def _b64d(s: str) -> bytes:
|
||
return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
|
||
|
||
|
||
def sign_session(payload: dict, secret: str) -> str:
|
||
"""把 {phone, username, role, exp} 签成会话票字符串。"""
|
||
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"),
|
||
sort_keys=True).encode("utf-8")
|
||
b = _b64e(body)
|
||
sig = hmac.new(secret.encode("utf-8"), b.encode("ascii"), hashlib.sha256).digest()
|
||
return b + "." + _b64e(sig)
|
||
|
||
|
||
def make_session(phone: str, username: str, role: str, secret: str,
|
||
ttl_hours: int, now: Optional[float] = None) -> str:
|
||
"""签发一张带过期时间的会话票。now 可注入，便于单测。"""
|
||
base = time.time() if now is None else now
|
||
exp = int(base + int(ttl_hours) * 3600)
|
||
return sign_session({"phone": phone, "username": username, "role": role, "exp": exp}, secret)
|
||
|
||
|
||
def verify_session(cookie: str, secret: str, now: Optional[float] = None) -> Optional[dict]:
|
||
"""验会话票。签名不对、过期、格式坏、空票或空密钥一律返回 None；否则返回 payload。"""
|
||
if not cookie or not secret:
|
||
return None
|
||
try:
|
||
b, sig = cookie.split(".", 1)
|
||
except ValueError:
|
||
return None
|
||
expect = hmac.new(secret.encode("utf-8"), b.encode("ascii"), hashlib.sha256).digest()
|
||
try:
|
||
got = _b64d(sig)
|
||
except Exception:
|
||
return None
|
||
if not hmac.compare_digest(expect, got):
|
||
return None
|
||
try:
|
||
payload = json.loads(_b64d(b).decode("utf-8"))
|
||
except Exception:
|
||
return None
|
||
base = time.time() if now is None else now
|
||
if int(payload.get("exp") or 0) < base:
|
||
return None
|
||
return payload
|
||
|
||
|
||
# ============================================================ 四、接口鉴权判定
|
||
# 分工按前端两块视图划分（用户 2026-08 拍板）：交易员工作台（前端 mode=trader）上能点到
|
||
# 的都归交易员；只有运维视图（mode=ops）里的系统维护操作归系统管理员。所以这里反过来只
|
||
# 登记「只给管理员」的那一小撮运维接口，其余已登录的人（交易员）默认放行 —— 交易员的能力
|
||
# 跟着他那一页走：日常下单、撤单、拍板、策略、参数微调、大盘冷热采纳，都算交易员的活。
|
||
# 只给管理员的接口族（前缀匹配，任何方法都算）：
|
||
_ADMIN_PREFIX = [
|
||
("api", "ops"), # 回放 / 对账 / 日终 / 日报 / 择时扫描 / plan-refresh 等全部运维操作
|
||
("api", "ws-channel"), # ws 直连通道管理
|
||
("api", "industry"), # 行业划分数据导入
|
||
]
|
||
# 只给管理员的单个接口（精确匹配方法与整条路径）：
|
||
_ADMIN_EXACT = [
|
||
("POST", ("api", "proposals")), # 人工补录提议：影子期造数调试口，没有交易员入口
|
||
]
|
||
|
||
|
||
def _seg(path: str) -> tuple:
|
||
return tuple(s for s in (path or "").strip("/").split("/") if s != "")
|
||
|
||
|
||
def _match(pattern: tuple, segs: tuple) -> bool:
|
||
if len(pattern) != len(segs):
|
||
return False
|
||
return all(p == "*" or p == s for p, s in zip(pattern, segs))
|
||
|
||
|
||
def authz_decision(method: str, path: str) -> str:
|
||
"""判这个请求所需的最低角色，返回 "admin"（要系统管理员）或 "any"（任何登录人可）。
|
||
规则：命中「只给管理员」的运维接口族，或那条人工补录提议的调试口，就要管理员；其余
|
||
已登录的人都放行。只对「已登录、且不在公开白名单」的请求调用（白名单在 is_public_path）。
|
||
"""
|
||
method = (method or "").upper()
|
||
segs = _seg(path)
|
||
for m, pat in _ADMIN_EXACT:
|
||
if m == method and _match(pat, segs):
|
||
return "admin"
|
||
for pref in _ADMIN_PREFIX:
|
||
if len(segs) >= len(pref) and tuple(segs[:len(pref)]) == pref:
|
||
return "admin"
|
||
return "any"
|
||
|
||
|
||
# 公开路径：不需要登录也放行。首页 / 必须放行，否则登录框自己都加载不出来。
|
||
_PUBLIC_EXACT = {"/", "/health", "/favicon.ico",
|
||
"/api/auth/login", "/api/auth/logout", "/api/me"}
|
||
|
||
|
||
def is_public_path(path: str) -> bool:
|
||
p = (path or "/")
|
||
if p in _PUBLIC_EXACT or (p.rstrip("/") or "/") in _PUBLIC_EXACT:
|
||
return True
|
||
if p == "/static" or p.startswith("/static/"):
|
||
return True
|
||
return False
|