OAuth Utilities

Generic OAuth flow handlers and token management

Overview

The OAuth utilities module provides generic, reusable functions for handling OAuth 2.0 authentication flows, token refresh, and PKCE (Proof Key for Code Exchange) operations.

Source: src/utils/oauth/util.py

Core Functions

run_oauth_flow

Generic OAuth flow handler that can be used by different services.

Source: src/utils/oauth/util.py:84

def run_oauth_flow(
    service_name: str,
    user_id: str,
    scopes: List[str],
    auth_url_base: str,
    token_url: str,
    auth_params_builder: Callable[[Dict[str, Any], str, List[str]], Dict[str, str]],
    token_data_builder: Callable[[Dict[str, Any], str, List[str], str], Dict[str, str]],
    process_token_response: Callable[[Dict[str, Any]], Dict[str, Any]] = None,
    token_header_builder: Callable[[Dict[str, Any]], Dict[str, str]] = None,
    port: int = 8080,
) -> Dict[str, Any]:

service_name · str · required

Name of the service (e.g., “microsoft”, “slack”)

user_id · str · required

ID of the user authenticating

scopes · List[str] · required

List of OAuth scopes to request

auth_url_base · str · required

Base authorization URL for the service

token_url · str · required

Token exchange URL for the service

auth_params_builder · Callable · required

Function to build auth URL parameters. Signature: (oauth_config, redirect_uri, scopes) -> Dict[str, str]

token_data_builder · Callable · required

Function to build token request data. Signature: (oauth_config, redirect_uri, scopes, auth_code) -> Dict[str, str]

process_token_response · Callable

Optional function to process token response. Signature: (token_response) -> Dict[str, Any]

token_header_builder · Callable

Optional function to build token request headers. Signature: (oauth_config) -> Dict[str, str]

port · int · default: "8080"

Port to use for local callback server

return · Dict[str, Any]

Dictionary containing the OAuth credentials including access_token, refresh_token, and expires_at

Behavior:

1. Gets OAuth config from auth client
2. Starts local HTTP server for callback
3. Opens browser for user authorization
4. Waits for callback (120 second timeout)
5. Exchanges authorization code for tokens
6. Saves credentials using auth client

refresh_token_if_needed

Checks if token needs refresh and handles the refresh process.

Source: src/utils/oauth/util.py:210

async def refresh_token_if_needed(
    user_id: str,
    service_name: str,
    token_url: str,
    token_data_builder: Callable[[Dict[str, Any], str, Dict[str, Any]], Dict[str, str]],
    process_token_response: Callable[[Dict[str, Any]], Dict[str, Any]] = None,
    token_header_builder: Callable[[Dict[str, Any]], Dict[str, str]] = None,
    api_key: Optional[str] = None,
    return_full_credentials: Optional[bool] = False,
) -> str:

user_id · str · required

ID of the user

service_name · str · required

Name of the service

token_url · str · required

URL for token refresh

token_data_builder · Callable · required

Function to build token refresh request data. Signature: (oauth_config, refresh_token, credentials_data) -> Dict[str, str]

process_token_response · Callable

Optional function to process token response

token_header_builder · Callable

Optional function to build token request headers

api_key · Optional[str]

Optional API key for auth client

return_full_credentials · bool · default: "False"

If True, returns full credentials dict; if False, returns only access_token

return · str | Dict[str, Any]

Access token string if return_full_credentials=False, otherwise full credentials dictionary

Behavior:

1. Gets credentials from auth client
2. Checks if token has expired or will expire in 5 minutes
3. Refreshes token if needed (only in local environment)
4. Saves updated credentials
5. Returns access token or full credentials

generate_code_verifier

Generates a code verifier for PKCE.

Source: src/utils/oauth/util.py:319

def generate_code_verifier() -> str:

return · str

Cryptographically secure random string of 64 characters using a-z, A-Z, 0-9, ”.”, ”-”, and ”_“

generate_code_challenge

Generates a code challenge from the code verifier.

Source: src/utils/oauth/util.py:331

def generate_code_challenge(code_verifier: str) -> str:

