fix(hub): SSO in hub_sso without Flask for FastAPI hub

hub_auth.py

@@ -1,29 +1,27 @@
-"""中控调用实例 API 时的鉴权；实例浏览器 SSO（复用 HUB_BRIDGE_TOKEN）。"""
+"""中控调用实例 API 时的鉴权（Flask request 头 X-Hub-Token）。SSO 见 hub_sso.py。"""
 from __future__ import annotations
 
-import base64
-import hashlib
-import hmac
-import json
 import os
-import secrets
-import threading
-import time
-from typing import Any
 
-# 中控打开实例链接有效期（秒），默认 2 小时
+from hub_sso import (
-HUB_SSO_TTL_SEC = int(os.getenv("HUB_SSO_TTL_SEC", "7200"))
+    HUB_SSO_TTL_SEC,
+    hub_bridge_token,
+    mint_hub_sso_token,
+    safe_next_path,
+    verify_hub_sso_token,
+)
 
-_used_nonces: dict[str, float] = {}
+__all__ = [
-_nonce_lock = threading.Lock()
+    "HUB_SSO_TTL_SEC",
-
+    "hub_bridge_token",
-
+    "mint_hub_sso_token",
-def hub_bridge_token() -> str:
+    "safe_next_path",
-    return (os.getenv("HUB_BRIDGE_TOKEN") or "").strip()
+    "verify_hub_sso_token",
+    "request_allowed",
+]
 
 
 def request_allowed(session_logged_in: bool, auth_disabled: bool) -> bool:
-    """各 Flask 实例 login_required / hub_bridge 用；延迟导入 flask，避免中控 FastAPI venv 强依赖 Flask。"""
     if auth_disabled or session_logged_in:
         return True
     tok = hub_bridge_token()
@@ -36,94 +34,3 @@ def request_allowed(session_logged_in: bool, auth_disabled: bool) -> bool:
     if request.headers.get("X-Hub-Token") == tok:
         return True
     return False
-
-
-def safe_next_path(raw: str | None) -> str:
-    """仅允许站内相对路径，防止开放重定向。"""
-    p = (raw or "/").strip()
-    if not p.startswith("/") or p.startswith("//"):
-        return "/"
-    if "://" in p:
-        return "/"
-    return p
-
-
-def _sso_secret() -> str:
-    return hub_bridge_token()
-
-
-def _b64url_encode(data: bytes) -> str:
-    return base64.urlsafe_b64encode(data).decode().rstrip("=")
-
-
-def _b64url_decode(data: str) -> bytes:
-    pad = "=" * (-len(data) % 4)
-    return base64.urlsafe_b64decode(data + pad)
-
-
-def _prune_used_nonces() -> None:
-    now = time.time()
-    with _nonce_lock:
-        dead = [k for k, exp in _used_nonces.items() if exp <= now]
-        for k in dead:
-            del _used_nonces[k]
-
-
-def mint_hub_sso_token(exchange_key: str, next_path: str = "/") -> str | None:
-    """签发实例浏览器 SSO token（exchange_key 与 hub_bridge install_on_app 的 exchange 一致）。"""
-    secret = _sso_secret()
-    ex = (exchange_key or "").strip().lower()
-    if not secret or not ex:
-        return None
-    payload = {
-        "ex": ex,
-        "exp": int(time.time()) + max(60, HUB_SSO_TTL_SEC),
-        "nonce": secrets.token_urlsafe(16),
-        "next": safe_next_path(next_path),
-    }
-    body = _b64url_encode(json.dumps(payload, separators=(",", ":")).encode())
-    sig = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
-    return f"{body}.{sig}"
-
-
-def verify_hub_sso_token(
-    token: str | None, expected_exchange: str
-) -> tuple[bool, str, str | None]:
-    """
-    校验 SSO token。成功返回 (True, next_path, None)；失败返回 (False, '/', 原因)。
-    单次使用：nonce 用过后在 exp 之前不可复用。
-    """
-    secret = _sso_secret()
-    expected = (expected_exchange or "").strip().lower()
-    if not secret or not expected:
-        return False, "/", "未配置 HUB_BRIDGE_TOKEN"
-    raw = (token or "").strip()
-    if "." not in raw:
-        return False, "/", "token 无效"
-    body, sig = raw.rsplit(".", 1)
-    try:
-        expect_sig = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
-        if not hmac.compare_digest(expect_sig, sig):
-            return False, "/", "签名校验失败"
-        payload = json.loads(_b64url_decode(body).decode())
-    except Exception:
-        return False, "/", "token 解析失败"
-    if not isinstance(payload, dict):
-        return False, "/", "payload 无效"
-    if str(payload.get("ex") or "").lower() != expected:
-        return False, "/", "实例不匹配"
-    try:
-        exp = int(payload.get("exp") or 0)
-    except (TypeError, ValueError):
-        return False, "/", "exp 无效"
-    if exp < int(time.time()):
-        return False, "/", "链接已过期"
-    nonce = str(payload.get("nonce") or "")
-    if not nonce:
-        return False, "/", "nonce 缺失"
-    _prune_used_nonces()
-    with _nonce_lock:
-        if nonce in _used_nonces:
-            return False, "/", "链接已使用"
-        _used_nonces[nonce] = float(exp)
-    return True, safe_next_path(str(payload.get("next") or "/")), None

