"""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 = """ Mock 单点登录门户模拟

Mock 单点登录门户模拟

模拟门户系统：管理用户与登录态，「模拟登录/退出」直接写入本地 sso-proxy 的 token 存储（sso-state.json）

用户ID	用户名	姓名	性别	部门ID	accessToken（点击复制）	refreshToken（点击复制）	操作

""" # ------------------------------------------------------------ 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//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()