OAuth 与 OIDC

curl https://tokentosea.com/api/oauth/token \
  -d grant_type=authorization_code \
  -d code="$CODE" \
  -d redirect_uri=https://your-app.example.com/callback \
  -d code_verifier="$CODE_VERIFIER" \
  -d client_id="$CLIENT_ID" \
  -d client_secret="$CLIENT_SECRET"

import base64
import hashlib
import secrets

import requests

ISSUER = "https://tokentosea.com/api/oauth"

# 发起授权前
code_verifier = secrets.token_urlsafe(48)  # 64 个字符
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).rstrip(b"=").decode()

# 收到回调后；client_secret 仅机密客户端提供，公共客户端传 None
def exchange(code, client_id, client_secret, redirect_uri):
    data = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": redirect_uri,
        "code_verifier": code_verifier,
        "client_id": client_id,
    }
    if client_secret:  # 仅机密客户端
        data["client_secret"] = client_secret
    response = requests.post(ISSUER + "/token", data=data, timeout=30)
    response.raise_for_status()
    return response.json()

import { createHash, randomBytes } from 'node:crypto';

const ISSUER = 'https://tokentosea.com/api/oauth';

// 发起授权前
const codeVerifier = randomBytes(32).toString('base64url'); // 43 个字符
const codeChallenge = createHash('sha256').update(codeVerifier).digest('base64url');

// 收到回调后
export async function exchange(code, { clientId, clientSecret, redirectUri }) {
  const body = new URLSearchParams({ grant_type: 'authorization_code', code, redirect_uri: redirectUri, code_verifier: codeVerifier, client_id: clientId });
  if (clientSecret) body.set('client_secret', clientSecret); // 仅机密客户端
  const response = await fetch(`${ISSUER}/token`, { method: 'POST', body });
  if (!response.ok) throw new Error(`Token exchange failed (${response.status})`);
  return response.json();
}

# pip install "PyJWT[crypto]"
import hmac

import jwt

ISSUER = "https://tokentosea.com/api/oauth"
jwks_client = jwt.PyJWKClient(ISSUER + "/jwks")


def verify_id_token(id_token: str, client_id: str, expected_nonce: str) -> dict:
    signing_key = jwks_client.get_signing_key_from_jwt(id_token)
    claims = jwt.decode(id_token, signing_key.key, algorithms=["RS256"], audience=client_id, issuer=ISSUER)
    if not hmac.compare_digest(claims.get("nonce", ""), expected_nonce):
        raise ValueError("nonce mismatch")
    return claims

import { createPublicKey, timingSafeEqual, verify } from 'node:crypto';

const ISSUER = 'https://tokentosea.com/api/oauth';
const equal = (a, b) => typeof a === 'string' && typeof b === 'string' && a.length === b.length && timingSafeEqual(Buffer.from(a), Buffer.from(b));

export async function verifyIdToken(token, clientId, nonce) {
  const [head, body, signature] = token.split('.');
  const header = JSON.parse(Buffer.from(head, 'base64url'));
  if (header.alg !== 'RS256') throw new Error('Unsupported algorithm');
  const { keys } = await (await fetch(`${ISSUER}/jwks`)).json(); // 生产环境请缓存
  const jwk = keys.find((k) => k.kid === header.kid && k.use === 'sig' && k.alg === 'RS256');
  if (!jwk || !verify('RSA-SHA256', Buffer.from(`${head}.${body}`), createPublicKey({ key: jwk, format: 'jwk' }), Buffer.from(signature, 'base64url'))) {
    throw new Error('Invalid signature');
  }
  const claims = JSON.parse(Buffer.from(body, 'base64url'));
  const now = Date.now() / 1000;
  if (claims.iss !== ISSUER || claims.aud !== clientId || !(claims.exp > now) || !(claims.iat <= now + 30) || !equal(claims.nonce, nonce)) {
    throw new Error('Invalid claims');
  }
  return claims;
}