OAuth Client Flows — Deep Dive

Implementing OAuth flows in Python

Python has mature OAuth libraries that handle the protocol complexity. The two leading options are authlib (full-featured, supports both client and server) and requests-oauthlib (simpler, requests-only). This guide uses authlib with httpx for modern async support.

Authorization Code flow with authlib

A complete web application flow using FastAPI and authlib:

from authlib.integrations.httpx_client import AsyncOAuth2Client
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
import secrets

app = FastAPI()

CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
AUTHORIZE_URL = "https://provider.com/oauth/authorize"
TOKEN_URL = "https://provider.com/oauth/token"
REDIRECT_URI = "https://yourapp.com/callback"

# In production, use a session store (Redis, database)
state_store: dict[str, str] = {}


@app.get("/login")
async def login():
    client = AsyncOAuth2Client(
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET,
        redirect_uri=REDIRECT_URI,
    )
    state = secrets.token_urlsafe(32)
    uri, _ = client.create_authorization_url(
        AUTHORIZE_URL, state=state
    )
    state_store[state] = "pending"
    return RedirectResponse(uri)


@app.get("/callback")
async def callback(request: Request):
    state = request.query_params.get("state", "")
    if state not in state_store:
        return {"error": "Invalid state parameter"}

    del state_store[state]

    client = AsyncOAuth2Client(
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET,
        redirect_uri=REDIRECT_URI,
    )
    token = await client.fetch_token(
        TOKEN_URL,
        authorization_response=str(request.url),
    )
    # Store token securely (encrypted in database, not session cookie)
    return {"access_token": token["access_token"][:8] + "..."}

The state parameter prevents CSRF attacks. Generate it randomly, store it server-side, and verify it in the callback. Never skip this step.

PKCE implementation

PKCE adds a code challenge to the authorization request. Authlib handles this automatically:

from authlib.integrations.httpx_client import AsyncOAuth2Client
import hashlib
import base64
import secrets


def generate_pkce_pair() -> tuple[str, str]:
    verifier = secrets.token_urlsafe(64)
    challenge = base64.urlsafe_b64encode(
        hashlib.sha256(verifier.encode()).digest()
    ).rstrip(b"=").decode()
    return verifier, challenge


async def pkce_authorize():
    verifier, challenge = generate_pkce_pair()

    client = AsyncOAuth2Client(
        client_id=CLIENT_ID,
        redirect_uri=REDIRECT_URI,
        code_challenge_method="S256",
    )
    uri, state = client.create_authorization_url(
        AUTHORIZE_URL,
        code_challenge=challenge,
        code_challenge_method="S256",
    )
    # Store verifier associated with state
    return uri, state, verifier


async def pkce_exchange(code: str, verifier: str) -> dict:
    client = AsyncOAuth2Client(
        client_id=CLIENT_ID,
        redirect_uri=REDIRECT_URI,
    )
    token = await client.fetch_token(
        TOKEN_URL,
        code=code,
        code_verifier=verifier,
    )
    return token

The verifier must be stored securely between the authorize and callback steps. For web apps, use an encrypted server-side session. For CLI tools, hold it in memory during the flow.

Client Credentials flow (machine-to-machine)

The simplest flow — no user interaction:

from authlib.integrations.httpx_client import AsyncOAuth2Client


async def get_machine_token() -> dict:
    client = AsyncOAuth2Client(
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET,
    )
    token = await client.fetch_token(
        TOKEN_URL,
        grant_type="client_credentials",
        scope="read:analytics write:reports",
    )
    return token

For long-running services, cache the token and refresh before expiry:

import time
from dataclasses import dataclass


@dataclass
class TokenCache:
    access_token: str = ""
    expires_at: float = 0.0

    @property
    def is_expired(self) -> bool:
        return time.time() >= self.expires_at - 60  # 60s buffer


class ServiceAuth:
    def __init__(self, client_id: str, client_secret: str, token_url: str):
        self._client_id = client_id
        self._client_secret = client_secret
        self._token_url = token_url
        self._cache = TokenCache()

    async def get_token(self) -> str:
        if not self._cache.is_expired:
            return self._cache.access_token

        client = AsyncOAuth2Client(
            client_id=self._client_id,
            client_secret=self._client_secret,
        )
        token = await client.fetch_token(
            self._token_url,
            grant_type="client_credentials",
        )
        self._cache.access_token = token["access_token"]
        self._cache.expires_at = time.time() + token.get("expires_in", 3600)
        return self._cache.access_token

Transparent token refresh with httpx auth

Build an httpx.Auth subclass that handles refresh automatically:

import httpx
import time
from dataclasses import dataclass
from typing import Generator


@dataclass
class OAuthToken:
    access_token: str
    refresh_token: str
    expires_at: float
    token_url: str
    client_id: str
    client_secret: str


class OAuthAuth(httpx.Auth):
    def __init__(self, token: OAuthToken):
        self._token = token

    def auth_flow(
        self, request: httpx.Request
    ) -> Generator[httpx.Request, httpx.Response, None]:
        if time.time() >= self._token.expires_at - 30:
            self._refresh()

        request.headers["Authorization"] = (
            f"Bearer {self._token.access_token}"
        )
        response = yield request

        # Handle 401 by refreshing and retrying once
        if response.status_code == 401:
            self._refresh()
            request.headers["Authorization"] = (
                f"Bearer {self._token.access_token}"
            )
            yield request

    def _refresh(self) -> None:
        resp = httpx.post(
            self._token.token_url,
            data={
                "grant_type": "refresh_token",
                "refresh_token": self._token.refresh_token,
                "client_id": self._token.client_id,
                "client_secret": self._token.client_secret,
            },
        )
        resp.raise_for_status()
        data = resp.json()
        self._token.access_token = data["access_token"]
        self._token.expires_at = time.time() + data.get("expires_in", 3600)
        if "refresh_token" in data:
            self._token.refresh_token = data["refresh_token"]

The auth_flow generator is called for every request. It checks expiry proactively (30-second buffer) and also handles unexpected 401s by refreshing and retrying. The retry happens within httpx’s auth protocol — the caller sees a successful response.

Secure token storage

Tokens must be stored securely. Options by deployment context:

# For web apps: encrypted database column
from cryptography.fernet import Fernet

ENCRYPTION_KEY = Fernet.generate_key()  # Store in env var
cipher = Fernet(ENCRYPTION_KEY)


def encrypt_token(token: str) -> bytes:
    return cipher.encrypt(token.encode())


def decrypt_token(encrypted: bytes) -> str:
    return cipher.decrypt(encrypted).decode()

For CLI tools, use the system keyring:

import keyring


def store_token(service: str, token: str) -> None:
    keyring.set_password(service, "oauth_token", token)


def load_token(service: str) -> str | None:
    return keyring.get_password(service, "oauth_token")

Never store tokens in plain text files, environment variables visible to all processes, or version control.

Device Code flow for CLI tools

import httpx
import time


async def device_code_flow(
    client_id: str,
    device_auth_url: str,
    token_url: str,
) -> dict:
    # Step 1: Request device code
    resp = httpx.post(
        device_auth_url,
        data={"client_id": client_id, "scope": "user:read"},
    )
    resp.raise_for_status()
    data = resp.json()

    user_code = data["user_code"]
    verification_uri = data["verification_uri"]
    device_code = data["device_code"]
    interval = data.get("interval", 5)

    print(f"Open {verification_uri} and enter code: {user_code}")

    # Step 2: Poll for token
    while True:
        time.sleep(interval)
        resp = httpx.post(
            token_url,
            data={
                "client_id": client_id,
                "device_code": device_code,
                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
            },
        )

        if resp.status_code == 200:
            return resp.json()

        error = resp.json().get("error")
        if error == "authorization_pending":
            continue
        elif error == "slow_down":
            interval += 5
        elif error == "expired_token":
            raise TimeoutError("Device code expired")
        else:
            raise RuntimeError(f"OAuth error: {error}")

Security checklist for production OAuth

Requirement	Implementation
State parameter	Random token, server-side validation
PKCE	S256 challenge for all public clients
Token storage	Encrypted at rest, never in logs
Scope minimization	Request only needed permissions
Token rotation	Accept new refresh tokens when provided
HTTPS only	Reject redirect URIs without TLS
Redirect URI validation	Exact match, no wildcards
Token expiry handling	Proactive refresh with buffer

Common OAuth implementation mistakes

1. Storing tokens in localStorage — vulnerable to XSS. Use HttpOnly cookies or server-side sessions.
2. Not validating the state parameter — enables CSRF attacks where an attacker’s token gets bound to the victim’s session.
3. Ignoring token rotation — some providers issue new refresh tokens with each refresh. If you don’t store the new one, the old one becomes invalid.
4. Logging tokens — access tokens in logs are a security incident waiting to happen. Mask them in all logging output.
5. Hardcoding redirect URIs — makes local development and staging environments painful. Use configuration.

The one thing to remember: A production OAuth implementation in Python combines the right flow for your client type, transparent token refresh via httpx auth, encrypted token storage, and strict CSRF/PKCE protections — security that your API callers never have to think about.