feat(auth,admin): add LinuxDO OAuth login and admin settings

Implement LinuxDO OAuth authorization-code login flow with state validation and local account binding, and add admin APIs/UI to manage OAuth runtime settings with URL validation and secret masking.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app.py

@@ -2,13 +2,16 @@ from __future__ import annotations
 
 import json
 import re
+import secrets
+from urllib.parse import urlencode, urlparse
 from typing import Any
 
+import requests
 from flask import Flask, jsonify, redirect, render_template, request, session, url_for
 from flask_socketio import SocketIO, emit, join_room as sio_join_room
 from werkzeug.security import check_password_hash, generate_password_hash
 
-from config import Config
+from config import Config, DEFAULT_SETTINGS, SETTINGS_FILE
 from game.action_resolver import ActionResolver
 from game.room_manager import RoomManager
 from game.state_manager import StateManager
@@ -416,6 +419,80 @@ def create_app(config_object: type[Config] = Config) -> tuple[Flask, SocketIO]:
     app.extensions["story_engine"] = story_engine
     app.extensions["room_manager"] = room_manager
 
+    def load_runtime_settings() -> dict[str, Any]:
+        defaults = dict(DEFAULT_SETTINGS)
+        defaults["linuxdo_oauth"] = dict(DEFAULT_SETTINGS.get("linuxdo_oauth", {}))
+        try:
+            payload = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            return defaults
+        if not isinstance(payload, dict):
+            return defaults
+
+        merged = dict(defaults)
+        merged.update(payload)
+        raw_oauth = payload.get("linuxdo_oauth", {})
+        if isinstance(raw_oauth, dict):
+            merged["linuxdo_oauth"] = {**defaults["linuxdo_oauth"], **raw_oauth}
+        else:
+            merged["linuxdo_oauth"] = dict(defaults["linuxdo_oauth"])
+        return merged
+
+    def save_runtime_settings(settings_payload: dict[str, Any]) -> None:
+        SETTINGS_FILE.write_text(
+            json.dumps(settings_payload, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+
+    def _is_valid_http_url(url_text: Any) -> bool:
+        url = str(url_text or "").strip()
+        if not url:
+            return False
+        parsed = urlparse(url)
+        return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
+
+    def get_linuxdo_oauth_config(*, mask_secret: bool = False) -> dict[str, Any]:
+        defaults = DEFAULT_SETTINGS.get("linuxdo_oauth", {})
+        runtime = load_runtime_settings().get("linuxdo_oauth", {})
+        raw = runtime if isinstance(runtime, dict) else {}
+        cfg = {
+            "enabled": bool(raw.get("enabled", defaults.get("enabled", False))),
+            "client_id": str(raw.get("client_id", defaults.get("client_id", ""))).strip(),
+            "client_secret": str(raw.get("client_secret", defaults.get("client_secret", ""))).strip(),
+            "authorize_url": str(raw.get("authorize_url", defaults.get("authorize_url", ""))).strip(),
+            "token_url": str(raw.get("token_url", defaults.get("token_url", ""))).strip(),
+            "userinfo_url": str(raw.get("userinfo_url", defaults.get("userinfo_url", ""))).strip(),
+            "scope": str(raw.get("scope", defaults.get("scope", ""))).strip(),
+        }
+        if mask_secret:
+            secret = cfg["client_secret"]
+            cfg["client_secret"] = "" if not secret else ("*" * 8)
+            cfg["client_secret_configured"] = bool(secret)
+        return cfg
+
+    def oauth_error_redirect(message: str) -> Any:
+        return redirect(url_for("login_page", oauth_error=message))
+
+    def resolve_linuxdo_identity(userinfo: dict[str, Any]) -> tuple[str, str]:
+        raw_id = userinfo.get("id")
+        if raw_id in {None, ""} and isinstance(userinfo.get("user"), dict):
+            raw_id = userinfo["user"].get("id")
+        if raw_id in {None, ""}:
+            raise ValueError("missing_user_id")
+
+        safe_id = re.sub(r"[^a-z0-9_]+", "_", str(raw_id).strip().lower()).strip("_") or "user"
+        username = f"linuxdo_{safe_id}"[:32]
+        if len(username) < 3:
+            username = "linuxdo_user"
+
+        display_name = str(
+            userinfo.get("name")
+            or userinfo.get("username")
+            or (userinfo.get("user", {}) if isinstance(userinfo.get("user"), dict) else {}).get("username")
+            or username
+        ).strip()
+        return username, (display_name[:24] or username)
+
     def current_user() -> dict[str, Any] | None:
         user_id = session.get("user_id")
         if not user_id:
@@ -476,7 +553,8 @@ def create_app(config_object: type[Config] = Config) -> tuple[Flask, SocketIO]:
 
     @app.get("/login")
     def login_page() -> str:
-        return render_template("login.html", user=serialize_user(current_user()))
+        oauth_error = str(request.args.get("oauth_error", "")).strip()[:120]
+        return render_template("login.html", user=serialize_user(current_user()), oauth_error=oauth_error)
 
     @app.get("/register")
     def register_page() -> str:
@@ -531,6 +609,103 @@ def create_app(config_object: type[Config] = Config) -> tuple[Flask, SocketIO]:
         session.pop("user_id", None)
         return jsonify({"ok": True})
 
+    @app.get("/api/auth/oauth/linuxdo/start")
+    def auth_oauth_linuxdo_start_api() -> Any:
+        cfg = get_linuxdo_oauth_config()
+        if not cfg.get("enabled"):
+            return jsonify({"ok": False, "error": "Linux DO OAuth 未启用。"}), 400
+        required_fields = ["client_id", "client_secret", "authorize_url", "token_url", "userinfo_url"]
+        missing = [field for field in required_fields if not str(cfg.get(field, "")).strip()]
+        if missing:
+            return jsonify({"ok": False, "error": f"Linux DO OAuth 配置缺失：{', '.join(missing)}"}), 400
+        for url_field in ("authorize_url", "token_url", "userinfo_url"):
+            if not _is_valid_http_url(cfg.get(url_field, "")):
+                return jsonify({"ok": False, "error": f"Linux DO OAuth 配置无效：{url_field}"}), 400
+
+        state = secrets.token_urlsafe(24)
+        session["linuxdo_oauth_state"] = state
+        callback_url = url_for("auth_oauth_linuxdo_callback_api", _external=True)
+        query = {
+            "client_id": cfg["client_id"],
+            "response_type": "code",
+            "redirect_uri": callback_url,
+            "state": state,
+        }
+        scope = str(cfg.get("scope", "")).strip()
+        if scope:
+            query["scope"] = scope
+        authorize_url = f"{cfg['authorize_url']}?{urlencode(query)}"
+        return redirect(authorize_url)
+
+    @app.get("/api/auth/oauth/linuxdo/callback")
+    def auth_oauth_linuxdo_callback_api() -> Any:
+        cfg = get_linuxdo_oauth_config()
+        if not cfg.get("enabled"):
+            return oauth_error_redirect("Linux DO OAuth 未启用")
+
+        state = str(request.args.get("state", "")).strip()
+        code = str(request.args.get("code", "")).strip()
+        expected_state = str(session.pop("linuxdo_oauth_state", "")).strip()
+        if not code:
+            return oauth_error_redirect("缺少授权 code")
+        if not state or state != expected_state:
+            return oauth_error_redirect("state 校验失败")
+
+        callback_url = url_for("auth_oauth_linuxdo_callback_api", _external=True)
+        try:
+            token_resp = requests.post(
+                cfg["token_url"],
+                data={
+                    "grant_type": "authorization_code",
+                    "code": code,
+                    "client_id": cfg["client_id"],
+                    "client_secret": cfg["client_secret"],
+                    "redirect_uri": callback_url,
+                },
+                timeout=12,
+            )
+            token_resp.raise_for_status()
+            token_payload = token_resp.json()
+        except Exception as exc:
+            return oauth_error_redirect(f"获取 access_token 失败: {exc}")
+
+        access_token = str(token_payload.get("access_token", "")).strip()
+        if not access_token:
+            return oauth_error_redirect("access_token 为空")
+
+        try:
+            userinfo_resp = requests.get(
+                cfg["userinfo_url"],
+                headers={"Authorization": f"Bearer {access_token}"},
+                timeout=12,
+            )
+            userinfo_resp.raise_for_status()
+            userinfo = userinfo_resp.json()
+            if not isinstance(userinfo, dict):
+                return oauth_error_redirect("userinfo 响应格式无效")
+        except Exception as exc:
+            return oauth_error_redirect(f"获取用户信息失败: {exc}")
+
+        try:
+            username, display_name = resolve_linuxdo_identity(userinfo)
+        except ValueError:
+            return oauth_error_redirect("用户信息缺少 id")
+
+        user = persistence.get_user_by_username(username)
+        if user is None:
+            random_password = secrets.token_urlsafe(32)
+            try:
+                user = persistence.create_user(
+                    username=username,
+                    password_hash=generate_password_hash(random_password, method="pbkdf2:sha256"),
+                    display_name=display_name,
+                )
+            except ValueError as exc:
+                return oauth_error_redirect(f"创建本地账号失败: {readable_error(str(exc))}")
+
+        session["user_id"] = user["user_id"]
+        return redirect(url_for("index"))
+
     @app.get("/api/me")
     def me_api() -> Any:
         user = current_user()
@@ -875,6 +1050,73 @@ def create_app(config_object: type[Config] = Config) -> tuple[Flask, SocketIO]:
             card["creator_display_name"] = creator["username"] if creator else ("官方" if card.get("is_official") else "玩家")
         return jsonify({"ok": True, "cards": cards})
 
+    @app.get("/api/admin/settings/oauth/linuxdo")
+    def admin_linuxdo_oauth_get_api() -> Any:
+        _, error_resp = require_admin_json()
+        if error_resp:
+            return error_resp
+        return jsonify({"ok": True, "config": get_linuxdo_oauth_config(mask_secret=True)})
+
+    @app.patch("/api/admin/settings/oauth/linuxdo")
+    def admin_linuxdo_oauth_patch_api() -> Any:
+        _, error_resp = require_admin_json()
+        if error_resp:
+            return error_resp
+
+        payload = request.get_json(silent=True) or {}
+        if not isinstance(payload, dict):
+            payload = {}
+
+        try:
+            raw_settings = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            raw_settings = {}
+        if not isinstance(raw_settings, dict):
+            raw_settings = {}
+
+        defaults = DEFAULT_SETTINGS.get("linuxdo_oauth", {})
+        existing_raw = raw_settings.get("linuxdo_oauth", {})
+        existing_cfg = {**defaults, **(existing_raw if isinstance(existing_raw, dict) else {})}
+
+        enabled = bool(payload.get("enabled", existing_cfg.get("enabled", False)))
+        client_id = str(payload.get("client_id", existing_cfg.get("client_id", ""))).strip()
+        incoming_secret = payload.get("client_secret")
+        if incoming_secret is None:
+            client_secret = str(existing_cfg.get("client_secret", "")).strip()
+        else:
+            normalized_secret = str(incoming_secret).strip()
+            client_secret = normalized_secret or str(existing_cfg.get("client_secret", "")).strip()
+        authorize_url = str(payload.get("authorize_url", existing_cfg.get("authorize_url", ""))).strip()
+        token_url = str(payload.get("token_url", existing_cfg.get("token_url", ""))).strip()
+        userinfo_url = str(payload.get("userinfo_url", existing_cfg.get("userinfo_url", ""))).strip()
+        scope = str(payload.get("scope", existing_cfg.get("scope", ""))).strip()
+
+        for field, url_value in (
+            ("authorize_url", authorize_url),
+            ("token_url", token_url),
+            ("userinfo_url", userinfo_url),
+        ):
+            if not _is_valid_http_url(url_value):
+                return jsonify({"ok": False, "error": f"Linux DO OAuth 配置无效：{field}"}), 400
+
+        if enabled and not client_id:
+            return jsonify({"ok": False, "error": "Linux DO OAuth 配置缺失：client_id"}), 400
+        if enabled and not client_secret:
+            return jsonify({"ok": False, "error": "Linux DO OAuth 配置缺失：client_secret"}), 400
+
+        raw_settings["linuxdo_oauth"] = {
+            "enabled": enabled,
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "authorize_url": authorize_url,
+            "token_url": token_url,
+            "userinfo_url": userinfo_url,
+            "scope": scope,
+        }
+        save_runtime_settings(raw_settings)
+
+        return jsonify({"ok": True, "config": get_linuxdo_oauth_config(mask_secret=True)})
+
     @app.patch("/api/admin/cards/<card_id>")
     def admin_patch_card_api(card_id: str) -> Any:
         _, error_resp = require_admin_json()

config.py

@@ -22,6 +22,15 @@ DEFAULT_SETTINGS: dict[str, Any] = {
     "app_host": "0.0.0.0",
     "app_port": 8095,
     "debug": True,
+    "linuxdo_oauth": {
+        "enabled": False,
+        "client_id": "",
+        "client_secret": "",
+        "authorize_url": "https://connect.linux.do/oauth2/authorize",
+        "token_url": "https://connect.linux.do/oauth2/token",
+        "userinfo_url": "https://connect.linux.do/api/user",
+        "scope": "read",
+    },
 }
 
 
@@ -49,6 +58,12 @@ def _load_settings() -> dict[str, Any]:
 
     merged = dict(DEFAULT_SETTINGS)
     merged.update(payload)
+    linuxdo_defaults = DEFAULT_SETTINGS["linuxdo_oauth"]
+    linuxdo_payload = payload.get("linuxdo_oauth", {})
+    if isinstance(linuxdo_payload, dict):
+        merged["linuxdo_oauth"] = {**linuxdo_defaults, **linuxdo_payload}
+    else:
+        merged["linuxdo_oauth"] = dict(linuxdo_defaults)
     return merged
 
 
@@ -82,6 +97,7 @@ class Config:
     APP_HOST = str(SETTINGS.get("app_host", DEFAULT_SETTINGS["app_host"]))
     APP_PORT = int(SETTINGS.get("app_port", DEFAULT_SETTINGS["app_port"]))
     DEBUG = bool(SETTINGS.get("debug", DEFAULT_SETTINGS["debug"]))
+    LINUXDO_OAUTH = SETTINGS.get("linuxdo_oauth", DEFAULT_SETTINGS["linuxdo_oauth"])
 
 
 class TestConfig(Config):

settings.json

@@ -12,5 +12,14 @@
   "max_action_length": 280,
   "app_host": "0.0.0.0",
   "app_port": 8095,
-  "debug": true
+  "debug": true,
+  "linuxdo_oauth": {
+    "enabled": false,
+    "client_id": "",
+    "client_secret": "",
+    "authorize_url": "https://connect.linux.do/oauth2/authorize",
+    "token_url": "https://connect.linux.do/oauth2/token",
+    "userinfo_url": "https://connect.linux.do/api/user",
+    "scope": "read"
+  }
 }

static/js/admin.js

@@ -12,6 +12,7 @@
   const usersTable = document.getElementById('users-table');
   const roomsTable = document.getElementById('rooms-table');
   const cardsTable = document.getElementById('cards-table');
+  const oauthForm = document.getElementById('linuxdo-oauth-form');
 
   const ROLE_OPTIONS = ['player', 'moderator', 'admin'];
 
@@ -252,6 +253,54 @@
     cardsTable.appendChild(table);
   };
 
