document-management/tools/mock-sso-proxy/mock_sso_proxy.py

"""本地单点登录代理 mock（需求2 开发联调用，stdlib 零依赖）。

模拟真实单点登录环境的两个 REST 对端：
  代理服务（本脚本 8080）：GET  /middle/token/get.json       → 颁发单点登录 token
  用户服务（本脚本 8800）：POST /auth/user/getUserBasicInfo  → 校验 Bearer token 并返回用户基本信息

响应信封与真实环境一致：{"msg": "成功", "code": 200, "data": {...}}，code=200 为成功。
真实环境中两个服务地址不同（代理 127.0.0.1:8080、用户服务 x.x.x.x:8800），
mock 用同一进程监听两个端口。

用户与登录态来源（优先级从高到低）：
  1. 共享状态文件 ../mock-sso-service/sso-state.json —— 由 mock-sso-service
     管理界面驱动（模拟登录/退出实时生效，本脚本每次请求重读）；
  2. 文件不存在时回落本文件写死的 USERS 实例（单独启动本脚本的旧用法不变）。

用法：python tools/mock-sso-proxy/mock_sso_proxy.py
"""

from __future__ import annotations

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

# 共享状态文件：mock-sso-service 拥有并原子写入，本脚本只读
STATE_FILE = Path(__file__).resolve().parent.parent / "mock-sso-service" / "sso-state.json"

# 回落实例用户：token → 用户基本信息（对端不提供机构字段，机构归属由本系统设密环节用户自选）
USERS = {
    "mock-sso-token-zhangsan": {
        "userId": "9001",
        "username": "zhangsan",
        "realName": "张三",
        "sex": "1",
        "deptId": "100",
        "createTime": "2019-03-20T15:04:10.000+0000",
        "cardId": "",
    },
    "mock-sso-token-lisi": {
        "userId": "9002",
        "username": "lisi",
        "realName": "李四",
        "sex": "0",
        "deptId": "101",
        "createTime": "2020-06-01T09:00:00.000+0000",
        "cardId": "",
    },
}

# 回落模式下代理当前颁发的 token（真实代理由大系统登录态写入；mock 固定取第一个实例用户）
CURRENT_TOKEN = next(iter(USERS))

PROXY_PORT = 8080
USER_SERVICE_PORT = 8800


def _ok(data: dict) -> dict:
    return {"msg": "成功", "code": 200, "data": data}


def _fail(code: int, msg: str) -> dict:
    return {"msg": msg, "code": code, "data": None}


def _load_shared_state():
    """读共享状态文件；返回 (token→用户, 当前颁发token) 或 None（文件缺失/损坏 → 回落内置实例）。"""
    try:
        state = json.loads(STATE_FILE.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    by_token = {}
    for user in state.get("users", []):
        token = user.get("accessToken")
        if token:
            by_token[token] = user
    current = None
    for user in state.get("users", []):
        if user.get("userId") == state.get("currentUserId") and user.get("accessToken"):
            current = user["accessToken"]
            break
    return by_token, current


class Handler(BaseHTTPRequestHandler):
    """两端口共用处理器：按端口与路径分发。"""

    server_version = "MockSsoProxy/0.3"

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

    def _write(self, payload: dict, status: int = 200) -> None:
        body = 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)

    # ----------------------------------------------------------- 代理（8080）

    def do_GET(self) -> None:  # noqa: N802 - stdlib 固定命名
        if self.server.server_port == PROXY_PORT and self.path == "/middle/token/get.json":
            shared = _load_shared_state()
            if shared is not None:
                _, current = shared
                if current is None:
                    # 门户未登录：信封格式不变（DMS 网关按 code!=200 转"未返回有效令牌"）
                    self._write(_fail(401, "门户未登录"))
                    return
                self._write(_ok({
                    "accessToken": current,
                    "tokenType": "Bearer",
                    "refreshToken": shared[0][current].get("refreshToken") or f"{current}-refresh",
                }))
                return
            self._write(_ok({
                "accessToken": CURRENT_TOKEN,
                "tokenType": "Bearer",
                "refreshToken": f"{CURRENT_TOKEN}-refresh",
            }))
            return
        self._write(_fail(404, "not found"), status=404)

    # ------------------------------------------------------- 用户服务（8800）

    def do_POST(self) -> None:  # noqa: N802 - stdlib 固定命名
        if self.server.server_port != USER_SERVICE_PORT or self.path != "/auth/user/getUserBasicInfo":
            self._write(_fail(404, "not found"), status=404)
            return
        token = self.headers.get("Authorization", "").removeprefix("Bearer ").strip()
        shared = _load_shared_state()
        user = shared[0].get(token) if shared else USERS.get(token)
        if user is None:
            self._write(_fail(401, "token非法或已过期"))
            return
        # 状态文件用户行携带 token 管理字段，对外只输出对端契约内的基本信息
        basic = {k: user[k] for k in
                 ("userId", "username", "realName", "sex", "deptId", "createTime", "cardId")}
        self._write(_ok({
            "basicUserInfo": basic,
            "expendUserInfo": {"officePhone": "", "phone": ""},
        }))


def main() -> None:
    proxy = ThreadingHTTPServer(("127.0.0.1", PROXY_PORT), Handler)
    user_service = ThreadingHTTPServer(("127.0.0.1", USER_SERVICE_PORT), Handler)
    import threading

    threading.Thread(target=user_service.serve_forever, daemon=True).start()
    shared = _load_shared_state()
    if shared is not None:
        _, current = shared
        print(f"[mock-sso] 状态来源  共享文件 {STATE_FILE}")
        print(f"[mock-sso] 当前登录态  {'token=' + current if current else '门户未登录（模拟登录请在管理界面操作）'}")
    else:
        print(f"[mock-sso] 状态来源  内置实例（未找到共享文件，可启动 mock-sso-service 接管用户与登录态）")
        print(f"[mock-sso] 当前实例用户：{USERS[CURRENT_TOKEN]['realName']}"
              f"（token={CURRENT_TOKEN}）")
    print(f"[mock-sso] 代理服务  http://127.0.0.1:{PROXY_PORT}/middle/token/get.json")
    print(f"[mock-sso] 用户服务  http://127.0.0.1:{USER_SERVICE_PORT}/auth/user/getUserBasicInfo")
    print("[mock-sso] Ctrl+C 停止")
    try:
        proxy.serve_forever()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()