OAuth 2.0 and OpenID Connect

Delegation and Login, Wired Up Correctly

The authentication concepts lesson explains the split: OAuth 2.0 lets an application act on a user's behalf against an API, and OpenID Connect (OIDC) adds a verified statement of who that user is. Here we implement both: picking a flow, validating what comes back, and testing your integration for the mistakes that turn "Sign in with..." into an account takeover.

Parties and Client Types

Four parties take part: the user (resource owner), your app (the client), the authorization server (AS) that authenticates the user and issues tokens (Google, Entra ID, Okta, Keycloak), and the resource server, the API that accepts access tokens. Clients come in two kinds. A confidential client runs on a server and can keep a client secret. A public client (a single-page app, a mobile app, a CLI) ships its code to the user, so any "secret" in it is readable by anyone and must not be relied on.

Which Flow to Use

The OAuth 2.0 Security Best Current Practice (RFC 9700) advises against the implicit and password grants and requires PKCE for public clients; the OAuth 2.1 draft extends PKCE to every client using the code flow. Use it everywhere.

Authorization Code with PKCE, Step by Step

1. The client redirects the browser to the AS. Before redirecting, it generates three random values and stores them in the user's session: state, nonce and a PKCE code_verifier. It sends the SHA-256 hash of the verifier as the code_challenge:

GET /authorize?response_type=code
    &client_id=webapp-123
    &redirect_uri=https%3A%2F%2Fapp.example.com%2Fauth%2Fcallback
    &scope=openid%20email%20profile
    &state=Xq3...&nonce=p9L...
    &code_challenge=Hk2...&code_challenge_method=S256 HTTP/1.1
Host: id.example.com

2. The user authenticates at the AS and consents. Your app never sees the password.

3. The AS redirects back to the registered redirect_uri with a short-lived, single-use code and the same state: https://app.example.com/auth/callback?code=SplxlOBe...&state=Xq3...

4. The client exchanges the code at the token endpoint, server to server: a form-encoded POST with grant_type=authorization_code, the code, the same redirect_uri and the original code_verifier, authenticated with the client credentials. The AS hashes the verifier, compares it with the challenge from step 1, and returns the tokens:

{
  "access_token": "eyJhbGciOiJSUzI1NiIs...",
  "token_type": "Bearer",
  "expires_in": 3600,
  "refresh_token": "8xLOxBtZp8",
  "id_token": "eyJhbGciOiJSUzI1NiIsImtpZCI6..."
}

Each random value closes a specific gap:

• PKCE binds the code to the client instance that started the flow. A code that leaks (through a log, a Referer header, or a malicious app registered for the same custom URL scheme on a phone) is useless without the verifier, which never left your side.
• state binds the callback to the browser session that started the login. Without it, another party can complete their own login flow in your user's browser, silently signing the victim into the wrong account (login CSRF, the OAuth variant of the attack in the CSRF lesson).
• nonce is copied by the AS into the ID token, so the client can reject an ID token that was not issued for this login attempt.

A Working OIDC Login in Flask

