摇月 Swaymoon 摇月 文档

Python 最小可运行示例 开发人员

上次更新：2026年8月12日

Python 最小可运行示例

本页提供两份 Authorization Code + PKCE 最小可运行示例：

  • 非机密客户端（public）：Python 只提供本地页面，浏览器完成发现、换票和 UserInfo 请求，不使用 client_secret。
  • 机密客户端（confidential）：Python 服务端保管 client_secret，完成发现、换票、JWKS、ID Token 校验和 UserInfo 请求；并演示接收「撤销并请求删除数据」的签名回调。

两份 demo 默认申请全部资料权限，并显式发送 prompt=consent（演示用；生产环境即使省略该参数，通行账户也会每次展示授权确认页）。用户仍可在同意页拒绝可选权限；程序会展示令牌最终获准的 scope 和 UserInfo，未获授权的字段仍是默认占位值。

数据删除回调依赖 HMAC 密钥：机密客户端使用创建时下发的 client_secret 明文（不是库内 bcrypt 哈希）。非机密客户端无 client_secret，若登记删除回调则会另发 webhookSecret；本页不在 public demo 中演示该回调，请用机密客户端示例联调。

非机密客户端（public）

这是一份非机密客户端的 Authorization Code + PKCE 最小示例。Python 只负责提供本地页面与回调地址；OIDC 发现、浏览器授权、换取令牌和 UserInfo 请求全部由浏览器完成。

