token-list-share/deploy/reset-card-server.py

#!/usr/bin/env python3
"""百炼重置卡「立即使用」本地助手（只监听 127.0.0.1，由同域 nginx 反代）。

为什么需要它：用卡接口 reset-card/use 必须带控制台 Cookie，而 Cookie 绝不能进
前端，所以由本机服务代发。安全约束：
  - 只绑定 127.0.0.1，外网只能经 nginx /bailian-admin/ 到达；
  - 必须带与前端一致的 Admin Key（x-api-key），常量时间比较；
  - 用卡前先 reset-card/list 校验该卡确实属于本账户，且卡号格式合法；
  - 全局串行锁，防止重复点击造成连刷多张卡；
  - 探针本体（bailian-probe.py）永远不调用 use，用卡只允许发生在本服务。
"""

import hmac
import importlib.util
import json
import os
import re
import subprocess
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PROBE_PATH = os.environ.get(
    "BAILIAN_PROBE_PATH", "/usr/dcits/token-list-share/probe/bailian-probe.py")
APP_ENV_PATH = os.environ.get("APP_ENV_PATH", "/usr/dcits/token-list-share/.env")
PROBE_ENV_PATH = os.environ.get("BAILIAN_PROBE_ENV", "/etc/bailian-probe.env")
BIND = os.environ.get("RESET_CARD_BIND", "127.0.0.1")
PORT = int(os.environ.get("RESET_CARD_PORT", "18082"))

API_USE = "zeldaHttp.apikeyMgr./tokenplan/personal/api/v2/reset-card/use"
CARD_NO_RE = re.compile(r"^[a-fA-F0-9]{16,64}$")

_use_lock = threading.Lock()


def load_probe():
    spec = importlib.util.spec_from_file_location("bailian_probe", PROBE_PATH)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


probe = load_probe()
probe.load_env_file(PROBE_ENV_PATH)


def read_admin_key():
    """从应用 .env 读 VITE_ADMIN_API_KEY（与烤进前端 JS 的是同一把）。"""
    try:
        with open(APP_ENV_PATH, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line.startswith("VITE_ADMIN_API_KEY="):
                    return line.split("=", 1)[1].strip().strip('"').strip("'")
    except OSError:
        pass
    return ""


def gateway_context():
    cookie = os.environ.get("BAILIAN_COOKIE", "")
    if not cookie:
        raise PermissionError("服务器未配置 BAILIAN_COOKIE")
    variant = probe.VARIANTS["cn-personal"]
    sec = probe.resolve_sec_token(cookie, variant)
    return cookie, variant, sec


def list_owned_cards():
    cookie, variant, sec = gateway_context()
    raw = probe.fetch_personal_api(
        cookie, variant, probe.API_RESET_CARDS, sec, {})
    return probe.extract_card_list(raw)


def use_card(card_no):
    """先验卡归属，再调用用卡接口，最后重跑探针刷新快照。"""
    owned = list_owned_cards()
    match = next((c for c in owned if str(c.get("cardNo") or "") == card_no), None)
    if match is None:
        raise LookupError("该卡不在当前账户的持有列表中（可能已使用或已过期）")

    cookie, variant, sec = gateway_context()
    result = probe.fetch_personal_api(
        cookie, variant, API_USE, sec, {"cardNo": card_no})

    # 用卡成功后立即刷新快照，让前端下一次读到的就是最新状态
    try:
        subprocess.run(["/usr/bin/python3", PROBE_PATH],
                       timeout=90, capture_output=True)
    except Exception:
        pass
    return {
        "cardType": match.get("cardType"),
        "result_excerpt": json.dumps(result, ensure_ascii=False)[:600],
    }


class Handler(BaseHTTPRequestHandler):
    server_version = "ResetCard/1.0"

    def _json(self, status, obj):
        body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def _authorized(self):
        key = read_admin_key()
        if not key:
            self._json(503, {"ok": False, "error": "服务端未配置 Admin Key"})
            return False
        given = self.headers.get("x-api-key", "")
        if not hmac.compare_digest(given, key):
            self._json(401, {"ok": False, "error": "未授权"})
            return False
        return True

    def do_GET(self):
        if self.path.split("?")[0] == "/health":
            self._json(200, {"ok": True})
            return
        self._json(404, {"ok": False, "error": "Not Found"})

    def do_POST(self):
        if self.path.split("?")[0] != "/use" or not self._authorized():
            if self.path.split("?")[0] != "/use":
                self._json(404, {"ok": False, "error": "Not Found"})
            return
        try:
            length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            length = 0
        if length <= 0 or length > 65536:
            self._json(400, {"ok": False, "error": "请求体不合法"})
            return
        try:
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
            card_no = str((payload or {}).get("card_no") or "")
        except Exception:
            self._json(400, {"ok": False, "error": "请求体不是合法 JSON"})
            return
        if not CARD_NO_RE.fullmatch(card_no):
            self._json(400, {"ok": False, "error": "卡号格式不合法"})
            return

        # 串行化：同一张卡的并发请求只有一个能进入用卡流程
        if not _use_lock.acquire(blocking=False):
            self._json(409, {"ok": False, "error": "已有用卡请求在处理中"})
            return
        try:
            info = use_card(card_no)
        except probe.ProbeError as e:
            kind = 401 if e.kind == "unauthorized" else 400
            self._json(kind, {"ok": False, "error": str(e), "error_kind": e.kind})
            return
        except (PermissionError, LookupError) as e:
            self._json(400, {"ok": False, "error": str(e)})
            return
        except Exception as e:
            self._json(502, {"ok": False, "error": f"{type(e).__name__}: {e}"})
            return
        finally:
            _use_lock.release()
        self._json(200, {"ok": True, **info})

    def log_message(self, fmt, *args):
        # 不记录请求体/卡号；journalctl 里只留方法与状态码
        super().log_message("%s %s", self.command, self.path.split("?")[0])


if __name__ == "__main__":
    httpd = ThreadingHTTPServer((BIND, PORT), Handler)
    print(f"reset-card-server listening on {BIND}:{PORT}", flush=True)
    httpd.serve_forever()