Minimal runnable Python example

This page provides two minimal runnable Authorization Code + PKCE examples:

• Public client: Python only serves a local page. The browser performs discovery, token exchange, and the UserInfo request. No client_secret is used.
• Confidential client: The Python server holds client_secret and performs discovery, token exchange, JWKS, ID Token verification, and the UserInfo request. It also demonstrates receiving the signed “Revoke and request data deletion” callback.

Both demos request all profile scopes by default and explicitly send prompt=consent (for demonstration; in production, Swaymoon Account still shows the authorization confirmation page on every sign-in even if you omit that parameter). Users can still decline optional scopes on the consent page. The program shows the scope actually granted on the token and UserInfo; fields that were not authorized remain default placeholders.

The data-deletion callback depends on an HMAC key: confidential clients use the plaintext client_secret issued at creation (not the bcrypt hash stored in the library). Public clients have no client_secret; if you register a deletion callback, a separate webhookSecret is issued. This page does not demonstrate that callback in the public demo—use the confidential-client example to integrate it.

Public client

This is a minimal public-client Authorization Code + PKCE example. Python only serves the local page and callback URL. OIDC discovery, browser authorization, token exchange, and the UserInfo request are all done by the browser.

Therefore the CMD, PowerShell, or python.exe process that runs the script does not need public-internet access. As long as the browser can reach Swaymoon Account, the example can complete sign-in. The server already enables CORS without credentials for /.well-known/**, /oauth2/token, and /userinfo, and security is provided by PKCE, an exactly matching Redirect URI, and the access token.

Confidential clients must not use this pattern: client_secret must never enter the browser. For confidential clients, read Connecting a confidential client with generative AI and exchange the code on a backend that can actually reach the public internet.

This demo explicitly sends prompt=consent so you can see the authorization confirmation page. In production, even without that parameter, the server still requires the user to confirm sign-in on the authorization page every time; when scopes are added or removed, a change comparison is also shown. Existing live clients do not need a code change for this. New clients should follow the latest Scopes and consent.

1. Register a client

In the Developer Portal, create a public client and register:

http://127.0.0.1:8765/callback

Keep required openid, and register email, name, nickname, picture, biography, gender, birthdate, region, and preferred_username all as optional scopes. The demo requests these scopes and shows the finally granted scope and UserInfo; the user can still decline items one by one on the consent page. redirect_uri must match character for character; 127.0.0.1, localhost, and [::1] are distinct registered values.

2. Run

You need Python 3.10 or later. No third-party packages. Save the next section’s code as swaymoon_oidc_mvp.py.

Windows PowerShell

$env:SWAYMOON_CLIENT_ID = "swm_YOUR_CLIENT_ID"
py .\swaymoon_oidc_mvp.py

Windows Command Prompt (CMD)

set "SWAYMOON_CLIENT_ID=swm_YOUR_CLIENT_ID"
py swaymoon_oidc_mvp.py

macOS / Linux

export SWAYMOON_CLIENT_ID='swm_YOUR_CLIENT_ID'
python3 swaymoon_oidc_mvp.py

The script prints and tries to open http://127.0.0.1:8765/. If it does not open automatically, visit that address in a browser.

3. Full code

#!/usr/bin/env python3
"""Swaymoon Account: public-client PKCE MVP where the browser makes the public-network requests."""

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 must be a local HTTP URL with a port, "
            "for example http://127.0.0.1:8765/callback"
        )
    if parsed.query or parsed.fragment:
        raise SystemExit("In this example, SWAYMOON_REDIRECT_URI must not contain a query or fragment")
    try:
        loopback = ipaddress.ip_address(parsed.hostname).is_loopback
    except ValueError:
        loopback = parsed.hostname == "localhost"
    if not loopback:
        raise SystemExit("This example only listens on 127.0.0.0/8, localhost, or ::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="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Swaymoon Account · 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>Swaymoon Account · PKCE MVP</h1>
  <p class="muted">Public OIDC requests are made by the browser. Python only serves this local page.</p>
  <div id="home" hidden>
    <p>Client: <code id="client"></code></p>
    <p>Scopes: <code id="scope"></code></p>
    <button id="login" type="button">Sign in with Swaymoon Account</button>
  </div>
  <div id="callback" hidden>
    <p id="status">Verifying the callback and exchanging tokens…</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 = "Sign-in did not complete";
    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} did not return JSON (HTTP ${response.status}): ${text.slice(0, 300)}`);
    }
    if (!response.ok) {
      throw new Error(`${url} returned 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(`Discovery document issuer mismatch: ${metadata.issuer}`);
    }
    for (const name of ["authorization_endpoint", "token_endpoint", "userinfo_endpoint"]) {
      if (typeof metadata[name] !== "string") throw new Error(`Discovery document is missing ${name}`);
    }
    return metadata;
  }

  async function startLogin() {
    const button = document.getElementById("login");
    button.disabled = true;
    button.textContent = "Preparing sign-in…";
    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 = "Sign in with Swaymoon Account";
      showError(error);
    }
  }

  async function finishLogin() {
    try {
      const query = new URLSearchParams(location.search);
      const savedText = sessionStorage.getItem(STORAGE_KEY);
      if (!savedText) throw new Error("Missing local PKCE state; start again from the home page");
      const saved = JSON.parse(savedText);
      if (Date.now() - Number(saved.createdAt) > 10 * 60 * 1000) {
        throw new Error("PKCE state expired; start again from the home page");
      }
      if (!query.get("state") || query.get("state") !== saved.state) {
        throw new Error("state check failed; start again from the home page");
      }
      if (query.get("error")) {
        throw new Error(`${query.get("error")}: ${query.get("error_description") || "Authorization did not complete"}`);
      }
      const code = query.get("code");
      if (!code) throw new Error("Callback is missing 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("Token response is missing access_token");
      }
      const userinfo = await fetchJson(metadata.userinfo_endpoint, {
        headers: { Authorization: `Bearer ${tokens.access_token}` },
      });
      document.getElementById("status").textContent = "Signed in";
      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='/'>Back to home</a></p>")
        except BrokenPipeError:
            pass
        except Exception:
            detail = traceback.format_exc()
            print(detail)
            try:
                self.send_html(500, f"<h1>Local server error</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("Set SWAYMOON_CLIENT_ID first")
    if os.environ.get("SWAYMOON_CLIENT_SECRET", "").strip():
        raise SystemExit("This page supports public clients only; do not give client_secret to the browser")
    if "openid" not in SCOPE.split():
        raise SystemExit("SWAYMOON_SCOPE must include openid")
    issuer = urllib.parse.urlsplit(ISSUER)
    if issuer.scheme not in ("http", "https") or not issuer.hostname:
        raise SystemExit("SWAYMOON_ISSUER is not a valid 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"Cannot listen on {LOCAL_ORIGIN}: {error}") from error

    print("Client type: public")
    print("Python public-network access: not required; the browser makes OIDC requests")
    print("Issuer: ", ISSUER)
    print("Scope: ", SCOPE)
    print("Browser entry: ", f"{LOCAL_ORIGIN}/")
    print("Redirect URI: ", REDIRECT_URI)
    try:
        webbrowser.open(f"{LOCAL_ORIGIN}/")
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopped")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()

4. Optional configuration

Windows PowerShell example that requests only openid email:

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

To send the callback itself over IPv6, first register http://[::1]:8765/callback in the portal, then set:

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

This only changes the local callback listen address. Public-network requests are always made by the browser under its normal IPv4/IPv6, proxy, and security policy.

5. Common errors

This code is for protocol integration. A production SPA should use a mature OAuth / OIDC library and implement reliable sessions, error handling, and token lifecycle management. Do not write Access Tokens, ID Tokens, or any secrets to logs or persistent storage.

Confidential client

This is a directly runnable Authorization Code + PKCE confidential-client example. The browser only handles redirects. OIDC discovery, authorization-code exchange, JWKS, ID Token verification, and the UserInfo request all happen on the Python server. client_secret and tokens never enter the browser.

The Python process for a confidential client must be able to reach the Issuer. If the current network resets non-browser TLS connections, run it on a normal network, a server, or a container. Do not move token exchange into the browser to work around network issues.

1. Register a client

In the Developer Portal, create a confidential client and register:

Immediately store the client_secret shown only once at creation (you also need it to verify deletion-callback signatures). Keep required openid, and register email, name, nickname, picture, biography, gender, birthdate, region, and preferred_username all as optional scopes. The demo requests every scope, but the user can still decline items one by one on the consent page.

Login Redirect and the deletion callback are not the same thing.
Redirect is visited by the browser, so http://127.0.0.1 is fine. The deletion callback is an outbound POST from the Passport server; the portal does not allow registering a local address. For local integration: start the demo, get https://….ngrok-free.app from a tunnel, and put https://….ngrok-free.app/privacy/data-deletion in the portal. When you choose “Revoke and request data deletion”, the tunnel and the demo must both be running. You can also use the curl below to self-check the signature on this machine (without going through Passport). Whether email is delivered is independent of whether the callback succeeds. See Registration and configuration in the same directory.

2. Install dependencies

The code itself uses the Python standard library for HTTP. The only extra package is cryptography, for ES256 signature verification.

Windows PowerShell / CMD

py -m pip install cryptography

macOS / Linux

python3 -m pip install cryptography

3. Configure and run

Save the next section’s code as swaymoon_confidential_mvp.py.

Windows PowerShell

$env:SWAYMOON_CLIENT_ID = "swm_YOUR_CLIENT_ID"
$env:SWAYMOON_CLIENT_SECRET = "YOUR_CLIENT_SECRET"
py .\swaymoon_confidential_mvp.py

Windows Command Prompt (CMD)

set "SWAYMOON_CLIENT_ID=swm_YOUR_CLIENT_ID"
set "SWAYMOON_CLIENT_SECRET=YOUR_CLIENT_SECRET"
py swaymoon_confidential_mvp.py

macOS / Linux

export SWAYMOON_CLIENT_ID='swm_YOUR_CLIENT_ID'
export SWAYMOON_CLIENT_SECRET='YOUR_CLIENT_SECRET'
python3 swaymoon_confidential_mvp.py

Open http://127.0.0.1:8766/ and choose the sign-in button. After you stop the server, clear the secret environment variables from the current terminal. Do not put real values in code, docs, or Git.

4. Full code

#!/usr/bin/env python3
"""Swaymoon Account: confidential-client PKCE MVP that exchanges tokens and verifies ES256 ID Tokens on the server."""

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()
# Local user profiles keyed by UserInfo / ID Token sub (for the deletion-callback demo)
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 must be a local HTTP URL with a port, "
            "for example http://127.0.0.1:8766/callback"
        )
    if parsed.query or parsed.fragment:
        raise SystemExit("In this example, SWAYMOON_REDIRECT_URI must not contain a query or fragment")
    try:
        loopback = ipaddress.ip_address(parsed.hostname).is_loopback
    except ValueError:
        loopback = parsed.hostname == "localhost"
    if not loopback:
        raise SystemExit("This example only listens on 127.0.0.0/8, localhost, or ::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} returned HTTP {error.code}: {detail}") from error
    except OSError as error:
        raise RuntimeError(f"Cannot connect to {url}: {type(error).__name__}: {error}") from error
    try:
        value = json.loads(raw)
    except json.JSONDecodeError as error:
        raise RuntimeError(f"{url} did not return JSON") from error
    if not isinstance(value, dict):
        raise RuntimeError(f"JSON from {url} is not a top-level object")
    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"Discovery document issuer mismatch: {value.get('issuer')}")
        for name in (
            "authorization_endpoint",
            "token_endpoint",
            "jwks_uri",
            "userinfo_endpoint",
        ):
            if not isinstance(value.get(name), str):
                raise RuntimeError(f"Discovery document is missing {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 format is invalid") from error
    if not isinstance(header, dict) or not isinstance(claims, dict):
        raise RuntimeError("ID Token header/payload format is invalid")
    if header.get("alg") != "ES256" or not isinstance(header.get("kid"), str):
        raise RuntimeError("ID Token must use ES256 with a kid")

    keys = get_jwks().get("keys")
    if not isinstance(keys, list):
        raise RuntimeError("JWKS is missing 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("No P-256 public key for this ID Token in JWKS")
    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 signature length is wrong")
        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 signature verification failed") from error

    now = time.time()
    if claims.get("iss") != ISSUER:
        raise RuntimeError("ID Token iss mismatch")
    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 does not include the current client_id")
    if len(audiences) > 1 and claims.get("azp") != CLIENT_ID:
        raise RuntimeError("ID Token azp mismatch")
    if not isinstance(claims.get("exp"), (int, float)) or claims["exp"] < now - 60:
        raise RuntimeError("ID Token is expired or missing exp")
    if not isinstance(claims.get("iat"), (int, float)) or claims["iat"] > now + 60:
        raise RuntimeError("ID Token iat is invalid")
    if "nbf" in claims and (
        not isinstance(claims["nbf"], (int, float)) or claims["nbf"] > now + 60
    ):
        raise RuntimeError("ID Token is not yet valid")
    if not secrets.compare_digest(str(claims.get("nonce", "")), expected_nonce):
        raise RuntimeError("ID Token nonce check failed")
    if not isinstance(claims.get("sub"), str) or not claims["sub"]:
        raise RuntimeError("ID Token is missing 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:
    """Verify X-Swaymoon-Signature: sha256=<hex> using the plaintext client_secret from creation."""
    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="en"><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='/'>Back to home</a></p>"
        )
        self.send_body(400, page("Sign-in did not complete", 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(
                        "Swaymoon Account · confidential-client PKCE MVP",
                        f"<p>Client: <code>{html.escape(CLIENT_ID)}</code></p>"
                        f"<p>Scopes: <code>{html.escape(SCOPE)}</code></p>"
                        f"<p>Local users: {user_count} · Deletion requests handled: {log_count}</p>"
                        "<p><a class='button' href='/login'>Sign in with Swaymoon Account</a></p>"
                        "<p><a href='/users'>View local users</a> · "
                        "<a href='/deletion-log'>Deletion request log</a></p>"
                        f"<p class='hint'>Data-deletion callback: <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='/'>Back to home</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(
                "Local users",
                "<p>Profiles stored in process memory after a successful sign-in. The matching <code>sub</code> is removed when a deletion callback arrives.</p>"
                f"<pre>{html.escape(snapshot)}</pre><p><a href='/'>Back to home</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(
                "Deletion request log",
                "<p>Written after Swaymoon Account <code>POST</code>s a data-deletion callback.</p>"
                f"<pre>{html.escape(snapshot)}</pre><p><a href='/'>Back to home</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("Request body is not valid 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("Sign-in state is missing or expired; start again from the home page")
        received_state = query.get("state", [""])[0]
        if not received_state or not secrets.compare_digest(received_state, saved["state"]):
            raise RuntimeError("state check failed")
        if query.get("error"):
            description = query.get("error_description", ["Authorization did not complete"])[0]
            raise RuntimeError(f"{query['error'][0]}: {description}")
        code = query.get("code", [""])[0]
        if not code:
            raise RuntimeError("Callback is missing 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("Token response is missing id_token or 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 does not match 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": "Written to process memory by sub, for the deletion-callback demo",
        }
        body = (
            "<p>Signed in. Tokens are handled only in server memory.</p>"
            "<p>In Swaymoon Account, go to Privacy → App authorizations → Revoke and request data deletion "
            "and watch whether this machine’s <code>/deletion-log</code> updates and the local user is cleared.</p>"
            "<pre>"
            + html.escape(json.dumps(summary, ensure_ascii=False, indent=2))
            + "</pre><p><a href='/'>Back to home</a></p>"
        )
        self.send_body(
            200,
            page("Signed in", 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("Set SWAYMOON_CLIENT_ID and SWAYMOON_CLIENT_SECRET first")
    if "openid" not in SCOPE.split():
        raise SystemExit("SWAYMOON_SCOPE must include openid")
    issuer = urllib.parse.urlsplit(ISSUER)
    if issuer.scheme != "https" or not issuer.hostname:
        raise SystemExit("SWAYMOON_ISSUER must be a valid 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"Cannot listen on {LOCAL_ORIGIN}: {error}") from error
    print("Client type: confidential (client_secret_basic + PKCE S256)")
    print("Browser entry: ", f"{LOCAL_ORIGIN}/")
    print("Redirect URI: ", REDIRECT_URI)
    print("Local deletion receiver: ", f"{LOCAL_ORIGIN}{DATA_DELETION_PATH}", " (register a tunnel HTTPS URL in the portal; do not use 127.0.0.1)")
    print("Note: the Python server must be able to reach", ISSUER)
    try:
        webbrowser.open(f"{LOCAL_ORIGIN}/")
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopped")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()

5. Demo the data-deletion callback

The portal does not allow registering http://127.0.0.1 as the data-deletion callback. End-to-end integration steps:

1. Start this demo (listening on 8766).
2. Expose local 8766 with an HTTPS tunnel (example: cloudflared tunnel --url http://127.0.0.1:8766 or ngrok) and get a public URL such as https://xxxx.example.
3. In the Developer Portal, set this app’s “Data-deletion callback URL” to https://xxxx.example/privacy/data-deletion (the path must match SWAYMOON_DATA_DELETION_PATH).
4. Complete sign-in, open http://127.0.0.1:8766/users, and confirm the local store has that user’s sub.
5. Sign in at passport.swaymoon.com, open Privacy → App authorizations, open this demo app’s details, and choose Revoke and request data deletion (step-up required).
6. Swaymoon Account revokes the grant, sends email, and POSTs signed JSON to the tunnel URL. The demo’s /deletion-log should have a record, and the matching sub should disappear from /users.

To self-check the signature without going through Passport, you can still use the curl below against http://127.0.0.1:8766.

Callback request contract (same as Scopes and consent):

You can self-check the signature with a local script (no need to go through Passport):

# macOS / Linux example: SECRET is the plaintext client_secret issued at creation
BODY='{"type":"swaymoon.passport.data_deletion_request","issued_at":"2026-01-01T00:00:00Z","client_id":"swm_demo","subject":"replace-with-real-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. Optional configuration

This demo explicitly sends prompt=consent. Production apps can omit that parameter: Swaymoon Account still shows the authorization confirmation page every time; when scopes change it shows an add/remove comparison; after the user revokes a grant they must authorize again.

7. Security boundaries

• The browser only receives the app’s own pages. It never receives client_secret, Access Token, ID Token, or Refresh Token.
• The example verifies the ES256 signature, iss, aud, azp, exp, iat, nbf, and nonce, and checks UserInfo sub.
• The data-deletion callback verifies X-Swaymoon-Signature (keyed with the plaintext client_secret from creation, over the raw body). Production should also check client_id, restrict source IPs / mTLS, and handle the request idempotently.
• state, nonce, the PKCE verifier, and tokens live only in process memory. After a restart, the sign-in flow is invalid.
• The local HTTP cookie does not set Secure, so it is only for loopback integration. Production must use HTTPS, Secure cookies, durable session storage, and a mature OIDC client library.
• Do not log full callback URLs, authorization codes, secrets, or tokens. This example’s success page only shows filtered claims and status.

8. Common errors