对接中控：只读持仓快照 API、SSO 打开链与 iframe 祖先配置。

Co-authored-by: Cursor <cursoragent@cursor.com>

.env.example

@@ -274,3 +274,12 @@ WECHAT_TIMEOUT_SECONDS=10
 
 APP_TIMEZONE=Asia/Shanghai
 # TRADING_DAY_RESET_HOUR 现在表示「北京时间」整点,默认 8 点起算新交易日;开仓整点限制见 TRADING_DAY_RESET_OPEN_GUARD_ENABLED
+
+# =============================================================================
+# 中控 Bridge(crypto_okx_hub 拉取持仓 / SSO 打开)
+# =============================================================================
+# 与中控登记的 Token 一致;留空则 /api/hub/* 拒绝访问
+# HUB_BRIDGE_TOKEN=
+# 允许嵌入本站的中控 Origin(空格分隔);例: http://192.168.1.10:5100 https://hub.example.com
+# 设为 * 允许任意父页(仅内网调试)
+# HUB_EMBED_FRAME_ANCESTORS=

app.py

@@ -7580,6 +7580,9 @@ from lib.options.options_review_register import install_options_review
 
 install_options_review(app, _REPO_ROOT, app_module=sys.modules[__name__])
 
+from lib.hub.hub_bridge_register import register_hub_bridge_routes
+
+register_hub_bridge_routes(app, sys.modules[__name__])
 
 # 独立部署无 hub_bridge:须自行注册 /embed 壳,否则登录后跳转 /embed 会 404
 from lib.instance.instance_embed_lib import register_embed_routes

lib/hub/__init__.py

@@ -0,0 +1 @@
+"""Hub bridge package (中控对接)."""

lib/hub/hub_bridge_auth.py

