OAuth 2.0 and OIDC Implementation Hardening: PKCE, Token Rotation, and JWT Validation Pitfalls

Problem

OAuth 2.0 and OpenID Connect are the dominant authorization and authentication protocols for web and API systems. The specifications are well-designed but complex — 20+ RFCs in the OAuth ecosystem, with extensions, profiles, and security best current practices that have evolved significantly since the original 2012 OAuth 2.0 RFC.

Real-world OAuth/OIDC implementations routinely contain vulnerabilities, many of which appear in the OAuth 2.0 Security Best Current Practice (RFC 9700, published 2024) as known attack patterns:

• Missing PKCE: Authorization code interception attacks exploit the absence of Proof Key for Code Exchange (RFC 7636). Without PKCE, a malicious app or a network attacker can steal the authorization code and exchange it for tokens.
• Broad token scopes: Tokens with openid profile email admin when the client only needs openid email. A stolen token grants all declared scopes.
• Long-lived access tokens: Tokens valid for 24 hours or more. A leaked token provides long-term access.
• JWT validation shortcuts: Applications that decode JWTs without verifying the signature, or that accept alg: none, are trivially bypassed. The iss and aud claims are frequently unchecked.
• No refresh token rotation: Long-lived refresh tokens without rotation. If a refresh token is leaked, it provides indefinite access.
• Implicit flow still in use: The implicit flow (directly returning tokens in the URL fragment) was deprecated in OAuth 2.1 and OIDC Security Profile. Fragment tokens appear in browser history, referer headers, and server logs.
• State parameter not validated: CSRF attacks on the authorization endpoint exploit absent or unchecked state parameter validation.
• Redirect URI not exactly matched: Authorization servers that do partial or substring matching on redirect URIs allow redirect to attacker-controlled domains.

By 2026, the implicit flow is effectively deprecated; authorization code + PKCE is the correct pattern for all client types. OAuth 2.1 (currently in final draft) codifies this as mandatory.

Target systems: Any OAuth 2.0 / OIDC authorization server (Keycloak 24+, Auth0, Okta, AWS Cognito, Azure Entra); client implementation using libraries in Go, Python, Node, or Java; API gateways performing JWT validation (Envoy, NGINX, Kong).

Threat Model

• Adversary 1 — Authorization code interception: A malicious application registered on the same device intercepts the authorization code redirect (via custom URL scheme hijacking on mobile, or via a tab-nabbing attack on web). Without PKCE, it exchanges the code for tokens.
• Adversary 2 — Token leakage via referer header: An application using the implicit flow embeds the access token in the redirect URI fragment. A third-party resource loaded by the application receives the token in the Referer header.
• Adversary 3 — JWT algorithm confusion attack: An API validates JWTs using the public key from JWKS endpoint. An attacker crafts a JWT with alg: HS256 (symmetric HMAC) and signs it using the server’s public key as the HMAC secret. An implementation that trusts the alg header field accepts it.
• Adversary 4 — Refresh token replay: A long-lived refresh token is exfiltrated. Without rotation, the attacker uses it indefinitely — even after the legitimate user changes their password.
• Adversary 5 — Open redirect via loose URI matching: The authorization server performs prefix matching on redirect URIs. An attacker registers a client with redirect_uri=https://legit.example.com.evil.com/ (or https://legit.example.com/ but is allowed to supply https://legit.example.com/../../attacker). Authorization codes or tokens are redirected to the attacker.
• Access level: Adversaries 1–3 have network-level or local access. Adversary 4 has access to a persistent storage system where tokens are cached. Adversary 5 needs a client registration.
• Objective: Obtain valid access tokens for unauthorized access; hijack user sessions; impersonate users.
• Blast radius: An unprotected OAuth/OIDC flow is equivalent to a password leak for all users who authenticate via that flow. Token scope determines what the attacker can do; an admin-scoped token provides full system access.

Configuration

Step 1: Enforce PKCE for All Authorization Code Flows

PKCE (RFC 7636) prevents authorization code interception by binding the code to a secret known only to the legitimate client.

Client-side (code verifier and challenge generation):

import base64, hashlib, os, secrets

def generate_pkce_pair():
    # Code verifier: 43-128 chars, URL-safe random.
    code_verifier = secrets.token_urlsafe(64)

    # Code challenge: SHA-256 of the verifier, base64url-encoded.
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()

    return code_verifier, code_challenge

code_verifier, code_challenge = generate_pkce_pair()

# Include in authorization request.
auth_url = (
    f"{AUTHORIZATION_ENDPOINT}"
    f"?response_type=code"
    f"&client_id={CLIENT_ID}"
    f"&redirect_uri={REDIRECT_URI}"
    f"&scope=openid email"
    f"&state={secrets.token_urlsafe(32)}"   # CSRF protection.
    f"&code_challenge={code_challenge}"
    f"&code_challenge_method=S256"
)

Token exchange (include verifier):

