搖月 Swaymoon 搖月 文件

Python 最小可執行範例 開發人員

上次更新：2026年8月15日

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-TW">
<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-TW"><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 自檢驗簽