灵眸账号登录接入

import express from "express";
import cookieParser from "cookie-parser";
import crypto from "node:crypto";

const BASE = process.env.LMU_SSO_BASE_URL;
const CLIENT_ID = process.env.LMU_SSO_CLIENT_ID;
const CLIENT_SECRET = process.env.LMU_SSO_CLIENT_SECRET;
const REDIRECT_URI = process.env.LMU_SSO_REDIRECT_URI;
const STATE_COOKIE = "lmu_sso_state";

const app = express();
app.use(cookieParser());

// 第一步：生成 state，跳转到灵眸授权页
app.get("/sso/login", (req, res) => {
  const state = crypto.randomBytes(32).toString("base64url");
  res.cookie(STATE_COOKIE, state, {
    httpOnly: true, secure: true, sameSite: "lax", maxAge: 5 * 60 * 1000, path: "/",
  });
  const url = new URL("/sso/authorize", BASE);
  url.searchParams.set("client_id", CLIENT_ID);
  url.searchParams.set("state", state);
  url.searchParams.set("redirect_uri", REDIRECT_URI);
  res.redirect(url.toString());
});

// 第二、三步：校验 state，服务端换取身份
app.get("/sso/callback", async (req, res) => {
  const { ticket, state, error } = req.query;
  const expected = req.cookies[STATE_COOKIE];
  res.clearCookie(STATE_COOKIE, { path: "/" });

  const ok = typeof state === "string" && typeof expected === "string" &&
    Buffer.byteLength(state) === Buffer.byteLength(expected) &&
    crypto.timingSafeEqual(Buffer.from(state), Buffer.from(expected));
  if (!ok) return res.redirect("/login?error=invalid_state");
  if (error === "access_denied") return res.redirect("/login?error=cancelled");
  if (typeof ticket !== "string" || !ticket) return res.redirect("/login?error=missing_ticket");

  let user;
  try {
    const resp = await fetch(new URL("/api/v1/sso/exchange", BASE), {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ client_id: CLIENT_ID, client_secret: CLIENT_SECRET, ticket }),
      signal: AbortSignal.timeout(10_000),
    });
    if (!resp.ok) return res.redirect("/login?error=exchange_failed");
    user = (await resp.json()).data.user; // { id, username, email, avatar }
  } catch {
    return res.redirect("/login?error=exchange_failed");
  }

  // 换成你自己的逻辑：按 user.id 查找或创建本站账号，然后写入本站会话
  // const localUser = await upsertUserByLmuId(user);
  // await createSession(res, localUser);

  res.redirect("/");
});

app.listen(3000);

import hmac
import os
import secrets
from urllib.parse import urlencode

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse

BASE = os.environ["LMU_SSO_BASE_URL"]
CLIENT_ID = os.environ["LMU_SSO_CLIENT_ID"]
CLIENT_SECRET = os.environ["LMU_SSO_CLIENT_SECRET"]
REDIRECT_URI = os.environ["LMU_SSO_REDIRECT_URI"]
STATE_COOKIE = "lmu_sso_state"

app = FastAPI()


def back_to_login(reason: str) -> RedirectResponse:
    resp = RedirectResponse(f"/login?error={reason}", status_code=302)
    resp.delete_cookie(STATE_COOKIE)
    return resp


# 第一步：生成 state，跳转到灵眸授权页
@app.get("/sso/login")
def sso_login():
    state = secrets.token_urlsafe(32)
    query = urlencode({"client_id": CLIENT_ID, "state": state, "redirect_uri": REDIRECT_URI})
    resp = RedirectResponse(f"{BASE}/sso/authorize?{query}", status_code=302)
    resp.set_cookie(STATE_COOKIE, state, max_age=300, httponly=True, secure=True, samesite="lax")
    return resp


# 第二、三步：校验 state，服务端换取身份
@app.get("/sso/callback")
async def sso_callback(request: Request, ticket: str | None = None,
                       state: str | None = None, error: str | None = None):
    expected = request.cookies.get(STATE_COOKIE)
    if not state or not expected or not hmac.compare_digest(state.encode(), expected.encode()):
        return back_to_login("invalid_state")
    if error == "access_denied":
        return back_to_login("cancelled")
    if not ticket:
        return back_to_login("missing_ticket")

    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.post(f"{BASE}/api/v1/sso/exchange", json={
                "client_id": CLIENT_ID,
                "client_secret": CLIENT_SECRET,
                "ticket": ticket,
            })
        resp.raise_for_status()
        user = resp.json()["data"]["user"]  # {"id", "username", "email", "avatar"}
    except (httpx.HTTPError, ValueError, KeyError, TypeError):
        return back_to_login("exchange_failed")

    # 换成你自己的逻辑：按 user["id"] 查找或创建本站账号，然后写入本站会话
    # local_user = upsert_user_by_lmu_id(user)

    done = RedirectResponse("/", status_code=302)
    done.delete_cookie(STATE_COOKIE)
    # create_session(done, local_user)
    return done