@@ -0,0 +1,96 @@
+"""中控 Bridge 鉴权与 SSO(短期登录链)."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import os
+import time
+from typing import Optional
+from urllib.parse import urlencode
+
+
+def hub_bridge_token() -> str:
+    return (os.getenv("HUB_BRIDGE_TOKEN") or "").strip()
+
+
+def extract_hub_token_from_request(headers) -> str:
+    raw = (headers.get("X-Hub-Token") or "").strip()
+    if raw:
+        return raw
+    auth = (headers.get("Authorization") or "").strip()
+    if auth.lower().startswith("bearer "):
+        return auth[7:].strip()
+    return ""
+
+
+def hub_bridge_token_ok(headers) -> bool:
+    expected = hub_bridge_token()
+    if not expected:
+        return False
+    got = extract_hub_token_from_request(headers)
+    if not got:
+        return False
+    return hmac.compare_digest(got, expected)
+
+
+def make_sso_token(bridge_token: str, *, ttl_sec: int = 120) -> str:
+    """exp.hex_hmac — 中控与客户端共用 bridge_token 签名."""
+    exp = int(time.time()) + max(30, int(ttl_sec))
+    payload = str(exp)
+    sig = hmac.new(
+        bridge_token.encode("utf-8"),
+        payload.encode("utf-8"),
+        hashlib.sha256,
+    ).hexdigest()
+    return f"{payload}.{sig}"
+
+
+def verify_sso_token(token: str, bridge_token: Optional[str] = None) -> bool:
+    expected = (bridge_token if bridge_token is not None else hub_bridge_token()).strip()
+    raw = (token or "").strip()
+    if not expected or not raw or "." not in raw:
+        return False
+    payload, sig = raw.rsplit(".", 1)
+    try:
+        exp = int(payload)
+    except ValueError:
+        return False
+    if exp < int(time.time()):
+        return False
+    want = hmac.new(
+        expected.encode("utf-8"),
+        payload.encode("utf-8"),
+        hashlib.sha256,
+    ).hexdigest()
+    return hmac.compare_digest(sig, want)
+
+
+def build_embed_open_url(
+    base_url: str,
+    *,
+    bridge_token: str,
+    embed_path: str = "/embed",
+    ttl_sec: int = 120,
+) -> str:
+    """中控生成带 SSO 的客户端打开地址."""
+    root = (base_url or "").rstrip("/")
+    path = embed_path or "/embed"
+    if not path.startswith("/"):
+        path = "/" + path
+    token = make_sso_token(bridge_token, ttl_sec=ttl_sec)
+    q = urlencode({"token": token, "next": path})
+    return f"{root}/hub_sso?{q}"
+
+
+def frame_ancestors_header_value() -> Optional[str]:
+    """
+    HUB_EMBED_FRAME_ANCESTORS: 空格分隔 Origin,如 https://hub.local http://192.168.1.10:5100
+    未配置则不改响应头(保持现状);配 * 则允许任意父页(仅内网慎用).
+    """
+    raw = (os.getenv("HUB_EMBED_FRAME_ANCESTORS") or "").strip()
+    if not raw:
+        return None
+    if raw == "*":
+        return "*"
+    parts = [p.strip() for p in raw.split() if p.strip()]
+    return " ".join(parts) if parts else None

lib/hub/hub_bridge_register.py

@@ -0,0 +1,134 @@
+"""中控只读 Bridge 路由 + SSO 登录入口."""
+from __future__ import annotations
+
+from typing import Any
+
+from flask import Flask, jsonify, redirect, request, session
+
+from lib.hub.hub_bridge_auth import (
+    frame_ancestors_header_value,
+    hub_bridge_token_ok,
+    verify_sso_token,
+)
+
+
+def _safe_next_path(raw: str, default: str = "/embed") -> str:
+    p = (raw or "").strip() or default
+    if not p.startswith("/"):
+        p = "/" + p
+    if "://" in p or p.startswith("//"):
+        return default
+    return p
+
+
+def _enrich_hub_protocol(snap: dict[str, Any], *, app_module: Any) -> dict[str, Any]:
+    """对齐中控设计文档 version=1 字段,同时保留原 snapshot 键."""
+    out = dict(snap or {})
+    out["version"] = 1
+    mode = "live"
+    try:
+        from lib.sim.mode_lib import is_sim_mode
+
+        get_db = getattr(app_module, "get_db", None)
+        if callable(get_db) and is_sim_mode(get_db):
+            mode = "sim"
+    except Exception:
+        mode = "live"
+    label = ""
+    try:
+        from lib.sim.mode_lib import exchange_mode_label
+
+        get_db = getattr(app_module, "get_db", None)
+        if callable(get_db):
+            label = exchange_mode_label(get_db=get_db, exchange_key="okx") or ""
+    except Exception:
+        label = ""
+    if not label:
+        label = "okx_sim" if mode == "sim" else "okx_live"
+
+    positions = out.get("positions") if isinstance(out.get("positions"), list) else []
+    slim_pos = []
+    for p in positions:
+        if not isinstance(p, dict):
+            continue
+        slim_pos.append(
+            {
+                "inst_id": p.get("inst_id"),
+                "opt_type": p.get("opt_type"),
+                "sheets": p.get("sheets") or p.get("pos"),
+                "pnl_usdt": p.get("display_pnl_usdt")
+                if p.get("display_pnl_usdt") is not None
+                else p.get("net_pnl_usdt"),
+                "mark": p.get("mark") or p.get("mark_px"),
+                "underlying": p.get("underlying"),
+                "source_label": p.get("source_label"),
+            }
+        )
+
+    out["instance"] = {
+        "label": label,
+        "mode": mode,
+        "options_enabled": bool(out.get("enabled")),
+        "account_label": out.get("account_label") or "",
+        "margin_mode": out.get("options_margin_mode") or "coin",
+        "underly": out.get("options_underly") or "ETH",
+    }
+    out["funds"] = {
+        "funding_usdt": out.get("funding_usdt"),
+        "trading_usdt": out.get("trading_usdt"),
+        "funding_usdc": out.get("funding_usdc"),
+        "trading_usdc": out.get("trading_usdc"),
+    }
+    out["pnl"] = {
+        "unrealized_usdt": out.get("upl_total_usdc"),
+    }
+    out["positions_brief"] = slim_pos
+    try:
+        from lib.options.options_time_lib import options_app_now_str
+
+        out["ts"] = options_app_now_str()
+    except Exception:
+        from datetime import datetime
+
+        out["ts"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    return out
+
+
+def register_hub_bridge_routes(app: Flask, app_module: Any) -> None:
+    @app.route("/api/hub/options/snapshot")
+    def api_hub_options_snapshot():
+        if not hub_bridge_token_ok(request.headers):
+            return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
+        hub_ctx = app.config.get("HUB_CTX") or {}
+        fn = hub_ctx.get("options_snapshot_fn")
+        if not callable(fn):
+            return jsonify(
+                {"ok": False, "msg": "期权快照未注册", "version": 1, "enabled": False}
+            ), 503
+        try:
+            snap = fn() or {}
+        except Exception as e:
+            return jsonify({"ok": False, "msg": str(e), "version": 1, "enabled": True}), 500
+        return jsonify(_enrich_hub_protocol(snap, app_module=app_module))
+
+    @app.route("/hub_sso")
+    def hub_sso_login():
+        token = (request.args.get("token") or "").strip()
+        next_path = _safe_next_path(request.args.get("next") or "/embed")
+        if not verify_sso_token(token):
+            return (
+                "<!doctype html><meta charset=utf-8><title>SSO 失败</title>"
+                "<p>中控登录链无效或已过期,请从中控重新打开.</p>",
+                403,
+            )
+        session["logged_in"] = True
+        session.permanent = True
+        return redirect(next_path)
+
+    @app.after_request
+    def _hub_frame_ancestors(resp):
+        ancestors = frame_ancestors_header_value()
+        if ancestors:
+            resp.headers["Content-Security-Policy"] = f"frame-ancestors {ancestors}"
+            resp.headers.pop("X-Frame-Options", None)
+        return resp

tests/test_hub_bridge_auth.py

@@ -0,0 +1,32 @@
+"""中控 Bridge SSO / Token 冒烟."""
+from __future__ import annotations
+
+import unittest
+
+from lib.hub.hub_bridge_auth import (
+    make_sso_token,
+    verify_sso_token,
+    build_embed_open_url,
+)
+
+
+class HubBridgeAuthTests(unittest.TestCase):
+    def test_sso_roundtrip(self):
+        tok = make_sso_token("secret-token", ttl_sec=60)
+        self.assertTrue(verify_sso_token(tok, "secret-token"))
+        self.assertFalse(verify_sso_token(tok, "other"))
+        self.assertFalse(verify_sso_token("bad", "secret-token"))
+
+    def test_embed_open_url(self):
+        url = build_embed_open_url(
+            "https://okx.example.com",
+            bridge_token="abc",
+            embed_path="/embed",
+        )
+        self.assertTrue(url.startswith("https://okx.example.com/hub_sso?"))
+        self.assertIn("token=", url)
+        self.assertIn("next=%2Fembed", url)
+
+
+if __name__ == "__main__":
+    unittest.main()