Comparing OAuth 2.0 Flows — Authorization Code vs Client Credentials

from flask import Flask, redirect, request, session, url_for
import requests
import secrets

app = Flask(__name__)
app.secret_key = 'your-secret-key'

# OAuth configuration
AUTH_URL = 'https://auth.example.com/authorize'
TOKEN_URL = 'https://auth.example.com/token'
CLIENT_ID = 'your_client_id'
CLIENT_SECRET = 'your_client_secret'
REDIRECT_URI = 'http://localhost:5000/callback'

@app.route('/login')
def login():
    # Generate random state for CSRF protection
    state = secrets.token_urlsafe(32)
    session['oauth_state'] = state
    
    # Build authorization URL
    params = {
        'response_type': 'code',
        'client_id': CLIENT_ID,
        'redirect_uri': REDIRECT_URI,
        'scope': 'read write',
        'state': state
    }
    
    auth_url = f"{AUTH_URL}?{'&'.join(f'{k}={v}' for k, v in params.items())}"
    return redirect(auth_url)

@app.route('/callback')
def callback():
    # Verify state to prevent CSRF
    if request.args.get('state') != session.get('oauth_state'):
        return 'Invalid state parameter', 400
    
    # Exchange authorization code for tokens
    code = request.args.get('code')
    token_data = {
        'grant_type': 'authorization_code',
        'code': code,
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'redirect_uri': REDIRECT_URI
    }
    
    response = requests.post(TOKEN_URL, data=token_data)
    tokens = response.json()
    
    # Store tokens in session
    session['access_token'] = tokens['access_token']
    session['refresh_token'] = tokens['refresh_token']
    
    return redirect(url_for('dashboard'))

@app.route('/dashboard')
def dashboard():
    if 'access_token' not in session:
        return redirect(url_for('login'))
    
    # Use access token to call API
    headers = {'Authorization': f"Bearer {session['access_token']}"}
    response = requests.get('https://api.example.com/user', headers=headers)
    
    return f"User data: {response.json()}"

@app.route('/refresh')
def refresh():
    # Refresh access token using refresh token
    token_data = {
        'grant_type': 'refresh_token',
        'refresh_token': session['refresh_token'],
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET
    }
    
    response = requests.post(TOKEN_URL, data=token_data)
    tokens = response.json()
    
    session['access_token'] = tokens['access_token']
    if 'refresh_token' in tokens:
        session['refresh_token'] = tokens['refresh_token']
    
    return 'Token refreshed successfully'

import requests
import time
from typing import Optional

class OAuth2Client:
    """OAuth 2.0 Client Credentials client with automatic token management."""
    
    def __init__(self, token_url: str, client_id: str, client_secret: str):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token: Optional[str] = None
        self.token_expires_at: float = 0
    
    def get_access_token(self) -> str:
        """Get a valid access token, refreshing if necessary."""
        # Return cached token if still valid
        if self.access_token and time.time() < self.token_expires_at:
            return self.access_token
        
        # Request new token
        response = requests.post(
            self.token_url,
            data={
                'grant_type': 'client_credentials',
                'client_id': self.client_id,
                'client_secret': self.client_secret,
                'scope': 'read write'
            },
            headers={'Content-Type': 'application/x-www-form-urlencoded'}
        )
        response.raise_for_status()
        
        token_data = response.json()
        self.access_token = token_data['access_token']
        # Set expiry with 60 second buffer
        self.token_expires_at = time.time() + token_data['expires_in'] - 60
        
        return self.access_token
    
    def call_api(self, api_url: str, method: str = 'GET', **kwargs) -> dict:
        """Make an authenticated API call with automatic retry on token expiration."""
        token = self.get_access_token()
        headers = kwargs.pop('headers', {})
        headers['Authorization'] = f'Bearer {token}'
        
        response = requests.request(method, api_url, headers=headers, **kwargs)
        
        # Retry once if token expired
        if response.status_code == 401:
            self.access_token = None  # Force token refresh
            token = self.get_access_token()
            headers['Authorization'] = f'Bearer {token}'
            response = requests.request(method, api_url, headers=headers, **kwargs)
        
        response.raise_for_status()
        return response.json()

# Usage example
client = OAuth2Client(
    token_url='https://auth.example.com/oauth/token',
    client_id='your_client_id',
    client_secret='your_client_secret'
)

# Make API calls - token management is automatic
data = client.call_api('https://api.example.com/resources')
print(data)

# Create a resource
new_resource = client.call_api(
    'https://api.example.com/resources',
    method='POST',
    json={'name': 'New Resource'}
)
print(new_resource)