In production use a maintained library (Authlib or the provider's SDK); writing it once shows what that library must do. This client uses requests and PyJWT (pip install flask requests "pyjwt[crypto]"):

import base64, hashlib, os, secrets
from urllib.parse import urlencode

import jwt
import requests
from flask import Flask, abort, redirect, request, session

app = Flask(__name__)
app.secret_key = os.environ["FLASK_SECRET_KEY"]

ISSUER = "https://id.example.com"
CLIENT_ID = os.environ["OIDC_CLIENT_ID"]
CLIENT_SECRET = os.environ["OIDC_CLIENT_SECRET"]
REDIRECT_URI = "https://app.example.com/auth/callback"

# Discovery: every OIDC provider publishes its endpoints here
meta = requests.get(f"{ISSUER}/.well-known/openid-configuration", timeout=10).json()
jwks = jwt.PyJWKClient(meta["jwks_uri"])

def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")

@app.get("/login")
def login():
    verifier = b64url(secrets.token_bytes(32))
    session["oidc"] = {
        "state": secrets.token_urlsafe(32),
        "nonce": secrets.token_urlsafe(32),
        "verifier": verifier,
    }
    params = {
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": "openid email profile",
        "state": session["oidc"]["state"],
        "nonce": session["oidc"]["nonce"],
        "code_challenge": b64url(hashlib.sha256(verifier.encode()).digest()),
        "code_challenge_method": "S256",
    }
    return redirect(f"{meta['authorization_endpoint']}?{urlencode(params)}")

@app.get("/auth/callback")
def callback():
    pending = session.pop("oidc", None)          # one use only
    if not pending or request.args.get("state") != pending["state"]:
        abort(400, "state mismatch")
    if "error" in request.args:
        abort(400, request.args["error"])

    resp = requests.post(meta["token_endpoint"], data={
        "grant_type": "authorization_code",
        "code": request.args["code"],
        "redirect_uri": REDIRECT_URI,
        "code_verifier": pending["verifier"],
    }, auth=(CLIENT_ID, CLIENT_SECRET), timeout=10)
    resp.raise_for_status()
    tokens = resp.json()

    id_token = tokens["id_token"]
    key = jwks.get_signing_key_from_jwt(id_token).key
    claims = jwt.decode(
        id_token, key,
        algorithms=["RS256"],
        audience=CLIENT_ID,
        issuer=ISSUER,
        options={"require": ["exp", "iat", "iss", "aud", "sub", "nonce"]},
        leeway=30,
    )
    if claims["nonce"] != pending["nonce"]:
        abort(400, "nonce mismatch")

    session.clear()                               # new session after login
    session["user"] = {"iss": claims["iss"], "sub": claims["sub"]}
    return redirect("/")

The lines that matter:

• session.pop("oidc") makes the state single-use, so a replayed callback URL fails.
• The state check runs before anything else, including the error branch.
• PyJWKClient fetches the provider's public keys and picks the one matching the token's kid, so key rotation at the provider works without a redeploy.
• algorithms=["RS256"] is pinned by you. Accepting whatever alg the token header claims is the classic JWT validation bug.
• The user is keyed by iss + sub, never by email. sub is stable per issuer; an email can change, be reused, or be unverified at some providers. If you link accounts by email, require email_verified and only trust issuers that actually verify it.

Checking That Validation Actually Rejects Things

The happy path proves little. Wrap the decode in a function and feed it tokens signed with a local test key, each with one claim wrong. With the settings above, every case is rejected:

wrong aud: InvalidAudienceError: Audience doesn't match
expired: ExpiredSignatureError: Signature has expired
replayed nonce: InvalidTokenError: nonce mismatch
no nonce: MissingRequiredClaimError: Token is missing the "nonce" claim

Keep those cases in your test suite. Do the same for the callback: a missing or different state must return 400, and replaying a callback URL must fail.

Redirect URI Validation

If an AS accepts a redirect URI other than the registered one, codes (and, with implicit flow, tokens) can be delivered somewhere the client does not control. If you operate an authorization server, compare the full string exactly:

REGISTERED = {"webapp-123": {"https://app.example.com/auth/callback"}}

def is_allowed_redirect(client_id: str, redirect_uri: str) -> bool:
    return redirect_uri in REGISTERED.get(client_id, set())

Compare it with a tempting redirect_uri.startswith(registered) check over a few variants (exact result, prefix result, URI):

True  True  https://app.example.com/auth/callback
False True  https://app.example.com/auth/callback/../../logout
False True  https://app.example.com/auth/callback?next=/admin
False True  https://app.example.com/auth/callbackx
False False http://app.example.com/auth/callback

Prefix, substring, regex and wildcard-subdomain matching all allow URIs you never registered, and an open redirect anywhere on an allowed path can then forward the code off-site. The one standard exception is native apps using a loopback redirect (http://127.0.0.1/...), where RFC 8252 requires the AS to accept any port because the app picks a free one at runtime.

As a client developer, audit the provider console: exact callback URLs only, no leftover localhost or staging entries on the production client, and no page under the callback path that redirects based on a query parameter.

Other Common Mistakes

• Implicit flow in an SPA. Tokens arrive in the URL fragment, where history, extensions and any script on the page can read them, and PKCE cannot protect them. Migrate to code + PKCE.
• Tokens in localStorage. Any XSS reads them. The current recommendation for browser apps is a backend-for-frontend (BFF): your server runs the code flow, keeps tokens server-side, and gives the browser an HttpOnly; Secure; SameSite session cookie. If the SPA must call an API on another origin, configure that API as described in CORS and its misconfigurations.
• Using the wrong token for the job. The ID token tells the client who logged in (aud is your client id); the access token is for the API (aud is the API). Never send ID tokens to APIs or treat an access token as proof of login.
• Over-broad scopes. Request the minimum; ask for more when a feature needs it.
• Client secrets in public clients. A secret shipped in a mobile app or SPA is public; register those as public clients and rely on PKCE. Keep real secrets in a manager, per secrets management.
• Refresh tokens without rotation. Enable rotation at the provider so each refresh returns a new refresh token and invalidates the old one; reuse of an old one signals theft.

Client Credentials for Service-to-Service Calls

With no user, the service authenticates as itself and caches the token until shortly before it expires:

import time, requests

_cache = {"token": None, "exp": 0}

def service_token() -> str:
    if time.time() < _cache["exp"] - 60:        # refresh a minute early
        return _cache["token"]
    r = requests.post("https://id.example.com/token",
                      data={"grant_type": "client_credentials",
                            "scope": "invoices:read"},
                      auth=(CLIENT_ID, CLIENT_SECRET), timeout=10)
    r.raise_for_status()
    body = r.json()
    _cache.update(token=body["access_token"], exp=time.time() + body["expires_in"])
    return _cache["token"]

Give each service its own client and narrow scopes, so one leaked secret does not open every API.

On the API Side: Checking Access Tokens

The API checks issuer, audience (this API), expiry, and the scope the endpoint needs:

def require_scope(token: str, needed: str) -> dict:
    key = jwks.get_signing_key_from_jwt(token).key
    claims = jwt.decode(token, key, algorithms=["RS256"],
                        audience="https://api.example.com", issuer=ISSUER)
    if needed not in claims.get("scope", "").split():
        raise PermissionError(f"missing scope {needed}")
    return claims

Scope says what the client may do; it does not replace per-object authorization. The API must still check that the user in sub owns the invoice being read, or you have an IDOR. If your provider issues opaque (non-JWT) access tokens, validate them by calling its token introspection endpoint (RFC 7662), which returns "active": true plus the claims, and cache the result briefly.

Practice

Register a test client with a local Keycloak container or a free developer tenant at any OIDC provider, run the Flask client against it, and then break it on purpose: remove the state check, drop audience= from the decode, and register a wildcard redirect URI. For each change, write the failing test that would have caught it before you restore the code.