diff --git a/.env.example b/.env.example
index dba96c8..f1f220e 100644
--- a/.env.example
+++ b/.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=
diff --git a/app.py b/app.py
index b2141f4..eeebf14 100644
--- a/app.py
+++ b/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
diff --git a/lib/hub/__init__.py b/lib/hub/__init__.py
new file mode 100644
index 0000000..46171b5
--- /dev/null
+++ b/lib/hub/__init__.py
@@ -0,0 +1 @@
+"""Hub bridge package (中控对接)."""
diff --git a/lib/hub/hub_bridge_auth.py b/lib/hub/hub_bridge_auth.py
new file mode 100644
index 0000000..493d53e
--- /dev/null
+++ b/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
diff --git a/lib/hub/hub_bridge_register.py b/lib/hub/hub_bridge_register.py
new file mode 100644
index 0000000..658ea6a
--- /dev/null
+++ b/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 (
+ "SSO 失败"
+ "中控登录链无效或已过期,请从中控重新打开.
",
+ 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
diff --git a/tests/test_hub_bridge_auth.py b/tests/test_hub_bridge_auth.py
new file mode 100644
index 0000000..d90658b
--- /dev/null
+++ b/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()