def exchange_code_for_tokens(code: str, code_verifier: str) -> dict:
    response = requests.post(
        TOKEN_ENDPOINT,
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": REDIRECT_URI,
            "client_id": CLIENT_ID,
            "code_verifier": code_verifier,   # Server verifies S256(verifier) == challenge.
        },
        auth=(CLIENT_ID, CLIENT_SECRET),    # Confidential client sends credentials.
    )
    response.raise_for_status()
    return response.json()

Authorization server configuration (Keycloak):

# Require PKCE for a specific client.
kcadm.sh update clients/<client-id> -r <realm> \
  -s 'attributes={"pkce.code.challenge.method":"S256"}'

For public clients (SPAs, mobile apps) where there is no client secret, PKCE is the only security mechanism. It must not be optional.

Step 2: Minimize Token Scope

Request only the scopes the client actually needs. Never request admin or write scopes speculatively.

# Bad: requesting all available scopes.
scope = "openid profile email admin write:all read:all"

# Good: minimum scopes for the operation.
scope = "openid email"   # For authentication only.
scope = "openid email profile:read"   # For profile display.
scope = "openid email resource:read"  # For resource access.

On the authorization server, configure per-client scope restrictions:

// Keycloak client scope configuration.
{
  "clientId": "payments-frontend",
  "defaultClientScopes": ["openid", "email"],
  "optionalClientScopes": ["profile"],
  // Prevent this client from requesting admin scopes even if present in the realm.
  "fullScopeAllowed": false
}

Validate scope on the resource server — not just on the authorization server:

def require_scope(required_scope: str):
    def decorator(f):
        def wrapper(*args, **kwargs):
            token = get_current_token()
            scopes = token.get("scope", "").split()
            if required_scope not in scopes:
                raise HTTPException(status_code=403, detail=f"Required scope: {required_scope}")
            return f(*args, **kwargs)
        return wrapper
    return decorator

@app.route("/api/payments", methods=["POST"])
@require_scope("payments:write")
def create_payment():
    ...

Step 3: Short-Lived Access Tokens with Refresh Token Rotation

Set access token lifetime to 5–15 minutes. Use refresh tokens for long-lived sessions, with rotation on every use.

Authorization server configuration:

# Keycloak realm settings.
kcadm.sh update realms/<realm> \
  -s accessTokenLifespan=300 \         # 5 minutes.
  -s ssoSessionMaxLifespan=28800 \     # 8 hours total session.
  -s refreshTokenMaxReuse=0            # Require rotation on every use.

Client-side token management:

import time

class TokenManager:
    def __init__(self):
        self._access_token = None
        self._refresh_token = None
        self._expires_at = 0

    def get_access_token(self) -> str:
        if time.time() >= self._expires_at - 30:   # 30s buffer.
            self._refresh()
        return self._access_token

    def _refresh(self):
        response = requests.post(
            TOKEN_ENDPOINT,
            data={
                "grant_type": "refresh_token",
                "refresh_token": self._refresh_token,
                "client_id": CLIENT_ID,
            },
            auth=(CLIENT_ID, CLIENT_SECRET),
        )
        response.raise_for_status()
        data = response.json()

        self._access_token = data["access_token"]
        self._refresh_token = data["refresh_token"]   # New rotated refresh token.
        self._expires_at = time.time() + data["expires_in"]

With refresh token rotation, each use of a refresh token invalidates it and issues a new one. A leaked refresh token is detected: when the attacker uses it, the legitimate client’s next refresh attempt fails (the token is already consumed), generating an alert.

Step 4: JWT Validation — Do It Correctly

JWT validation is the most frequently botched step in OAuth/OIDC implementations. The complete checklist:

import jwt   # PyJWT 2.x
from jwt import PyJWKClient

JWKS_URL = f"{ISSUER}/.well-known/jwks.json"

# Cache the JWKS client (fetches and caches public keys).
jwks_client = PyJWKClient(JWKS_URL, cache_jwk_set=True, lifespan=360)

def validate_access_token(token: str) -> dict:
    # 1. Get the signing key that matches the token's `kid` header.
    signing_key = jwks_client.get_signing_key_from_jwt(token)

    # 2. Verify signature + standard claims.
    payload = jwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256", "ES256"],   # Explicit allowlist; NEVER include "none" or "HS256" for RS/ES tokens.
        audience=CLIENT_ID,              # Verify `aud` claim.
        issuer=ISSUER,                   # Verify `iss` claim.
        options={
            "verify_exp": True,          # Always.
            "verify_nbf": True,          # Not-before.
            "verify_iat": True,          # Issued-at.
            "verify_aud": True,          # Audience.
            "verify_iss": True,          # Issuer.
            "require": ["exp", "iat", "sub", "aud", "iss"],
        },
        leeway=10,                       # 10s clock skew tolerance.
    )

    # 3. Validate additional business claims.
    if payload.get("token_use") not in ("access", None):
        raise ValueError(f"Unexpected token_use: {payload.get('token_use')}")

    return payload

Critical validations in order:

