Remove login defaults; open review via device token SSO.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-18 08:30:08 +08:00
parent bc50f300ba
commit 8c4afe7781
6 changed files with 171 additions and 18 deletions
+97 -5
View File
@@ -7,7 +7,11 @@ import time
from datetime import datetime, timedelta, timezone
from functools import wraps
from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
BJT = timezone(timedelta(hours=8))
SSO_SALT = "gold-monitor-device-web-sso"
SSO_MAX_AGE = 120 # 秒;设备免密登录票据有效期
from flask import (
Flask,
@@ -70,6 +74,24 @@ def register_routes(app: Flask) -> None:
"is_admin": session.get("role") == "admin",
}
def _sso_serializer() -> URLSafeTimedSerializer:
return URLSafeTimedSerializer(Config.SECRET_KEY, salt=SSO_SALT)
def _establish_session(u: dict, account_id=None) -> None:
session.clear()
session["user_id"] = u["id"]
session["username"] = u["username"]
session["role"] = u["role"]
# 设备免密:会话账户以激活绑定账户为准,便于只看本账户复盘
session["account_id"] = account_id if account_id is not None else u.get("account_id")
session["user"] = u["username"]
def _safe_next(raw: str) -> str:
nxt = (raw or "").strip() or url_for("records_page")
if not nxt.startswith("/") or nxt.startswith("//"):
return url_for("records_page")
return nxt
@app.get("/login")
def login():
if session.get("user_id"):
@@ -84,11 +106,7 @@ def register_routes(app: Flask) -> None:
if not u:
flash("用户名或密码错误", "error")
return render_template("login.html"), 401
session["user_id"] = u["id"]
session["username"] = u["username"]
session["role"] = u["role"]
session["account_id"] = u.get("account_id")
session["user"] = u["username"] # 兼容旧模板
_establish_session(u)
return redirect(request.args.get("next") or url_for("dashboard"))
@app.get("/logout")
@@ -96,6 +114,46 @@ def register_routes(app: Flask) -> None:
session.clear()
return redirect(url_for("login"))
@app.get("/auth/device-login")
def device_login():
"""设备 token / 短期票据 → 网页会话(免密进复盘)。"""
nxt = _safe_next(request.args.get("next") or "")
ticket = (request.args.get("ticket") or "").strip()
token = (request.args.get("token") or "").strip()
u = None
account_id = None
try:
if ticket:
data = _sso_serializer().loads(ticket, max_age=SSO_MAX_AGE)
with database.get_db() as conn:
u = database.get_user(conn, int(data["uid"]))
account_id = data.get("aid")
if u and int(u.get("enabled") or 0) != 1:
u = None
elif token:
with database.get_db() as conn:
act = database.auth_token(conn, token)
if not act:
flash("设备未激活或 token 无效", "error")
return redirect(url_for("login"))
u = database.resolve_web_user_for_activation(conn, act)
account_id = act.get("account_id")
else:
flash("缺少登录凭证", "error")
return redirect(url_for("login"))
except SignatureExpired:
flash("登录链接已过期,请从客户端重新打开复盘", "error")
return redirect(url_for("login"))
except (BadSignature, KeyError, TypeError, ValueError):
flash("登录凭证无效", "error")
return redirect(url_for("login"))
if not u:
flash("该激活账户未绑定网页用户,请管理员在「用户」中创建并绑定账户", "error")
return redirect(url_for("login"))
_establish_session(u, account_id=account_id)
return redirect(nxt)
@app.get("/settings")
@login_required
def settings_page():
@@ -555,6 +613,40 @@ def register_routes(app: Flask) -> None:
)
return jsonify({"ok": True, "server_time": database.now()})
@app.post("/api/v1/web-sso")
def api_web_sso():
"""用设备 token 换短期免密登录链接(打开复盘用)。"""
token = request.headers.get("X-Device-Token") or ""
data = request.get_json(silent=True) or {}
nxt = _safe_next(data.get("next") or "/records")
with database.get_db() as conn:
act = database.auth_token(conn, token)
if not act:
return jsonify({"ok": False, "error": "未激活"}), 401
u = database.resolve_web_user_for_activation(conn, act)
if not u:
return jsonify({
"ok": False,
"error": "该激活账户未绑定网页用户,请管理员在「用户」中创建并绑定账户",
}), 403
if act.get("machine_id"):
conn.execute(
"UPDATE machines SET last_seen_at=? WHERE id=?",
(database.now(), act["machine_id"]),
)
ticket = _sso_serializer().dumps({
"uid": int(u["id"]),
"aid": act.get("account_id"),
"mid": act.get("machine_id"),
})
path = url_for("device_login", ticket=ticket, next=nxt)
return jsonify({
"ok": True,
"ticket": ticket,
"path": path,
"expires_in": SSO_MAX_AGE,
})
@app.post("/api/v1/trades")
def api_trades_upsert():
token = request.headers.get("X-Device-Token") or ""