增加密码

manual_trading_hub/hub_web_auth.py

@@ -0,0 +1,78 @@
+"""中控 Web 登录：HUB_PASSWORD 非空时启用会话 Cookie。"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import time
+from secrets import compare_digest
+
+SESSION_COOKIE = "hub_sess"
+SESSION_MAX_AGE_SEC = max(3600, int(os.getenv("HUB_SESSION_DAYS", "7")) * 86400)
+
+
+def password_required() -> bool:
+    return bool((os.getenv("HUB_PASSWORD") or "").strip())
+
+
+def verify_password(password: str) -> bool:
+    expected = (os.getenv("HUB_PASSWORD") or "").strip()
+    if not expected:
+        return True
+    return compare_digest(expected, (password or "").strip())
+
+
+def _secret() -> bytes:
+    raw = (os.getenv("HUB_SESSION_SECRET") or os.getenv("HUB_PASSWORD") or "").strip()
+    if not raw:
+        return b"hub-dev-insecure"
+    return raw.encode("utf-8")
+
+
+def _b64url_encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
+
+
+def _b64url_decode(text: str) -> bytes:
+    pad = "=" * (-len(text) % 4)
+    return base64.urlsafe_b64decode(text + pad)
+
+
+def create_session_token() -> str:
+    payload = {"exp": int(time.time()) + SESSION_MAX_AGE_SEC, "v": 1}
+    body = _b64url_encode(json.dumps(payload, separators=(",", ":")).encode("utf-8"))
+    sig = hmac.new(_secret(), body.encode("ascii"), hashlib.sha256).hexdigest()
+    return f"{body}.{sig}"
+
+
+def validate_session_token(token: str | None) -> bool:
+    if not token or "." not in token:
+        return False
+    body, sig = token.rsplit(".", 1)
+    expected = hmac.new(_secret(), body.encode("ascii"), hashlib.sha256).hexdigest()
+    if not compare_digest(expected, sig):
+        return False
+    try:
+        payload = json.loads(_b64url_decode(body))
+    except Exception:
+        return False
+    exp = int(payload.get("exp") or 0)
+    return exp > int(time.time())
+
+
+def cookie_secure() -> bool:
+    return (os.getenv("HUB_COOKIE_SECURE") or "").strip().lower() in ("1", "true", "yes", "on")
+
+
+def is_public_path(path: str, method: str) -> bool:
+    p = (path or "").split("?")[0].rstrip("/") or "/"
+    if p.startswith("/assets"):
+        return True
+    if p in ("/login", "/api/auth/login", "/api/auth/status"):
+        return True
+    if p == "/api/auth/logout" and method.upper() == "POST":
+        return True
+    return False