1. Algorithm restriction: Use an explicit allowlist of ["RS256"] or ["ES256"]. Never include "none" or allow the alg header to control the verification algorithm unchecked.
2. Signature verification: Fetch the JWKS from the authorization server; match the key by kid header field.
3. iss (issuer): Reject tokens from unexpected issuers. Common mistake: accepting any token that is a valid JWT without checking iss.
4. aud (audience): Reject tokens not intended for this resource server. Without this check, a token issued for service A can be replayed at service B.
5. exp (expiration): Reject expired tokens. Apply only a small leeway (10–30s) for clock skew.
6. nbf (not before) and iat (issued at): Reject tokens with invalid time bounds.

Step 5: State Parameter and Redirect URI Validation

State parameter:

import secrets

def start_auth_flow(session):
    state = secrets.token_urlsafe(32)
    session["oauth_state"] = state
    session["oauth_nonce"] = secrets.token_urlsafe(32)
    # ... build authorization URL with state and nonce ...

def handle_callback(request, session):
    # Validate state before anything else.
    received_state = request.args.get("state")
    expected_state = session.pop("oauth_state", None)

    if not expected_state or not secrets.compare_digest(
        received_state.encode(), expected_state.encode()
    ):
        raise SecurityError("State mismatch — possible CSRF attack")

    # Validate nonce in ID token.
    id_token = validate_id_token(request.args.get("code"))
    if id_token.get("nonce") != session.pop("oauth_nonce", None):
        raise SecurityError("Nonce mismatch — possible replay attack")

Redirect URI validation (authorization server side):

Configure your authorization server to require exact match, not prefix or substring match:

# Keycloak: valid redirect URIs must be exact or use wildcards explicitly.
kcadm.sh update clients/<client-id> -r <realm> \
  -s 'redirectUris=["https://app.example.com/callback"]'
# NOT "https://app.example.com/*" — too broad.

For multi-environment deployments, register each environment separately:

{
  "redirectUris": [
    "https://app.example.com/callback",
    "https://staging.example.com/callback",
    "http://localhost:3000/callback"   // Dev only; remove in production client.
  ]
}

Step 6: Token Storage Security

Where tokens are stored determines how they can be stolen:

The Backend-for-Frontend pattern:

Browser ──── session cookie ────► BFF Server ──── access token ────► Resource API
                                    (holds tokens server-side; browser never sees them)

# BFF: exchange auth code server-side; store tokens in server session.
@app.route("/api/auth/callback")
def auth_callback():
    code = request.args.get("code")
    tokens = exchange_code_for_tokens(code, session.pop("pkce_verifier"))
    session["access_token"] = tokens["access_token"]
    session["refresh_token"] = tokens["refresh_token"]
    return redirect("/dashboard")

@app.route("/api/resource")
def proxy_resource():
    token = session.get("access_token")
    if not token or is_expired(token):
        token = refresh_access_token(session)
    resp = requests.get(RESOURCE_API, headers={"Authorization": f"Bearer {token}"})
    return resp.json()

Step 7: Token Revocation and Introspection

For sensitive operations, validate tokens are not revoked using the introspection endpoint (RFC 7662):

def introspect_token(token: str) -> dict:
    response = requests.post(
        INTROSPECTION_ENDPOINT,
        data={"token": token, "token_type_hint": "access_token"},
        auth=(RESOURCE_SERVER_CLIENT_ID, RESOURCE_SERVER_CLIENT_SECRET),
    )
    result = response.json()
    if not result.get("active"):
        raise TokenRevoked("Token is not active")
    return result

Use introspection selectively — it adds latency (network round-trip to auth server). Apply it for:

• Admin operations or high-privilege actions.
• Operations following a reported credential compromise.
• Any action with financial or data-exfiltration potential.

For normal API calls, JWT local validation is sufficient (short-lived tokens bound the revocation window to the token lifetime).

Step 8: Telemetry

oauth_token_issued_total{client_id, scope, grant_type}
oauth_token_validation_failure_total{reason, client_id}
oauth_token_refresh_total{client_id, result}
oauth_pkce_validation_failure_total{client_id}
oauth_state_mismatch_total{client_id}
oauth_redirect_uri_mismatch_total{client_id}
oauth_refresh_token_reuse_detected_total{client_id}

Alert on:

• oauth_refresh_token_reuse_detected_total non-zero — a rotated refresh token was replayed; possible token exfiltration.
• oauth_token_validation_failure_total{reason="alg_mismatch"} — possible algorithm confusion attack.
• oauth_state_mismatch_total non-zero — possible CSRF on the authorization endpoint.
• oauth_redirect_uri_mismatch_total — attempt to redirect to an unregistered URI; possible open redirect probe.

Expected Behaviour

Trade-offs

Failure Modes

Related Articles

• OIDC Federation Hardening for CI-to-Cloud
• Secrets Rotation Orchestration
• API Key Lifecycle at Scale
• Production Access Management with Teleport and Boundary
• Zero-Trust Networking