OAuth 2.0 Deep Dive

// Generate PKCE verifier and challenge
function generatePKCE() {
  // Step 1: Generate a cryptographically random code_verifier (43-128 chars)
  const array = new Uint8Array(32);
  crypto.getRandomValues(array);
  const codeVerifier = base64URLEncode(array);

  // Step 2: Derive code_challenge from code_verifier
  // SHA-256 hash, then base64url encode
  const encoder = new TextEncoder();
  const data = encoder.encode(codeVerifier);
  crypto.subtle.digest('SHA-256', data).then((hash) => {
    const codeChallenge = base64URLEncode(new Uint8Array(hash));
  });

  return { codeVerifier, codeChallenge };
}

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

// Step 3: Send code_challenge in authorization request
// Step 4: Send code_verifier in token request (server verifies challenge)

1. Authorization Request (browser redirect):
   GET /authorize?
     response_type=code
     &client_id=my_app_id
     &redirect_uri=https://myapp.com/callback
     &scope=openid profile email
     &state=abc123
     &code_challenge=E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM
     &code_challenge_method=S256

2. User authenticates and consents

3. Authorization Response (redirect):
   https://myapp.com/callback?
     code=SplxlOBeZQQYbYS6WxSbIA
     &state=abc123

4. Token Request (server-to-server, POST):
   POST /token
   Content-Type: application/x-www-form-urlencoded

   grant_type=authorization_code
   &code=SplxlOBeZQQYbYS6WxSbIA
   &client_id=my_app_id
   &client_secret=my_app_secret
   &redirect_uri=https://myapp.com/callback
   &code_verifier=dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk

5. Token Response:
   {
     "access_token": "eyJhbGci...",
     "token_type": "Bearer",
     "expires_in": 3600,
     "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA",
     "id_token": "eyJhbGci..."
   }

1. POST /device/authorize
   Response:
   {
     "device_code": "GmRhmhXhwThko6mVU...",
     "user_code": "WDJB-MJHT",
     "verification_uri": "https://example.com/device",
     "expires_in": 1800,
     "interval": 5
   }

2. User visits verification_uri and enters user_code

3. Client polls POST /token with device_code until user completes authorization

eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.   # Header (base64url)
eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.   # Payload (base64url)
SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c   # Signature

def validate_jwt(token, expected_issuer, expected_audience, jwks_uri):
    # 1. Verify signature using JWKS from the authorization server
    jwks = fetch_jwks(jwks_uri)
    public_key = jwks.get_key(token.header['kid'])
    verify_signature(token, public_key)

    # 2. Verify issuer
    assert token.payload['iss'] == expected_issuer

    # 3. Verify audience
    assert token.payload['aud'] in [expected_audience]

    # 4. Verify expiration
    assert token.payload['exp'] > time.time()

    # 5. Verify not-before (if present)
    if 'nbf' in token.payload:
        assert token.payload['nbf'] <= time.time()

    # 6. Verify issuer-signed claims (e.g., email_verified)
    return token.payload

1. POST /token with refresh_token=ABC
   Response: { access_token: "new_access", refresh_token: "XYZ" }

2. ABC is now invalid. Use XYZ for the next refresh.
3. If ABC is used again, ALL tokens for this user are revoked (theft detected).

{
  "iss": "https://auth.example.com",
  "sub": "user-12345",
  "aud": "my_app_id",
  "exp": 1700000000,
  "iat": 1699996400,
  "email": "alice@example.com",
  "email_verified": true,
  "name": "Alice Smith",
  "picture": "https://example.com/alice.jpg"
}

// Recommended: httpOnly cookie for tokens (set by server)
// Set-Cookie: access_token=xxx; HttpOnly; Secure; SameSite=Strict; Path=/; Max-Age=3600

// SPA: store access token in memory, refresh token in httpOnly cookie
let accessToken = null;

// On page load, use refresh token to get a new access token
async function initializeAuth() {
  const response = await fetch('/api/auth/refresh', {
    method: "POST'',
    credentials: "include', // sends httpOnly cookie
  });
  const data = await response.json();
  accessToken = data.access_token;
}

// Access token is lost on page refresh (acceptable: silent refresh restores it)

// Create a DPoP proof for each request
async function createDPoPProof(url, method) {
  const jwk = await crypto.subtle.generateKey({ name: "ECDSA'', namedCurve: "P-256' }, true, [
    'sign',
  ]);

  const header = { typ: "dpop+jwt'', jwk: exportJWK(jwk) };
  const payload = {
    jti: crypto.randomUUID(),
    htu: url,
    htm: method,
    iat: Math.floor(Date.now() / 1000),
  };

  const signed = await signJWT(header, payload, jwk);
  return signed;
}

// Include DPoP header in every request
fetch("/api/resource', {
  headers: {
    Authorization: "Bearer xxx'',
    DPoP: dpopProof,
  },
});

# ALWAYS validate aud and iss on every token
def validate_token(token):
    claims = decode_jwt(token)

    # Validate issuer
    if claims["iss'] != 'https://auth.example.com':
        raise InvalidTokenError(f"Wrong issuer: {claims['iss']}")

    # Validate audience (token must be intended for your service)
    if 'my-api-id' not in claims.get('aud', []):
        raise InvalidTokenError(f"Wrong audience: {claims.get('aud')}")

    return claims

// VULNERABLE: Accept any redirect URI
app.get('/callback', (req, res) => {
  const redirectUri = req.query.redirect_uri;
  // If attacker provides redirect_uri=https://evil.com, the auth code goes to evil.com
});

// SAFE: Whitelist redirect URIs
const ALLOWED_REDIRECTS = ['https://myapp.com/callback', 'https://admin.myapp.com/callback'];

app.get('/callback', (req, res) => {
  const redirectUri = req.query.redirect_uri;
  if (!ALLOWED_REDIRECTS.includes(redirectUri)) {
    return res.status(400).send('Invalid redirect URI');
  }
});

// Step 1: Generate state before redirecting to auth server
const state = crypto.randomUUID();
session.oauthState = state; // Store in server-side session

// Step 2: Include state in authorization request
const authUrl = `https://auth.example.com/authorize?state=${state}&...`;

// Step 3: On callback, verify state matches
app.get('/callback', (req, res) => {
  if (req.query.state !== session.oauthState) {
    return res.status(400).send('CSRF detected: state mismatch');
  }
  session.oauthState = null; // Consume the state
  // Proceed with token exchange
});

def require_scope(required_scope):
    def decorator(f):
        def wrapper(*args, **kwargs):
            token = get_bearer_token()
            claims = validate_token(token)
            if required_scope not in claims.get('scope', '').split():
                raise ForbiddenError("Insufficient scope")
            return f(*args, **kwargs)
        return wrapper
    return decorator

@app.route('/api/admin/users')
@require_scope('admin:users')
def list_users():
    ...