document-management/tools/mock-sso-service/mock_sso_service.py

"""Mock 单点登录门户模拟（开发联调用，stdlib 零依赖）。

模拟真实环境的"门户系统"（大系统）：维护门户用户库与登录态，
界面上的"模拟登录/模拟退出"直接驱动本地 sso-proxy 颁发的 token：

  模拟登录 → 随机生成 accessToken/refreshToken 挂到该用户，
             并把该用户设为 proxy 当前颁发的登录态（currentUserId）
  模拟退出 → 清除该用户的 token；若正是当前登录态则一并清空

与 tools/mock-sso-proxy 的联动走共享状态文件 sso-state.json（本服务
拥有并原子写入，proxy 每次请求重读；文件不存在时 proxy 回落内置实例）。

用法：python tools/mock-sso-service/mock_sso_service.py
界面：http://127.0.0.1:8890/
"""

from __future__ import annotations

import json
import os
import secrets
import threading
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

SERVICE_PORT = 8890

STATE_FILE = Path(__file__).resolve().parent / "sso-state.json"

# 首次启动写入的预置用户（对齐 proxy 原内置实例，无 token 即未登录）
_INITIAL_STATE = {
    "users": [
        {
            "userId": "9001",
            "username": "zhangsan",
            "realName": "张三",
            "sex": "1",
            "deptId": "100",
            "createTime": "2019-03-20T15:04:10.000+0000",
            "cardId": "",
            "accessToken": None,
            "refreshToken": None,
            "tokenIssuedAt": None,
        },
        {
            "userId": "9002",
            "username": "lisi",
            "realName": "李四",
            "sex": "0",
            "deptId": "101",
            "createTime": "2020-06-01T09:00:00.000+0000",
            "cardId": "",
            "accessToken": None,
            "refreshToken": None,
            "tokenIssuedAt": None,
        },
    ],
    # proxy 当前颁发的登录态用户；null = 门户未登录（proxy 颁发 401 信封）
    "currentUserId": None,
}

# 读改写串行化（ThreadingHTTPServer 多线程请求共享状态）
_lock = threading.Lock()


# ------------------------------------------------------------ 状态存取

def _read_state() -> dict:
    """读状态（不加锁）：首次访问落预置文件；已持锁的调用方统一走本函数避免重入死锁。"""
    if not STATE_FILE.exists():
        _write_state_unlocked(_INITIAL_STATE)
        return json.loads(json.dumps(_INITIAL_STATE))
    return json.loads(STATE_FILE.read_text(encoding="utf-8"))


def _load_state() -> dict:
    with _lock:
        return _read_state()


def _write_state_unlocked(state: dict) -> None:
    # 原子写：proxy 侧多线程重读，半截文件会解析失败
    temp = STATE_FILE.with_suffix(".json.tmp")
    temp.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
    os.replace(temp, STATE_FILE)


def _find_user(state: dict, user_id: str) -> dict | None:
    for user in state["users"]:
        if user["userId"] == user_id:
            return user
    return None