hub_bridge.py

@@ -11,7 +11,8 @@ from functools import wraps
 
 from flask import current_app, get_flashed_messages, jsonify, redirect, request, session
 
-from hub_auth import request_allowed, safe_next_path, verify_hub_sso_token
+from hub_auth import request_allowed
+from hub_sso import safe_next_path, verify_hub_sso_token
 
 
 def _hub_auth_required(f):

hub_sso.py

@@ -0,0 +1,107 @@
+"""
+实例浏览器 SSO（复用 HUB_BRIDGE_TOKEN）。无 Flask 依赖，供中控 FastAPI 与各实例共用。
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import threading
+import time
+
+HUB_SSO_TTL_SEC = int(os.getenv("HUB_SSO_TTL_SEC", "7200"))
+
+_used_nonces: dict[str, float] = {}
+_nonce_lock = threading.Lock()
+
+
+def hub_bridge_token() -> str:
+    return (os.getenv("HUB_BRIDGE_TOKEN") or "").strip()
+
+
+def safe_next_path(raw: str | None) -> str:
+    p = (raw or "/").strip()
+    if not p.startswith("/") or p.startswith("//"):
+        return "/"
+    if "://" in p:
+        return "/"
+    return p
+
+
+def _sso_secret() -> str:
+    return hub_bridge_token()
+
+
+def _b64url_encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).decode().rstrip("=")
+
+
+def _b64url_decode(data: str) -> bytes:
+    pad = "=" * (-len(data) % 4)
+    return base64.urlsafe_b64decode(data + pad)
+
+
+def _prune_used_nonces() -> None:
+    now = time.time()
+    with _nonce_lock:
+        dead = [k for k, exp in _used_nonces.items() if exp <= now]
+        for k in dead:
+            del _used_nonces[k]
+
+
+def mint_hub_sso_token(exchange_key: str, next_path: str = "/") -> str | None:
+    secret = _sso_secret()
+    ex = (exchange_key or "").strip().lower()
+    if not secret or not ex:
+        return None
+    payload = {
+        "ex": ex,
+        "exp": int(time.time()) + max(60, HUB_SSO_TTL_SEC),
+        "nonce": secrets.token_urlsafe(16),
+        "next": safe_next_path(next_path),
+    }
+    body = _b64url_encode(json.dumps(payload, separators=(",", ":")).encode())
+    sig = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
+    return f"{body}.{sig}"
+
+
+def verify_hub_sso_token(
+    token: str | None, expected_exchange: str
+) -> tuple[bool, str, str | None]:
+    secret = _sso_secret()
+    expected = (expected_exchange or "").strip().lower()
+    if not secret or not expected:
+        return False, "/", "未配置 HUB_BRIDGE_TOKEN"
+    raw = (token or "").strip()
+    if "." not in raw:
+        return False, "/", "token 无效"
+    body, sig = raw.rsplit(".", 1)
+    try:
+        expect_sig = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
+        if not hmac.compare_digest(expect_sig, sig):
+            return False, "/", "签名校验失败"
+        payload = json.loads(_b64url_decode(body).decode())
+    except Exception:
+        return False, "/", "token 解析失败"
+    if not isinstance(payload, dict):
+        return False, "/", "payload 无效"
+    if str(payload.get("ex") or "").lower() != expected:
+        return False, "/", "实例不匹配"
+    try:
+        exp = int(payload.get("exp") or 0)
+    except (TypeError, ValueError):
+        return False, "/", "exp 无效"
+    if exp < int(time.time()):
+        return False, "/", "链接已过期"
+    nonce = str(payload.get("nonce") or "")
+    if not nonce:
+        return False, "/", "nonce 缺失"
+    _prune_used_nonces()
+    with _nonce_lock:
+        if nonce in _used_nonces:
+            return False, "/", "链接已使用"
+        _used_nonces[nonce] = float(exp)
+    return True, safe_next_path(str(payload.get("next") or "/")), None

manual_trading_hub/hub.py

@@ -40,7 +40,7 @@ from hub_web_auth import (
     expected_username,
     verify_credentials,
 )
-from hub_auth import HUB_SSO_TTL_SEC, mint_hub_sso_token, safe_next_path
+from hub_sso import HUB_SSO_TTL_SEC, mint_hub_sso_token, safe_next_path
 from url_public import browser_url, default_review_url, public_origin
 from urllib.parse import urlencode