OAuth2 Authorization Code with PKCE Deep Dive — Securing Public Clients

// PKCE utility functions
function base64URLEncode(buffer) {
  return btoa(String.fromCharCode(...new Uint8Array(buffer)))
    .replace(/\+/g, '-')
    .replace(/\//g, '_')
    .replace(/=+$/, '');
}

async function generateCodeVerifier() {
  const array = new Uint8Array(64);
  crypto.getRandomValues(array);
  return base64URLEncode(array);
}

async function generateCodeChallenge(verifier) {
  const encoder = new TextEncoder();
  const data = encoder.encode(verifier);
  const digest = await crypto.subtle.digest('SHA-256', data);
  return base64URLEncode(digest);
}

// Step 1: Initiate PKCE login
async function initiatePKCELogin() {
  const verifier = await generateCodeVerifier();
  const challenge = await generateCodeChallenge(verifier);

  // Store verifier in session storage (temporary, same session only)
  sessionStorage.setItem('pkce_verifier', verifier);

  const params = new URLSearchParams({
    response_type: 'code',
    client_id: 'dodatech-spa',
    redirect_uri: 'https://app.dodatech.com/callback',
    code_challenge: challenge,
    code_challenge_method: 'S256',
    state: crypto.randomUUID(),
    scope: 'openid profile email'
  });

  window.location.href =
    `https://auth.dodatech.com/authorize?${params}`;
}

// Step 2: Handle callback and exchange code
async function handleCallback() {
  const params = new URLSearchParams(window.location.search);
  const code = params.get('code');
  const verifier = sessionStorage.getItem('pkce_verifier');

  if (!code || !verifier) return;

  const resp = await fetch('https://auth.dodatech.com/token', {
    method: 'POST',
    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
    body: new URLSearchParams({
      grant_type: 'authorization_code',
      code: code,
      redirect_uri: 'https://app.dodatech.com/callback',
      client_id: 'dodatech-spa',
      code_verifier: verifier
    })
  });

  const tokens = await resp.json();
  sessionStorage.removeItem('pkce_verifier');

  return tokens;
}

import hashlib, base64, secrets
from flask import Flask, request, jsonify
from urllib.parse import urlencode

app = Flask(__name__)

# In-memory store (use Redis in production)
authorization_codes = {}
clients = {
    "dodatech-spa": {
        "redirect_uris": ["https://app.dodatech.com/callback"],
        "public": True  # No client_secret for public clients
    }
}

def base64url_decode(value):
    return base64.urlsafe_b64decode(value + '==')

def base64url_encode(value):
    return base64.urlsafe_b64encode(value).rstrip(b'=').decode()

def verify_pkce(code_verifier, code_challenge, method):
    if method == 'plain':
        return code_verifier == code_challenge
    elif method == 'S256':
        computed = base64url_encode(
            hashlib.sha256(code_verifier.encode()).digest()
        )
        return computed == code_challenge
    return False

@app.route("/authorize", methods=["GET"])
def authorize():
    client_id = request.args.get("client_id")
    redirect_uri = request.args.get("redirect_uri")
    code_challenge = request.args.get("code_challenge")
    challenge_method = request.args.get("code_challenge_method", "S256")

    client = clients.get(client_id)
    if not client or redirect_uri not in client["redirect_uris"]:
        return "Invalid client", 400

    auth_code = secrets.token_urlsafe(32)
    authorization_codes[auth_code] = {
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "code_challenge": code_challenge,
        "challenge_method": challenge_method,
        "expires_at": __import__('time').time() + 60
    }

    params = urlencode({"code": auth_code, "state": request.args.get("state", "")})
    return f"Redirect to {redirect_uri}?{params}"

@app.route("/token", methods=["POST"])
def token():
    code = request.form.get("code")
    verifier = request.form.get("code_verifier")
    client_id = request.form.get("client_id")

    stored = authorization_codes.get(code)
    if not stored:
        return jsonify({"error": "invalid_grant"}), 400

    if __import__('time').time() > stored["expires_at"]:
        return jsonify({"error": "code_expired"}), 400

    # PKCE verification
    if not verify_pkce(verifier, stored["code_challenge"], stored["challenge_method"]):
        return jsonify({"error": "invalid_grant", "description": "PKCE verification failed"}), 400

    # Clean up used code
    del authorization_codes[code]

    # Issue tokens
    access_token = secrets.token_urlsafe(32)
    refresh_token = secrets.token_urlsafe(32)

    return jsonify({
        "access_token": access_token,
        "token_type": "Bearer",
        "expires_in": 3600,
        "refresh_token": refresh_token
    })

# Additional method: using SHA-512 for stronger binding
import hashlib

def generate_code_challenge(verifier, method="S256"):
    if method == "S256":
        digest = hashlib.sha256(verifier.encode()).digest()
    elif method == "S512":
        digest = hashlib.sha512(verifier.encode()).digest()
    else:
        return verifier  # plain

    return base64.urlsafe_b64encode(digest).rstrip(b'=').decode()

# Verifier validation
def validate_verifier(verifier):
    """Validate code_verifier per RFC 7636."""
    if len(verifier) < 43 or len(verifier) > 128:
        return False
    # Must contain only unreserved characters
    import re
    return bool(re.match(r'^[A-Za-z0-9\-._~]+$', verifier))