def _issue_tokens(user: dict) -> None:
    user["accessToken"] = secrets.token_hex(16)
    user["refreshToken"] = secrets.token_hex(16)
    user["tokenIssuedAt"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "+0000"


def _clear_tokens(user: dict) -> None:
    user["accessToken"] = None
    user["refreshToken"] = None
    user["tokenIssuedAt"] = None


# ------------------------------------------------------------ 内嵌前端

_PAGE = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>Mock 单点登录门户模拟</title>
<style>
  body { font-family: "Microsoft YaHei", sans-serif; font-size: 16px; margin: 0;
         background: #eef3f8; color: #1e3a5f; }
  .wrap { max-width: 1180px; margin: 0 auto; padding: 20px; }
  h1 { font-size: 22px; margin: 0 0 6px; }
  .sub { color: #5b7695; font-size: 14px; margin-bottom: 16px; }
  .banner { background: #fff; border: 1px solid #d6e0ec; border-radius: 8px;
            padding: 12px 16px; margin-bottom: 16px; display: flex;
            align-items: center; gap: 12px; flex-wrap: wrap; }
  .badge { padding: 3px 12px; border-radius: 999px; font-weight: bold; font-size: 14px; }
  .badge.on { background: #e3f2e9; color: #1b7a3d; }
  .badge.off { background: #f3e6e6; color: #a33; }
  form, table { background: #fff; border: 1px solid #d6e0ec; border-radius: 8px; }
  form { display: flex; gap: 10px; padding: 12px 16px; margin-bottom: 16px; flex-wrap: wrap; align-items: center; }
  form label { font-size: 14px; color: #5b7695; }
  input, select { font-size: 16px; padding: 5px 8px; border: 1px solid #c4d2e3;
                  border-radius: 4px; }
  table { width: 100%; border-collapse: collapse; }
  th { background: #1e3a5f; color: #fff; font-size: 15px; padding: 9px 10px; text-align: left; }
  td { padding: 9px 10px; border-bottom: 1px solid #e4ebf3; font-size: 15px; }
  tr.current td { background: #f2f8ff; }
  .token { font-family: Consolas, monospace; font-size: 13px; color: #47678c;
           cursor: copy; max-width: 150px; overflow: hidden; text-overflow: ellipsis;
           white-space: nowrap; display: inline-block; vertical-align: middle; }
  button { font-size: 14px; padding: 5px 12px; border: 1px solid #c4d2e3;
           border-radius: 4px; background: #fff; cursor: pointer; }
  button:hover { background: #f0f5fb; }
  button.primary { background: #2563eb; border-color: #2563eb; color: #fff; }
  button.primary:hover { background: #1d4fd7; }
  button.danger { color: #b33; border-color: #ddb3b3; }
  .muted { color: #93a7bd; }
  #toast { position: fixed; bottom: 24px; left: 50%; transform: translateX(-50%);
           background: #1e3a5f; color: #fff; padding: 8px 20px; border-radius: 6px;
           font-size: 14px; display: none; }
</style>
</head>
<body>
<div class="wrap">
  <h1>Mock 单点登录门户模拟</h1>
  <div class="sub">模拟门户系统：管理用户与登录态，「模拟登录/退出」直接写入本地 sso-proxy 的 token 存储（sso-state.json）</div>
  <div class="banner">
    <span>当前登录态：</span><span id="current" class="badge off">未登录</span>
    <span class="muted">proxy 颁发地址 http://127.0.0.1:8080/middle/token/get.json</span>
  </div>
  <form id="add">
    <label>新增用户</label>
    <input name="username" placeholder="用户名（必填）" required>
    <input name="realName" placeholder="姓名（必填）" required>
    <select name="sex"><option value="1">男</option><option value="0">女</option></select>
    <input name="deptId" placeholder="部门ID" style="width:90px">
    <button class="primary" type="submit">添加</button>
  </form>
  <table>
    <thead><tr>
      <th>用户ID</th><th>用户名</th><th>姓名</th><th>性别</th><th>部门ID</th>
      <th>accessToken（点击复制）</th><th>refreshToken（点击复制）</th><th>操作</th>
    </tr></thead>
    <tbody id="rows"></tbody>
  </table>
</div>
<div id="toast"></div>
<script>
const $ = (id) => document.getElementById(id);

function toast(text) {
  const el = $("toast");
  el.textContent = text;
  el.style.display = "block";
  setTimeout(() => (el.style.display = "none"), 1600);
}

async function call(method, path, body) {
  const res = await fetch(path, {
    method,
    headers: body ? { "Content-Type": "application/json" } : undefined,
    body: body ? JSON.stringify(body) : undefined,
  });
  const data = await res.json().catch(() => ({}));
  if (!res.ok) throw new Error(data.error || res.statusText);
  await render();
  return data;
}

function short(token) {
  if (!token) return "—";
  return token.length <= 16 ? token : token.slice(0, 10) + "…" + token.slice(-4);
}

function tokenCell(token) {
  if (!token) return '<span class="muted">—</span>';
  return `<span class="token" title="${token}（点击复制）">${short(token)}</span>`;
}

async function render() {
  const state = await (await fetch("/api/state")).json();
  $("current").textContent = state.currentUserId
    ? "已登录：" + state.users.find((u) => u.userId === state.currentUserId)?.realName
    : "未登录";
  $("current").className = "badge " + (state.currentUserId ? "on" : "off");
  $("rows").innerHTML = state.users.map((u) => `
    <tr class="${u.userId === state.currentUserId ? "current" : ""}">
      <td>${u.userId}</td>
      <td>${u.username}</td>
      <td>${u.realName}</td>
      <td>${u.sex === "1" ? "男" : "女"}</td>
      <td>${u.deptId || "—"}</td>
      <td>${tokenCell(u.accessToken)}</td>
      <td>${tokenCell(u.refreshToken)}</td>
      <td style="white-space:nowrap">
        <button class="primary" onclick="login('${u.userId}')">模拟登录</button>
        <button onclick="logout('${u.userId}')" ${u.accessToken ? "" : "disabled"}>模拟退出</button>
        <button class="danger" onclick="remove('${u.userId}')">删除</button>
      </td>
    </tr>`).join("");
}

async function login(id) {
  await call("POST", `/api/users/${id}/login`);
  toast("已模拟登录并设为 proxy 当前登录态");
}
async function logout(id) {
  await call("POST", `/api/users/${id}/logout`);
  toast("已模拟退出并清除 token");
}
async function remove(id) {
  if (!confirm("确认删除该用户？")) return;
  await call("DELETE", `/api/users/${id}`);
  toast("已删除");
}

document.addEventListener("click", (e) => {
  if (e.target.classList?.contains("token")) {
    navigator.clipboard?.writeText(e.target.title.split("（")[0]);
    toast("token 已复制");
  }
});

$("add").addEventListener("submit", async (e) => {
  e.preventDefault();
  const form = new FormData(e.target);
  try {
    await call("POST", "/api/users", {
      username: form.get("username").trim(),
      realName: form.get("realName").trim(),
      sex: form.get("sex"),
      deptId: form.get("deptId").trim(),
    });
    e.target.reset();
    toast("已添加用户");
  } catch (err) {
    toast("添加失败：" + err.message);
  }
});

render();
</script>
</body>
</html>
"""


# ------------------------------------------------------------ HTTP 服务

class Handler(BaseHTTPRequestHandler):
    """管理界面与状态 API。"""

    server_version = "MockSsoService/0.1"

    def log_message(self, fmt, *args):  # noqa: A003 - 保留 stdlib 签名
        print(f"[mock-sso-service] {self.address_string()} {fmt % args}")

    def _write(self, payload, status: int = 200) -> None:
        body = payload if isinstance(payload, bytes) else json.dumps(
            payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _error(self, status: int, message: str) -> None:
        self._write({"error": message}, status)

    def _read_body(self) -> dict:
        length = int(self.headers.get("Content-Length") or 0)
        if length <= 0:
            return {}
        return json.loads(self.rfile.read(length).decode("utf-8"))

    # ------------------------------------------------------------------ 路由

    def do_GET(self) -> None:  # noqa: N802 - stdlib 固定命名
        if self.path == "/" or self.path.startswith("/index"):
            body = _PAGE.encode("utf-8")
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if self.path == "/api/state":
            self._write(_load_state())
            return
        self._error(404, "not found")

    def do_POST(self) -> None:  # noqa: N802
        if self.path == "/api/users":
            body = self._read_body()
            username = str(body.get("username") or "").strip()
            real_name = str(body.get("realName") or "").strip()
            if not username or not real_name:
                self._error(400, "username与realName必填")
                return
            with _lock:
                state = _read_state()
                if any(u["username"] == username for u in state["users"]):
                    self._error(409, f"用户名已存在：{username}")
                    return
                state["users"].append({
                    "userId": str(max(int(u["userId"]) for u in state["users"]) + 1),
                    "username": username,
                    "realName": real_name,
                    "sex": "1" if str(body.get("sex") or "1") == "1" else "0",
                    "deptId": str(body.get("deptId") or "").strip(),
                    "createTime": datetime.now(timezone.utc).strftime(
                        "%Y-%m-%dT%H:%M:%S.000+0000"),
                    "cardId": "",
                    "accessToken": None,
                    "refreshToken": None,
                    "tokenIssuedAt": None,
                })
                _write_state_unlocked(state)
            self._write({"ok": True})
            return

        # /api/users/<userId>/login | /logout
        parts = self.path.strip("/").split("/")
        if len(parts) == 4 and parts[0] == "api" and parts[1] == "users" and parts[3] in ("login", "logout"):
            action = parts[3]
            with _lock:
                state = _read_state()
                user = _find_user(state, parts[2])
                if user is None:
                    self._error(404, "用户不存在")
                    return
                if action == "login":
                    _issue_tokens(user)
                    state["currentUserId"] = user["userId"]
                else:
                    _clear_tokens(user)
                    if state.get("currentUserId") == user["userId"]:
                        state["currentUserId"] = None
                _write_state_unlocked(state)
            self._write({"ok": True})
            return
        self._error(404, "not found")

    def do_DELETE(self) -> None:  # noqa: N802
        parts = self.path.strip("/").split("/")
        if len(parts) == 3 and parts[0] == "api" and parts[1] == "users":
            with _lock:
                state = _read_state()
                user = _find_user(state, parts[2])
                if user is None:
                    self._error(404, "用户不存在")
                    return
                state["users"].remove(user)
                if state.get("currentUserId") == user["userId"]:
                    state["currentUserId"] = None
                _write_state_unlocked(state)
            self._write({"ok": True})
            return
        self._error(404, "not found")


def main() -> None:
    server = ThreadingHTTPServer(("127.0.0.1", SERVICE_PORT), Handler)
    _load_state()  # 首次启动即落预置用户文件，proxy 可立即联动
    print(f"[mock-sso-service] 管理界面  http://127.0.0.1:{SERVICE_PORT}/")
    print(f"[mock-sso-service] 状态文件  {STATE_FILE}")
    print("[mock-sso-service] 模拟登录/退出将写入该文件，sso-proxy 每次请求实时重读，Ctrl+C 停止")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()