因此，运行脚本的 CMD、PowerShell 或 python.exe 不需要访问公网。只要浏览器能访问摇月通行账户，示例就能完成登录。服务端已对 /.well-known/**、/oauth2/token 和 /userinfo 开放非凭据 CORS，并通过 PKCE、精确匹配的 Redirect URI 和访问令牌保证安全。

机密客户端不能使用这种模式：client_secret 绝不能进入浏览器。机密客户端请阅读《使用生成式 AI 接入机密客户端》，由真正可访问公网的后端完成换票。

此 demo 会显式发送 prompt=consent，用于演示授权确认页。生产环境即使不发送该参数，服务端也会要求用户每次在授权页确认登录；权限有增减时还会展示变更对照。已上线的旧客户端无需因此改代码；新客户端请以最新《权限与同意》为准。

1. 注册客户端

在开发者门户创建非机密客户端，并登记：

http://127.0.0.1:8765/callback

在门户保留必选的 openid，并将 email、name、nickname、picture、biography、gender、birthdate、region、preferred_username 全部登记为可选权限。demo 会请求这些权限并展示最终获准的 scope 和 UserInfo；用户仍可在同意页逐项拒绝。redirect_uri 必须逐字符一致；127.0.0.1、localhost 和 [::1] 是不同的登记值。

2. 运行

需要 Python 3.10 或更高版本，不安装任何第三方包。把下一节代码保存为 swaymoon_oidc_mvp.py。

Windows PowerShell

$env:SWAYMOON_CLIENT_ID = "swm_你的客户端ID"
py .\swaymoon_oidc_mvp.py

Windows 命令提示符（CMD）

set "SWAYMOON_CLIENT_ID=swm_你的客户端ID"
py swaymoon_oidc_mvp.py

macOS / Linux

export SWAYMOON_CLIENT_ID='swm_你的客户端ID'
python3 swaymoon_oidc_mvp.py

脚本会打印并尝试打开 http://127.0.0.1:8765/。如果没有自动打开，用浏览器手动访问该地址。

3. 完整代码

#!/usr/bin/env python3
"""摇月通行账户：浏览器完成公网请求的非机密客户端 PKCE MVP。"""

from __future__ import annotations

import http.server
import html
import ipaddress
import json
import os
import socket
import traceback
import urllib.parse
import webbrowser


ISSUER = os.environ.get(
    "SWAYMOON_ISSUER", "https://api-passport.swaymoon.com"
).strip().rstrip("/")
CLIENT_ID = os.environ.get("SWAYMOON_CLIENT_ID", "").strip()
REDIRECT_URI = os.environ.get(
    "SWAYMOON_REDIRECT_URI", "http://127.0.0.1:8765/callback"
).strip()
SCOPE = os.environ.get(
    "SWAYMOON_SCOPE",
    "openid email name nickname picture biography gender birthdate region preferred_username",
).strip()


def parse_callback() -> tuple[str, int, str, str]:
    parsed = urllib.parse.urlsplit(REDIRECT_URI)
    if parsed.scheme != "http" or not parsed.hostname or parsed.port is None:
        raise SystemExit(
            "SWAYMOON_REDIRECT_URI 必须是带端口的本地 HTTP 地址，"
            "例如 http://127.0.0.1:8765/callback"
        )
    if parsed.query or parsed.fragment:
        raise SystemExit("此示例的 SWAYMOON_REDIRECT_URI 不能包含 query 或 fragment")
    try:
        loopback = ipaddress.ip_address(parsed.hostname).is_loopback
    except ValueError:
        loopback = parsed.hostname == "localhost"
    if not loopback:
        raise SystemExit("此示例只监听 127.0.0.0/8、localhost 或 ::1")
    path = parsed.path or "/callback"
    origin = urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, "", "", ""))
    return parsed.hostname, parsed.port, path, origin


CALLBACK_HOST, CALLBACK_PORT, CALLBACK_PATH, LOCAL_ORIGIN = parse_callback()


HTML = r"""<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>摇月通行账户 · PKCE MVP</title>
  <style>
    body { font: 16px/1.55 system-ui, sans-serif; max-width: 760px; margin: 48px auto; padding: 0 20px; }
    button { font: inherit; padding: 10px 18px; cursor: pointer; }
    pre { padding: 16px; overflow: auto; background: #f5f5f7; border-radius: 10px; white-space: pre-wrap; }
    code { word-break: break-word; }
    .muted { color: #666; }
  </style>
</head>
<body>
  <h1>摇月通行账户 · PKCE MVP</h1>
  <p class="muted">公网 OIDC 请求由浏览器发出，Python 只提供这个本地页面。</p>
  <div id="home" hidden>
    <p>客户端：<code id="client"></code></p>
    <p>权限：<code id="scope"></code></p>
    <button id="login" type="button">使用摇月通行账户登录</button>
  </div>
  <div id="callback" hidden>
    <p id="status">正在校验回调并换取令牌……</p>
  </div>
  <pre id="output" hidden></pre>

  <script>
  "use strict";
  const CONFIG = __CONFIG__;
  const PAGE = __PAGE__;
  const STORAGE_KEY = "swaymoon-pkce-mvp";
  const output = document.getElementById("output");

  function showError(error) {
    output.hidden = false;
    output.textContent = error instanceof Error ? `${error.name}: ${error.message}` : String(error);
    const status = document.getElementById("status");
    if (status) status.textContent = "登录未完成";
    console.error(error);
  }

  function base64url(bytes) {
    let binary = "";
    for (const byte of bytes) binary += String.fromCharCode(byte);
    return btoa(binary).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/, "");
  }

  function randomValue(length = 32) {
    return base64url(crypto.getRandomValues(new Uint8Array(length)));
  }

  async function fetchJson(url, options = {}) {
    const response = await fetch(url, {
      credentials: "omit",
      cache: "no-store",
      ...options,
    });
    const text = await response.text();
    let value;
    try {
      value = JSON.parse(text);
    } catch {
      throw new Error(`${url} 未返回 JSON（HTTP ${response.status}）：${text.slice(0, 300)}`);
    }
    if (!response.ok) {
      throw new Error(`${url} 返回 HTTP ${response.status}：${JSON.stringify(value)}`);
    }
    return value;
  }

  async function discover() {
    const metadata = await fetchJson(`${CONFIG.issuer}/.well-known/openid-configuration`);
    if (String(metadata.issuer || "").replace(/\/$/, "") !== CONFIG.issuer) {
      throw new Error(`发现文档 issuer 不匹配：${metadata.issuer}`);
    }
    for (const name of ["authorization_endpoint", "token_endpoint", "userinfo_endpoint"]) {
      if (typeof metadata[name] !== "string") throw new Error(`发现文档缺少 ${name}`);
    }
    return metadata;
  }

  async function startLogin() {
    const button = document.getElementById("login");
    button.disabled = true;
    button.textContent = "正在准备登录……";
    try {
      const metadata = await discover();
      const verifier = randomValue(64);
      const digest = await crypto.subtle.digest(
        "SHA-256", new TextEncoder().encode(verifier)
      );
      const state = randomValue(32);
      sessionStorage.setItem(STORAGE_KEY, JSON.stringify({
        state,
        verifier,
        createdAt: Date.now(),
      }));

      const authorize = new URL(metadata.authorization_endpoint);
      authorize.search = new URLSearchParams({
        response_type: "code",
        client_id: CONFIG.clientId,
        redirect_uri: CONFIG.redirectUri,
        scope: CONFIG.scope,
        prompt: "consent",
        state,
        code_challenge: base64url(new Uint8Array(digest)),
        code_challenge_method: "S256",
      });
      location.assign(authorize);
    } catch (error) {
      button.disabled = false;
      button.textContent = "使用摇月通行账户登录";
      showError(error);
    }
  }

  async function finishLogin() {
    try {
      const query = new URLSearchParams(location.search);
      const savedText = sessionStorage.getItem(STORAGE_KEY);
      if (!savedText) throw new Error("缺少本地 PKCE 状态，请从首页重新开始");
      const saved = JSON.parse(savedText);
      if (Date.now() - Number(saved.createdAt) > 10 * 60 * 1000) {
        throw new Error("PKCE 状态已过期，请从首页重新开始");
      }
      if (!query.get("state") || query.get("state") !== saved.state) {
        throw new Error("state 校验失败，请从首页重新开始");
      }
      if (query.get("error")) {
        throw new Error(`${query.get("error")}: ${query.get("error_description") || "授权未完成"}`);
      }
      const code = query.get("code");
      if (!code) throw new Error("回调缺少 code");

      const metadata = await discover();
      const tokens = await fetchJson(metadata.token_endpoint, {
        method: "POST",
        headers: { "Content-Type": "application/x-www-form-urlencoded" },
        body: new URLSearchParams({
          grant_type: "authorization_code",
          code,
          redirect_uri: CONFIG.redirectUri,
          client_id: CONFIG.clientId,
          code_verifier: saved.verifier,
        }),
      });
      sessionStorage.removeItem(STORAGE_KEY);
      if (typeof tokens.access_token !== "string" || !tokens.access_token) {
        throw new Error("令牌响应缺少 access_token");
      }
      const userinfo = await fetchJson(metadata.userinfo_endpoint, {
        headers: { Authorization: `Bearer ${tokens.access_token}` },
      });
      document.getElementById("status").textContent = "登录成功";
      output.hidden = false;
      output.textContent = JSON.stringify({
        token: {
          token_type: tokens.token_type,
          expires_in: tokens.expires_in,
          scope: tokens.scope,
          has_id_token: typeof tokens.id_token === "string",
          has_refresh_token: typeof tokens.refresh_token === "string",
        },
        userinfo,
      }, null, 2);
    } catch (error) {
      showError(error);
    }
  }

  if (PAGE === "callback") {
    document.getElementById("callback").hidden = false;
    finishLogin();
  } else {
    document.getElementById("home").hidden = false;
    document.getElementById("client").textContent = CONFIG.clientId;
    document.getElementById("scope").textContent = CONFIG.scope;
    document.getElementById("login").addEventListener("click", startLogin);
  }
  </script>
</body>
</html>
"""


def javascript_json(value: object) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":")).replace("<", "\\u003c")


def render(page: str) -> str:
    config = {
        "issuer": ISSUER,
        "clientId": CLIENT_ID,
        "redirectUri": REDIRECT_URI,
        "scope": SCOPE,
    }
    return HTML.replace("__CONFIG__", javascript_json(config)).replace(
        "__PAGE__", javascript_json(page)
    )


class Handler(http.server.BaseHTTPRequestHandler):
    def log_message(self, fmt: str, *args: object) -> None:
        print("[http]", fmt % args)

    def send_html(self, status: int, body: str) -> None:
        data = body.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("Referrer-Policy", "no-referrer")
        self.end_headers()
        if data:
            self.wfile.write(data)

    def do_GET(self) -> None:
        try:
            path = urllib.parse.urlsplit(self.path).path
            if path == "/favicon.ico":
                self.send_html(204, "")
            elif path == "/":
                self.send_html(200, render("home"))
            elif path == CALLBACK_PATH:
                self.send_html(200, render("callback"))
            else:
                self.send_html(404, "<h1>404</h1><p><a href='/'>返回首页</a></p>")
        except BrokenPipeError:
            pass
        except Exception:
            detail = traceback.format_exc()
            print(detail)
            try:
                self.send_html(500, f"<h1>本地服务器错误</h1><pre>{html.escape(detail)}</pre>")
            except Exception:
                pass


class ThreadingServer(http.server.ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True


class IPv6ThreadingServer(ThreadingServer):
    address_family = socket.AF_INET6


def main() -> None:
    if not CLIENT_ID:
        raise SystemExit("请先设置 SWAYMOON_CLIENT_ID")
    if os.environ.get("SWAYMOON_CLIENT_SECRET", "").strip():
        raise SystemExit("此页面仅支持非机密客户端；不要把 client_secret 交给浏览器")
    if "openid" not in SCOPE.split():
        raise SystemExit("SWAYMOON_SCOPE 必须包含 openid")
    issuer = urllib.parse.urlsplit(ISSUER)
    if issuer.scheme not in ("http", "https") or not issuer.hostname:
        raise SystemExit("SWAYMOON_ISSUER 不是有效的 HTTP(S) URL")

    server_class = IPv6ThreadingServer if ":" in CALLBACK_HOST else ThreadingServer
    try:
        server = server_class((CALLBACK_HOST, CALLBACK_PORT), Handler)
    except OSError as error:
        raise SystemExit(f"无法监听 {LOCAL_ORIGIN}：{error}") from error

    print("客户端类型：public")
    print("Python 公网访问：不需要；OIDC 请求由浏览器完成")
    print("Issuer：", ISSUER)
    print("Scope：", SCOPE)
    print("浏览器入口：", f"{LOCAL_ORIGIN}/")
    print("重定向 URI：", REDIRECT_URI)
    try:
        webbrowser.open(f"{LOCAL_ORIGIN}/")
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n已停止")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()

4. 可选配置

环境变量默认值用途
SWAYMOON_ISSUERhttps://api-passport.swaymoon.com浏览器读取其 OIDC 发现文档
SWAYMOON_REDIRECT_URIhttp://127.0.0.1:8765/callback本地回调；必须与门户登记值完全一致
SWAYMOON_SCOPE全部可用资料权限空格分隔的权限；须先在门户登记

Windows PowerShell 只测试 openid email 的例子：

$env:SWAYMOON_SCOPE = "openid email"
py .\swaymoon_oidc_mvp.py

如果希望回调本身走 IPv6，先在门户登记 http://[::1]:8765/callback，再设置：

$env:SWAYMOON_REDIRECT_URI = "http://[::1]:8765/callback"
py .\swaymoon_oidc_mvp.py

这只改变本地回调监听地址。公网请求始终由浏览器按其正常的 IPv4/IPv6、代理和安全策略完成。

5. 常见错误

现象处理
页面提示发现文档 Failed to fetch在同一浏览器直接打开发现文档；检查浏览器扩展、CORS 报错和开发者工具 Network 面板
invalid_client必须使用门户创建的非机密客户端；本页不接受 client_secret
invalid_grant从本地首页重新登录；授权码只能使用一次，且 Redirect URI 必须完全一致
invalid_scope将 SWAYMOON_SCOPE 改回 openid，或先在门户启用相应功能
state 校验失败不要刷新或复用旧回调页；从本地首页重新开始
无法监听端口结束占用 8765 的旧进程，或登记并使用另一个端口

这份代码用于协议联调。生产 SPA 应使用成熟的 OAuth / OIDC 库，并建立可靠的会话、错误处理和令牌生命周期管理；不要把 Access Token、ID Token 或任何密钥写入日志或持久化存储。

机密客户端（confidential）

这是一份可直接运行的 Authorization Code + PKCE 机密客户端示例。浏览器只负责跳转；OIDC 发现、授权码换票、JWKS、ID Token 校验和 UserInfo 请求全部在 Python 服务端完成，client_secret 和令牌不会进入浏览器。

机密客户端所在的 Python 进程必须能够访问 Issuer。若当前网络会重置非浏览器 TLS 连接，请在正常网络、服务器或容器中运行；不要把换票操作改到浏览器来规避网络问题。

1. 注册客户端

在开发者门户创建机密客户端，并登记：

字段值说明
重定向 URIhttp://127.0.0.1:8766/callback浏览器跳转；本地 loopback 可用
数据删除回调地址https://<你的隧道主机>/privacy/data-deletion仅 HTTPS；禁止 http://、127.0.0.1、localhost。用 ngrok / Cloudflare Tunnel 等把本机 8766 暴露为 HTTPS 后登记

立即安全保存创建时仅展示一次的 client_secret（校验删除回调签名时也要用它）。在门户保留必选的 openid，并将 email、name、nickname、picture、biography、gender、birthdate、region、preferred_username 全部登记为可选权限。demo 会请求全部权限，但用户仍可在同意页逐项拒绝。

登录 Redirect 与删除回调不是一回事。
Redirect 由浏览器访问，可用 http://127.0.0.1。删除回调由 Passport 服务端出站 POST，门户不允许登记本地地址。本地联调：先启 demo，再用隧道得到 https://….ngrok-free.app，把 https://….ngrok-free.app/privacy/data-deletion 填进门户；点「撤销并请求删除数据」时隧道与 demo 须在跑。也可用下文 curl 在本机自检验签（不经过 Passport）。邮件是否送达与回调是否成功无关。详见同目录《注册与配置》。

2. 安装依赖

代码本身使用 Python 标准库处理 HTTP，只额外使用 cryptography 校验 ES256 签名。

Windows PowerShell / CMD

py -m pip install cryptography

macOS / Linux

python3 -m pip install cryptography

3. 配置并运行

把下一节代码保存为 swaymoon_confidential_mvp.py。

Windows PowerShell

$env:SWAYMOON_CLIENT_ID = "swm_你的客户端ID"
$env:SWAYMOON_CLIENT_SECRET = "你的客户端密钥"
py .\swaymoon_confidential_mvp.py

Windows 命令提示符（CMD）

set "SWAYMOON_CLIENT_ID=swm_你的客户端ID"
set "SWAYMOON_CLIENT_SECRET=你的客户端密钥"
py swaymoon_confidential_mvp.py

macOS / Linux

export SWAYMOON_CLIENT_ID='swm_你的客户端ID'
export SWAYMOON_CLIENT_SECRET='你的客户端密钥'
python3 swaymoon_confidential_mvp.py

打开 http://127.0.0.1:8766/，选择登录按钮。停止服务后清除当前终端中的密钥环境变量；不要把真实值写入代码、文档或 Git。

4. 完整代码

#!/usr/bin/env python3
"""摇月通行账户：服务端换票并校验 ES256 ID Token 的机密客户端 PKCE MVP。"""

from __future__ import annotations

import base64
import hashlib
import hmac
import html
import http.cookies
import http.server
import ipaddress
import json
import os
import secrets
import socket
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
from typing import Any

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature


ISSUER = os.environ.get(
    "SWAYMOON_ISSUER", "https://api-passport.swaymoon.com"
).strip().rstrip("/")
CLIENT_ID = os.environ.get("SWAYMOON_CLIENT_ID", "").strip()
CLIENT_SECRET = os.environ.get("SWAYMOON_CLIENT_SECRET", "").strip()
REDIRECT_URI = os.environ.get(
    "SWAYMOON_REDIRECT_URI", "http://127.0.0.1:8766/callback"
).strip()
SCOPE = os.environ.get(
    "SWAYMOON_SCOPE",
    "openid email name nickname picture biography gender birthdate region preferred_username",
).strip()
DATA_DELETION_PATH = os.environ.get(
    "SWAYMOON_DATA_DELETION_PATH", "/privacy/data-deletion"
).strip() or "/privacy/data-deletion"
HTTP_TIMEOUT = 15
SESSION_TTL = 10 * 60

SESSIONS: dict[str, dict[str, Any]] = {}
SESSION_LOCK = threading.Lock()
# 本地用户档案：以 UserInfo / ID Token 的 sub 为键（演示删除回调）
LOCAL_USERS: dict[str, dict[str, Any]] = {}
LOCAL_USERS_LOCK = threading.Lock()
DELETION_LOG: list[dict[str, Any]] = []
METADATA: dict[str, Any] | None = None
JWKS: dict[str, Any] | None = None


def b64url_encode(value: bytes) -> str:
    return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii")


def b64url_decode(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def parse_callback() -> tuple[str, int, str, str]:
    parsed = urllib.parse.urlsplit(REDIRECT_URI)
    if parsed.scheme != "http" or not parsed.hostname or parsed.port is None:
        raise SystemExit(
            "SWAYMOON_REDIRECT_URI 必须是带端口的本地 HTTP 地址，"
            "例如 http://127.0.0.1:8766/callback"
        )
    if parsed.query or parsed.fragment:
        raise SystemExit("此示例的 SWAYMOON_REDIRECT_URI 不能包含 query 或 fragment")
    try:
        loopback = ipaddress.ip_address(parsed.hostname).is_loopback
    except ValueError:
        loopback = parsed.hostname == "localhost"
    if not loopback:
        raise SystemExit("此示例只监听 127.0.0.0/8、localhost 或 ::1")
    path = parsed.path or "/callback"
    origin = urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, "", "", ""))
    return parsed.hostname, parsed.port, path, origin


CALLBACK_HOST, CALLBACK_PORT, CALLBACK_PATH, LOCAL_ORIGIN = parse_callback()


def request_json(
    url: str,
    *,
    data: dict[str, str] | None = None,
    headers: dict[str, str] | None = None,
) -> dict[str, Any]:
    body = urllib.parse.urlencode(data).encode("utf-8") if data is not None else None
    request = urllib.request.Request(url, data=body, headers=headers or {})
    try:
        with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT) as response:
            raw = response.read()
    except urllib.error.HTTPError as error:
        detail = error.read(500).decode("utf-8", "replace")
        raise RuntimeError(f"{url} 返回 HTTP {error.code}：{detail}") from error
    except OSError as error:
        raise RuntimeError(f"无法连接 {url}：{type(error).__name__}: {error}") from error
    try:
        value = json.loads(raw)
    except json.JSONDecodeError as error:
        raise RuntimeError(f"{url} 未返回 JSON") from error
    if not isinstance(value, dict):
        raise RuntimeError(f"{url} 的 JSON 顶层不是对象")
    return value


def discover() -> dict[str, Any]:
    global METADATA
    if METADATA is None:
        value = request_json(f"{ISSUER}/.well-known/openid-configuration")
        if value.get("issuer") != ISSUER:
            raise RuntimeError(f"发现文档 issuer 不匹配：{value.get('issuer')}")
        for name in (
            "authorization_endpoint",
            "token_endpoint",
            "jwks_uri",
            "userinfo_endpoint",
        ):
            if not isinstance(value.get(name), str):
                raise RuntimeError(f"发现文档缺少 {name}")
        METADATA = value
    return METADATA


def get_jwks() -> dict[str, Any]:
    global JWKS
    if JWKS is None:
        JWKS = request_json(discover()["jwks_uri"])
    return JWKS


def verify_id_token(token: str, expected_nonce: str) -> dict[str, Any]:
    try:
        encoded_header, encoded_payload, encoded_signature = token.split(".")
        header = json.loads(b64url_decode(encoded_header))
        claims = json.loads(b64url_decode(encoded_payload))
        signature = b64url_decode(encoded_signature)
    except (ValueError, UnicodeError, json.JSONDecodeError) as error:
        raise RuntimeError("ID Token 格式无效") from error
    if not isinstance(header, dict) or not isinstance(claims, dict):
        raise RuntimeError("ID Token header/payload 格式无效")
    if header.get("alg") != "ES256" or not isinstance(header.get("kid"), str):
        raise RuntimeError("ID Token 必须使用带 kid 的 ES256")

    keys = get_jwks().get("keys")
    if not isinstance(keys, list):
        raise RuntimeError("JWKS 缺少 keys")
    jwk = next(
        (
            key
            for key in keys
            if isinstance(key, dict)
            and key.get("kid") == header["kid"]
            and key.get("kty") == "EC"
            and key.get("crv") == "P-256"
        ),
        None,
    )
    if jwk is None:
        raise RuntimeError("JWKS 中找不到 ID Token 使用的 P-256 公钥")
    try:
        public_key = ec.EllipticCurvePublicNumbers(
            int.from_bytes(b64url_decode(jwk["x"]), "big"),
            int.from_bytes(b64url_decode(jwk["y"]), "big"),
            ec.SECP256R1(),
        ).public_key()
        if len(signature) != 64:
            raise ValueError("ES256 签名长度错误")
        der_signature = encode_dss_signature(
            int.from_bytes(signature[:32], "big"),
            int.from_bytes(signature[32:], "big"),
        )
        public_key.verify(
            der_signature,
            f"{encoded_header}.{encoded_payload}".encode("ascii"),
            ec.ECDSA(hashes.SHA256()),
        )
    except (InvalidSignature, KeyError, TypeError, ValueError) as error:
        raise RuntimeError("ID Token ES256 签名校验失败") from error

    now = time.time()
    if claims.get("iss") != ISSUER:
        raise RuntimeError("ID Token iss 不匹配")
    audience = claims.get("aud")
    audiences = [audience] if isinstance(audience, str) else audience
    if not isinstance(audiences, list) or CLIENT_ID not in audiences:
        raise RuntimeError("ID Token aud 不包含当前 client_id")
    if len(audiences) > 1 and claims.get("azp") != CLIENT_ID:
        raise RuntimeError("ID Token azp 不匹配")
    if not isinstance(claims.get("exp"), (int, float)) or claims["exp"] < now - 60:
        raise RuntimeError("ID Token 已过期或缺少 exp")
    if not isinstance(claims.get("iat"), (int, float)) or claims["iat"] > now + 60:
        raise RuntimeError("ID Token iat 无效")
    if "nbf" in claims and (
        not isinstance(claims["nbf"], (int, float)) or claims["nbf"] > now + 60
    ):
        raise RuntimeError("ID Token 尚未生效")
    if not secrets.compare_digest(str(claims.get("nonce", "")), expected_nonce):
        raise RuntimeError("ID Token nonce 校验失败")
    if not isinstance(claims.get("sub"), str) or not claims["sub"]:
        raise RuntimeError("ID Token 缺少 sub")
    return claims


def basic_authorization() -> str:
    client = urllib.parse.quote_plus(CLIENT_ID, safe="")
    secret = urllib.parse.quote_plus(CLIENT_SECRET, safe="")
    value = base64.b64encode(f"{client}:{secret}".encode("utf-8")).decode("ascii")
    return f"Basic {value}"


def verify_data_deletion_signature(raw_body: bytes, signature_header: str | None) -> bool:
    """校验 X-Swaymoon-Signature: sha256=<hex>，密钥为创建时 client_secret 明文。"""
    if not signature_header or not CLIENT_SECRET:
        return False
    provided = signature_header.strip()
    if provided.lower().startswith("sha256="):
        provided = provided[7:].strip()
    digest = hmac.new(
        CLIENT_SECRET.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()
    return secrets.compare_digest(digest.lower(), provided.lower())


def cookie_session_id(cookie_header: str | None) -> str | None:
    if not cookie_header:
        return None
    cookie = http.cookies.SimpleCookie()
    try:
        cookie.load(cookie_header)
    except http.cookies.CookieError:
        return None
    morsel = cookie.get("swaymoon_mvp_session")
    return morsel.value if morsel else None


def page(title: str, content: str) -> str:
    return f"""<!doctype html>
<html lang="zh-CN"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{html.escape(title)}</title><style>
body{{font:16px/1.55 system-ui,sans-serif;max-width:760px;margin:48px auto;padding:0 20px}}
a.button{{display:inline-block;padding:10px 18px;background:#111;color:#fff;text-decoration:none;border-radius:8px}}
pre{{padding:16px;background:#f5f5f7;border-radius:10px;white-space:pre-wrap;overflow:auto}}
.hint{{color:#6e6e73;font-size:14px}}
</style></head><body><h1>{html.escape(title)}</h1>{content}</body></html>"""


class Handler(http.server.BaseHTTPRequestHandler):
    def log_message(self, fmt: str, *args: object) -> None:
        print("[http]", fmt % args)

    def send_body(
        self,
        status: int,
        body: str,
        *,
        content_type: str = "text/html; charset=utf-8",
        location: str | None = None,
        cookie: str | None = None,
    ) -> None:
        data = body.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("Referrer-Policy", "no-referrer")
        if content_type.startswith("text/html"):
            self.send_header(
                "Content-Security-Policy",
                "default-src 'none'; style-src 'unsafe-inline'",
            )
        if location:
            self.send_header("Location", location)
        if cookie:
            self.send_header("Set-Cookie", cookie)
        self.end_headers()
        self.wfile.write(data)

    def redirect(self, location: str, cookie: str | None = None) -> None:
        self.send_body(302, "", location=location, cookie=cookie)

    def show_error(self, error: Exception) -> None:
        print(f"[error] {type(error).__name__}: {error}")
        body = (
            f"<p>{html.escape(type(error).__name__ + ': ' + str(error))}</p>"
            "<p><a href='/'>返回首页</a></p>"
        )
        self.send_body(400, page("登录未完成", body))

    def do_GET(self) -> None:
        try:
            parsed = urllib.parse.urlsplit(self.path)
            if parsed.path == "/favicon.ico":
                self.send_body(204, "")
            elif parsed.path == "/":
                with LOCAL_USERS_LOCK:
                    user_count = len(LOCAL_USERS)
                    log_count = len(DELETION_LOG)
                self.send_body(
                    200,
                    page(
                        "摇月通行账户 · 机密客户端 PKCE MVP",
                        f"<p>客户端：<code>{html.escape(CLIENT_ID)}</code></p>"
                        f"<p>权限：<code>{html.escape(SCOPE)}</code></p>"
                        f"<p>本地用户：{user_count} · 已处理删除请求：{log_count}</p>"
                        "<p><a class='button' href='/login'>使用摇月通行账户登录</a></p>"
                        "<p><a href='/users'>查看本地用户</a> · "
                        "<a href='/deletion-log'>删除请求日志</a></p>"
                        f"<p class='hint'>数据删除回调：<code>POST {html.escape(DATA_DELETION_PATH)}</code></p>",
                    ),
                )
            elif parsed.path == "/login":
                self.start_login()
            elif parsed.path == CALLBACK_PATH:
                self.finish_login(parsed.query)
            elif parsed.path == "/users":
                self.show_local_users()
            elif parsed.path == "/deletion-log":
                self.show_deletion_log()
            else:
                self.send_body(404, page("404", "<p><a href='/'>返回首页</a></p>"))
        except BrokenPipeError:
            pass
        except Exception as error:
            self.show_error(error)

    def do_POST(self) -> None:
        try:
            parsed = urllib.parse.urlsplit(self.path)
            if parsed.path == DATA_DELETION_PATH:
                self.handle_data_deletion()
            else:
                self.send_body(
                    404,
                    json.dumps({"status": "not-found"}, ensure_ascii=False),
                    content_type="application/json; charset=utf-8",
                )
        except BrokenPipeError:
            pass
        except Exception as error:
            print(f"[deletion-error] {type(error).__name__}: {error}")
            self.send_body(
                500,
                json.dumps({"status": "error"}, ensure_ascii=False),
                content_type="application/json; charset=utf-8",
            )

    def show_local_users(self) -> None:
        with LOCAL_USERS_LOCK:
            snapshot = json.dumps(LOCAL_USERS, ensure_ascii=False, indent=2)
        self.send_body(
            200,
            page(
                "本地用户",
                "<p>以下为登录成功后保存在进程内存中的档案；收到删除回调后会移除对应 <code>sub</code>。</p>"
                f"<pre>{html.escape(snapshot)}</pre><p><a href='/'>返回首页</a></p>",
            ),
        )

    def show_deletion_log(self) -> None:
        with LOCAL_USERS_LOCK:
            snapshot = json.dumps(DELETION_LOG, ensure_ascii=False, indent=2)
        self.send_body(
            200,
            page(
                "删除请求日志",
                "<p>通行账户 <code>POST</code> 数据删除回调后写入。</p>"
                f"<pre>{html.escape(snapshot)}</pre><p><a href='/'>返回首页</a></p>",
            ),
        )

    def handle_data_deletion(self) -> None:
        length = int(self.headers.get("Content-Length", "0") or "0")
        raw = self.rfile.read(length) if length > 0 else b""
        signature = self.headers.get("X-Swaymoon-Signature")
        if not verify_data_deletion_signature(raw, signature):
            self.send_body(
                401,
                json.dumps({"status": "invalid-signature"}, ensure_ascii=False),
                content_type="application/json; charset=utf-8",
            )
            return
        try:
            payload = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as error:
            raise RuntimeError("请求体不是合法 JSON") from error
        if payload.get("type") != "swaymoon.passport.data_deletion_request":
            self.send_body(
                400,
                json.dumps({"status": "unsupported-type"}, ensure_ascii=False),
                content_type="application/json; charset=utf-8",
            )
            return
        subject = payload.get("subject")
        removed = False
        if isinstance(subject, str) and subject:
            with LOCAL_USERS_LOCK:
                removed = LOCAL_USERS.pop(subject, None) is not None
                DELETION_LOG.append(
                    {
                        "request_id": payload.get("request_id"),
                        "client_id": payload.get("client_id"),
                        "subject": subject,
                        "relay_email": payload.get("relay_email"),
                        "removed_local_user": removed,
                        "received_at": int(time.time()),
                    }
                )
        print(
            f"[deletion] request_id={payload.get('request_id')} "
            f"subject={subject} removed={removed}"
        )
        self.send_body(
            200,
            json.dumps(
                {"status": "ok", "removed": removed},
                ensure_ascii=False,
            ),
            content_type="application/json; charset=utf-8",
        )

    def start_login(self) -> None:
        metadata = discover()
        session_id = secrets.token_urlsafe(32)
        state = secrets.token_urlsafe(32)
        nonce = secrets.token_urlsafe(32)
        verifier = secrets.token_urlsafe(64)
        challenge = b64url_encode(hashlib.sha256(verifier.encode("ascii")).digest())
        with SESSION_LOCK:
            now = time.time()
            for key, value in list(SESSIONS.items()):
                if now - float(value.get("created_at", 0)) > SESSION_TTL:
                    SESSIONS.pop(key, None)
            SESSIONS[session_id] = {
                "state": state,
                "nonce": nonce,
                "verifier": verifier,
                "created_at": now,
            }
        authorize = metadata["authorization_endpoint"] + "?" + urllib.parse.urlencode(
            {
                "response_type": "code",
                "client_id": CLIENT_ID,
                "redirect_uri": REDIRECT_URI,
                "scope": SCOPE,
                "prompt": "consent",
                "state": state,
                "nonce": nonce,
                "code_challenge": challenge,
                "code_challenge_method": "S256",
            }
        )
        cookie = (
            f"swaymoon_mvp_session={session_id}; Path=/; HttpOnly; SameSite=Lax; "
            f"Max-Age={SESSION_TTL}"
        )
        self.redirect(authorize, cookie)

    def finish_login(self, raw_query: str) -> None:
        query = urllib.parse.parse_qs(raw_query)
        session_id = cookie_session_id(self.headers.get("Cookie"))
        with SESSION_LOCK:
            saved = SESSIONS.pop(session_id, None) if session_id else None
        if not saved or time.time() - saved["created_at"] > SESSION_TTL:
            raise RuntimeError("登录状态缺失或已过期，请从首页重新开始")
        received_state = query.get("state", [""])[0]
        if not received_state or not secrets.compare_digest(received_state, saved["state"]):
            raise RuntimeError("state 校验失败")
        if query.get("error"):
            description = query.get("error_description", ["授权未完成"])[0]
            raise RuntimeError(f"{query['error'][0]}: {description}")
        code = query.get("code", [""])[0]
        if not code:
            raise RuntimeError("回调缺少 code")

        metadata = discover()
        tokens = request_json(
            metadata["token_endpoint"],
            data={
                "grant_type": "authorization_code",
                "code": code,
                "redirect_uri": REDIRECT_URI,
                "code_verifier": saved["verifier"],
            },
            headers={
                "Authorization": basic_authorization(),
                "Content-Type": "application/x-www-form-urlencoded",
            },
        )
        id_token = tokens.get("id_token")
        access_token = tokens.get("access_token")
        if not isinstance(id_token, str) or not isinstance(access_token, str):
            raise RuntimeError("令牌响应缺少 id_token 或 access_token")
        claims = verify_id_token(id_token, saved["nonce"])
        userinfo = request_json(
            metadata["userinfo_endpoint"],
            headers={"Authorization": f"Bearer {access_token}"},
        )
        if userinfo.get("sub") != claims["sub"]:
            raise RuntimeError("UserInfo sub 与 ID Token sub 不一致")

        subject = str(claims["sub"])
        with LOCAL_USERS_LOCK:
            LOCAL_USERS[subject] = {
                "sub": subject,
                "email": userinfo.get("email"),
                "name": userinfo.get("name"),
                "preferred_username": userinfo.get("preferred_username"),
                "saved_at": int(time.time()),
            }

        summary = {
            "token": {
                "token_type": tokens.get("token_type"),
                "expires_in": tokens.get("expires_in"),
                "scope": tokens.get("scope"),
                "id_token_verified": True,
                "has_refresh_token": isinstance(tokens.get("refresh_token"), str),
            },
            "id_token": {
                "iss": claims.get("iss"),
                "aud": claims.get("aud"),
                "sub": claims.get("sub"),
                "exp": claims.get("exp"),
            },
            "userinfo": userinfo,
            "local_store": "已按 sub 写入进程内存，可供删除回调演示",
        }
        body = (
            "<p>登录成功；令牌仅在服务端内存中处理。</p>"
            "<p>可前往通行账户「隐私 → 应用授权 → 撤销并请求删除数据」，"
            "观察本机 <code>/deletion-log</code> 与本地用户是否被清除。</p>"
            "<pre>"
            + html.escape(json.dumps(summary, ensure_ascii=False, indent=2))
            + "</pre><p><a href='/'>返回首页</a></p>"
        )
        self.send_body(
            200,
            page("登录成功", body),
            cookie="swaymoon_mvp_session=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0",
        )


class ThreadingServer(http.server.ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True


class IPv6ThreadingServer(ThreadingServer):
    address_family = socket.AF_INET6


def main() -> None:
    if not CLIENT_ID or not CLIENT_SECRET:
        raise SystemExit("请先设置 SWAYMOON_CLIENT_ID 和 SWAYMOON_CLIENT_SECRET")
    if "openid" not in SCOPE.split():
        raise SystemExit("SWAYMOON_SCOPE 必须包含 openid")
    issuer = urllib.parse.urlsplit(ISSUER)
    if issuer.scheme != "https" or not issuer.hostname:
        raise SystemExit("SWAYMOON_ISSUER 必须是有效的 HTTPS URL")
    server_class = IPv6ThreadingServer if ":" in CALLBACK_HOST else ThreadingServer
    try:
        server = server_class((CALLBACK_HOST, CALLBACK_PORT), Handler)
    except OSError as error:
        raise SystemExit(f"无法监听 {LOCAL_ORIGIN}：{error}") from error
    print("客户端类型：confidential（client_secret_basic + PKCE S256）")
    print("浏览器入口：", f"{LOCAL_ORIGIN}/")
    print("重定向 URI：", REDIRECT_URI)
    print("本地删除接收：", f"{LOCAL_ORIGIN}{DATA_DELETION_PATH}", "（门户须登记隧道 HTTPS，勿填 127.0.0.1）")
    print("注意：Python 服务端必须能够访问", ISSUER)
    try:
        webbrowser.open(f"{LOCAL_ORIGIN}/")
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n已停止")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()

5. 演示数据删除回调

门户不允许把数据删除回调登记为 http://127.0.0.1。端到端联调步骤：

  1. 启动本 demo（监听 8766）。
  2. 用 HTTPS 隧道把本机 8766 暴露出去（示例：cloudflared tunnel --url http://127.0.0.1:8766 或 ngrok），得到形如 https://xxxx.example 的公网地址。
  3. 在开发者门户将该应用的「数据删除回调地址」设为 https://xxxx.example/privacy/data-deletion（path 与 SWAYMOON_DATA_DELETION_PATH 一致）。
  4. 完成登录，打开 http://127.0.0.1:8766/users，确认本地已有该用户的 sub。
  5. 登录 passport.swaymoon.com，打开 隐私 → 应用授权，进入本 demo 应用详情，选择 撤销并请求删除数据（需 step-up）。
  6. 通行账户会撤销授权、发邮件，并向隧道地址 POST 签名 JSON；demo 的 /deletion-log 应有记录，/users 中对应 sub 消失。

不经过 Passport、仅自检验签时，仍可对 http://127.0.0.1:8766 使用下方 curl。

回调请求约定（与《权限与同意》一致）：

项值
方法 / 路径POST /privacy/data-deletion（可用环境变量改路径）
请求头Content-Type: application/json；X-Swaymoon-Signature: sha256=<hmac_hex>；X-Swaymoon-Request-Id
签名对原始请求体字节做 HMAC-SHA256，密钥为创建时保存的 client_secret 明文（勿用 bcrypt / {bcrypt} 存储串）
正文字段type（固定 swaymoon.passport.data_deletion_request）、issued_at、client_id、subject、relay_email、request_id

可用本地脚本自检验签（不必走 Passport）：

# macOS / Linux 示例：SECRET 为创建时下发的 client_secret 明文
BODY='{"type":"swaymoon.passport.data_deletion_request","issued_at":"2026-01-01T00:00:00Z","client_id":"swm_demo","subject":"替换为真实sub","relay_email":null,"request_id":"req1"}'
SIG=$(printf '%s' "$BODY" | openssl dgst -sha256 -hmac "$SWAYMOON_CLIENT_SECRET" | awk '{print $2}')
curl -sS -X POST "http://127.0.0.1:8766/privacy/data-deletion" \
  -H "Content-Type: application/json" \
  -H "X-Swaymoon-Signature: sha256=$SIG" \
  -H "X-Swaymoon-Request-Id: req1" \
  --data "$BODY"

6. 可选配置

环境变量默认值用途
SWAYMOON_ISSUERhttps://api-passport.swaymoon.comPython 服务端读取发现文档、换票和请求 UserInfo
SWAYMOON_REDIRECT_URIhttp://127.0.0.1:8766/callback本地回调；必须与门户登记值完全一致
SWAYMOON_SCOPE全部可用资料权限空格分隔的权限；必须包含 openid，且须先在门户登记
SWAYMOON_DATA_DELETION_PATH/privacy/data-deletion本地接收删除回调的路径；须与门户「数据删除回调地址」的 path 一致

此 demo 会显式发送 prompt=consent。生产应用可省略该参数：通行账户仍会每次展示授权确认页；权限变更时展示增减对照，用户撤销授权后须重新授权。

7. 安全边界

  • 浏览器只收到应用自己的页面，不会收到 client_secret、Access Token、ID Token 或 Refresh Token。
  • 示例校验 ES256 签名、iss、aud、azp、exp、iat、nbf、nonce，并核对 UserInfo sub。
  • 数据删除回调校验 X-Swaymoon-Signature（密钥为创建时 client_secret 明文，对 raw body 验签）；生产环境还应校验 client_id、限制来源 IP / mTLS，并做幂等处理。
  • state、nonce、PKCE verifier 和令牌只保存在进程内存；重启后登录流程失效。
  • 本地 HTTP Cookie 未设置 Secure，仅适用于 loopback 联调。生产环境必须使用 HTTPS、Secure Cookie、持久会话存储和成熟 OIDC 客户端库。
  • 不要记录完整回调 URL、授权码、密钥或令牌；本示例的成功页只显示经过筛选的声明和状态。

8. 常见错误

现象处理
无法连接 ... / TLS reset机密客户端服务端必须能直连 Issuer；更换正常网络或部署到服务器，不要把密钥交给浏览器
invalid_client检查是否创建了机密客户端，以及 client_id / client_secret 是否正确；修改后重启
invalid_grant从首页重新登录；授权码只能使用一次，Redirect URI 和 PKCE verifier 必须匹配
invalid_scope改回 openid，或先在门户登记需要的权限
ID Token 校验失败不要跳过校验；检查系统时间、Issuer、客户端 ID，并重新登录
没有授权确认页已存在覆盖相同 scope 的同意记录；撤销应用授权或改用新客户端后重试
删除回调 invalid-signature使用创建客户端时保存的同一 client_secret；签名必须对原始 body 计算
点了「请求删除数据」有邮件但 demo 无日志须在门户登记隧道 HTTPS 回调（禁止 127.0.0.1）；点删除时隧道与 demo 须在跑。可先用下文 curl 自检验签