tradingSystem/app/web/auth.py

# -*- coding: utf-8 -*-
"""
PMS 登录与权限 —— 认证解析 / 角色判定 / 会话票签验 / 接口鉴权判定（纯函数）
=======================================================================
方案见「PMS 加登录与两个角色（系统管理员 / 交易员）」。本模块刻意只放纯函数：
不连库、几乎不联网，喂进去字典或字符串就能算，便于单测
（scripts/test_batch18_units.py）。唯一碰网络的是 call_bshop_login 一个函数，
它把真正的 HTTP 调用隔离出来，其余全部可离线测。

一个人可以同时持有多个角色（既是系统管理员又是交易员），所以角色一律按「集合」处理，
不折叠成单一角色。

会话票格式：base64url( json({phone, username, roles, exp}) ) + "." + base64url( hmac_sha256 )
  —— 只用标准库 hmac / hashlib，不引新依赖。exp 是 Unix 秒。roles 是角色字符串列表。
"""
from __future__ import annotations

import base64
import hashlib
import hmac
import json
import time
from typing import List, Optional

# 角色常量：既是写进会话票的字面值，也是回给前端、供前端判断的字面值。
ROLE_ADMIN = "系统管理员"
ROLE_TRADER = "交易员"


# ============================================================ 一、认证：解析 bshop 返回
def parse_bshop_login(resp_json: dict) -> dict:
    """把 bshop 登录返回解析成统一结构（不判角色，角色交给 roles_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 roles_from_permissions(permissions, admin_perms_csv: str,
                           trader_perms_csv: str = "") -> List[str]:
    """返回这个人持有的**全部**角色（一个人可以同时是系统管理员和交易员）。
    - 系统管理员：权限里含 PMS_ADMIN_PERMS 任一条。
    - 交易员：未配 PMS_TRADER_PERMS（留空=人人都是交易员），或权限里含其任一条。
    两个角色都不占 -> 空列表，代表无权进入 PMS。
    permissions 是 "system_key:menu_key" 字符串列表（parse_bshop_login 已归一）。
    返回顺序固定为 [系统管理员, 交易员] 中此人占到的那些。
    """
    have = set(permissions or [])
    admin = _markers(admin_perms_csv)
    trader = _markers(trader_perms_csv)
    roles: List[str] = []
    if admin & have:
        roles.append(ROLE_ADMIN)
    if (not trader) or (trader & have):
        roles.append(ROLE_TRADER)
    return roles


def is_admin(roles) -> bool:
    """会话/前端统一用这个判管理员，避免各处各写一遍。"""
    return ROLE_ADMIN in (roles or [])


# ============================================================ 三、会话票签验
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, roles, 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, roles, secret: str,
                 ttl_hours: int, now: Optional[float] = None) -> str:
    """签发一张带过期时间的会话票。roles 是角色列表。now 可注入，便于单测。"""
    base = time.time() if now is None else now
    exp = int(base + int(ttl_hours) * 3600)
    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]:
    """验会话票。签名不对、过期、格式坏、空票或空密钥一律返回 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