code_verifier · str · required

The code verifier string

return · str

Base64 URL-encoded SHA-256 hash of the code verifier with padding removed

OAuthCallbackHandler

HTTP request handler for OAuth callback.

Source: src/utils/oauth/util.py:22

class OAuthCallbackHandler(BaseHTTPRequestHandler):
    """HTTP request handler for OAuth callback."""

do_GET

Handles GET requests with OAuth callback.

Source: src/utils/oauth/util.py:25

def do_GET(self):

Behavior:

• Parses query parameters from callback URL
• Extracts authorization code or error
• Parses state parameter for code_verifier (PKCE)
• Stores additional parameters from callback
• Returns HTML page with success/error message

Example Usage

Basic OAuth Flow

from src.utils.oauth.util import run_oauth_flow

def build_auth_params(oauth_config, redirect_uri, scopes):
    return {
        "client_id": oauth_config["client_id"],
        "redirect_uri": redirect_uri,
        "scope": " ".join(scopes),
        "response_type": "code",
    }

def build_token_data(oauth_config, redirect_uri, scopes, auth_code):
    return {
        "client_id": oauth_config["client_id"],
        "client_secret": oauth_config["client_secret"],
        "code": auth_code,
        "redirect_uri": redirect_uri,
        "grant_type": "authorization_code",
    }

credentials = run_oauth_flow(
    service_name="myservice",
    user_id="user123",
    scopes=["read", "write"],
    auth_url_base="https://auth.example.com/oauth/authorize",
    token_url="https://auth.example.com/oauth/token",
    auth_params_builder=build_auth_params,
    token_data_builder=build_token_data,
)

Token Refresh

from src.utils.oauth.util import refresh_token_if_needed

def build_refresh_data(oauth_config, refresh_token, credentials_data):
    return {
        "client_id": oauth_config["client_id"],
        "client_secret": oauth_config["client_secret"],
        "refresh_token": refresh_token,
        "grant_type": "refresh_token",
    }

access_token = await refresh_token_if_needed(
    user_id="user123",
    service_name="myservice",
    token_url="https://auth.example.com/oauth/token",
    token_data_builder=build_refresh_data,
)

PKCE Flow

from src.utils.oauth.util import (
    generate_code_verifier,
    generate_code_challenge,
    run_oauth_flow
)
import json

def build_auth_params_with_pkce(oauth_config, redirect_uri, scopes):
    code_verifier = generate_code_verifier()
    code_challenge = generate_code_challenge(code_verifier)
    
    # Store code_verifier in state
    state = json.dumps({"code_verifier": code_verifier})
    
    return {
        "client_id": oauth_config["client_id"],
        "redirect_uri": redirect_uri,
        "scope": " ".join(scopes),
        "response_type": "code",
        "code_challenge": code_challenge,
        "code_challenge_method": "S256",
        "state": state,
    }

def build_token_data_with_pkce(oauth_config, redirect_uri, scopes, auth_code):
    return {
        "client_id": oauth_config["client_id"],
        "client_secret": oauth_config["client_secret"],
        "code": auth_code,
        "redirect_uri": redirect_uri,
        "grant_type": "authorization_code",
        "code_verifier": oauth_config.get("code_verifier"),
    }

credentials = run_oauth_flow(
    service_name="myservice",
    user_id="user123",
    scopes=["read", "write"],
    auth_url_base="https://auth.example.com/oauth/authorize",
    token_url="https://auth.example.com/oauth/token",
    auth_params_builder=build_auth_params_with_pkce,
    token_data_builder=build_token_data_with_pkce,
)

Error Handling

from src.utils.oauth.util import run_oauth_flow

try:
    credentials = run_oauth_flow(
        # ... parameters
    )
except ValueError as e:
    if "timed out" in str(e):
        print("User didn't complete authentication in time")
    elif "failed" in str(e):
        print(f"Authentication error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

See Also

• Service Helper Utilities - Service-specific OAuth implementations
• BaseAuthClient - Authentication client interface
• create_auth_client() - Auth client factory