+  const loadLinuxdoOauth = async () => {
+    if (!oauthForm) return true;
+    const data = await getJson('/api/admin/settings/oauth/linuxdo');
+    if (!data.ok) {
+      setStatus(data.error || '加载 OAuth 配置失败');
+      return false;
+    }
+    const cfg = data.config || {};
+    document.getElementById('linuxdo-enabled').value = cfg.enabled ? 'true' : 'false';
+    document.getElementById('linuxdo-client-id').value = cfg.client_id || '';
+    document.getElementById('linuxdo-client-secret').value = '';
+    document.getElementById('linuxdo-authorize-url').value = cfg.authorize_url || '';
+    document.getElementById('linuxdo-token-url').value = cfg.token_url || '';
+    document.getElementById('linuxdo-userinfo-url').value = cfg.userinfo_url || '';
+    document.getElementById('linuxdo-scope').value = cfg.scope || '';
+    return true;
+  };
+
+  if (oauthForm) {
+    oauthForm.addEventListener('submit', async (event) => {
+      event.preventDefault();
+      setStatus('正在保存 OAuth 配置...');
+      const payload = {
+        enabled: document.getElementById('linuxdo-enabled').value === 'true',
+        client_id: document.getElementById('linuxdo-client-id').value.trim(),
+        client_secret: document.getElementById('linuxdo-client-secret').value,
+        authorize_url: document.getElementById('linuxdo-authorize-url').value.trim(),
+        token_url: document.getElementById('linuxdo-token-url').value.trim(),
+        userinfo_url: document.getElementById('linuxdo-userinfo-url').value.trim(),
+        scope: document.getElementById('linuxdo-scope').value.trim(),
+      };
+      if (!payload.client_secret) {
+        delete payload.client_secret;
+      }
+      try {
+        const data = await patchJson('/api/admin/settings/oauth/linuxdo', payload);
+        if (!data.ok) {
+          setStatus(data.error || '保存失败');
+          return;
+        }
+        document.getElementById('linuxdo-client-secret').value = '';
+        setStatus('OAuth 配置已保存。');
+      } catch (error) {
+        setStatus(`网络错误: ${error}`);
+      }
+    });
+  }
+
   const loadAll = async () => {
     if (!user || user.role !== 'admin') {
       window.location.href = '/';
@@ -259,12 +308,13 @@
     }
     setStatus('正在加载后台数据...');
     try {
-      const [usersResp, roomsResp, cardsResp] = await Promise.all([
+      const [usersResp, roomsResp, cardsResp, oauthOk] = await Promise.all([
         getJson('/api/admin/users'),
         getJson('/api/admin/rooms'),
         getJson('/api/admin/cards'),
+        loadLinuxdoOauth(),
       ]);
-      if (!usersResp.ok || !roomsResp.ok || !cardsResp.ok) {
+      if (!usersResp.ok || !roomsResp.ok || !cardsResp.ok || !oauthOk) {
         const msg = usersResp.error || roomsResp.error || cardsResp.error || '加载失败';
         setStatus(msg);
         return;

templates/admin.html

@@ -21,6 +21,37 @@
   </header>
 
   <main class="admin-grid">
+    <section class="card admin-panel">
+      <h2>Linux DO OAuth 配置</h2>
+      <form id="linuxdo-oauth-form">
+        <label for="linuxdo-enabled">启用 OAuth</label>
+        <select id="linuxdo-enabled" name="enabled">
+          <option value="false">关闭</option>
+          <option value="true">开启</option>
+        </select>
+
+        <label for="linuxdo-client-id">Client ID</label>
+        <input id="linuxdo-client-id" name="client_id" placeholder="请输入 client_id" />
+
+        <label for="linuxdo-client-secret">Client Secret</label>
+        <input id="linuxdo-client-secret" name="client_secret" type="password" placeholder="留空则保持不变" />
+
+        <label for="linuxdo-authorize-url">Authorize URL</label>
+        <input id="linuxdo-authorize-url" name="authorize_url" placeholder="https://connect.linux.do/oauth2/authorize" />
+
+        <label for="linuxdo-token-url">Token URL</label>
+        <input id="linuxdo-token-url" name="token_url" placeholder="https://connect.linux.do/oauth2/token" />
+
+        <label for="linuxdo-userinfo-url">Userinfo URL</label>
+        <input id="linuxdo-userinfo-url" name="userinfo_url" placeholder="https://connect.linux.do/api/user" />
+
+        <label for="linuxdo-scope">Scope</label>
+        <input id="linuxdo-scope" name="scope" placeholder="read" />
+
+        <button id="linuxdo-save" type="submit" class="secondary-btn">保存 OAuth 配置</button>
+      </form>
+    </section>
+
     <section class="card admin-panel">
       <h2>用户权限管理</h2>
       <div id="users-table" class="table-wrap"></div>

templates/login.html

@@ -16,9 +16,10 @@
       <input id="password" name="password" type="password" required placeholder="至少 6 位" />
       <button type="submit">登录</button>
     </form>
+    <a class="link-btn" href="{{ url_for('auth_oauth_linuxdo_start_api') }}" style="margin-top: 0.75rem;">使用 Linux DO 登录</a>
     <p class="auth-links">没有账号？<a href="{{ url_for('register_page') }}">去注册</a></p>
     <p class="auth-links"><a href="{{ url_for('index') }}">返回大厅</a></p>
-    <section id="status" class="status-banner"></section>
+    <section id="status" class="status-banner">{{ oauth_error or '' }}</section>
   </main>
   <script src="{{ url_for('static', filename='js/theme.js') }}"></script>
   <script src="{{ url_for('static', filename='js/auth.js') }}"></script>