OAuth 2.0 and OpenID Connect: The Complete Guide

oauth_roles = {
    "resource_owner": {
        "description": "The user",
        "example": "You, granting access to your photos"
    },
    
    "client": {
        "description": "The application requesting access",
        "example": "My Photo Printing App"
    },
    
    "authorization_server": {
        "description": "The server that authenticates and issues tokens",
        "example": "Google, Auth0, Okta"
    },
    
    "resource_server": {
        "description": "The API that hosts protected resources",
        "example": "Google Photos API"
    }
}

# 1. Authorization Code Flow (for web apps)
# Most secure, uses server-side token exchange

flow_authorization_code:
  steps:
    - "User clicks 'Login with Google'"
    - "Redirect to Google OAuth"
    - "User authorizes"
    - "Redirect back with auth code"
    - "Server exchanges code for token"
    
  use_when:
    - "Server-side web apps"
    - "Mobile apps (with PKCE)"

# 2. Implicit Flow (deprecated)
# Don't use - security issues

# 3. Client Credentials Flow
# Server-to-server communication

flow_client_credentials:
  use_when:
    - "Machine-to-machine"
    - "Background jobs"
    
  example:
    "Service A accessing Service B"

# 4. Device Code Flow
# For devices without browsers

# 5. Refresh Token Flow
# Get new access tokens

# Authorization Code Flow

class AuthorizationCodeFlow:
    """Web app OAuth flow"""
    
    def __init__(self, client_id, client_secret, redirect_uri):
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri
    
    def get_authorization_url(self, state, scope):
        """Step 1: Get URL to redirect user"""
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "scope": scope,
            "state": state  # CSRF protection
        }
        
        # Example: https://auth.example.com/authorize?...
        return f"https://auth.example.com/authorize?" \
               f"{urllib.parse.urlencode(params)}"
    
    def exchange_code_for_token(self, code):
        """Step 2: Exchange code for token"""
        response = requests.post(
            "https://auth.example.com/token",
            data={
                "grant_type": "authorization_code",
                "code": code,
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "redirect_uri": self.redirect_uri
            }
        )
        
        return response.json()
    
    def refresh_token(self, refresh_token):
        """Step 3: Get new access token"""
        response = requests.post(
            "https://auth.example.com/token",
            data={
                "grant_type": "refresh_token",
                "refresh_token": refresh_token,
                "client_id": self.client_id,
                "client_secret": self.client_secret
            }
        )
        
        return response.json()

# Flask implementation
from flask import Flask, request, session, redirect
import requests

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

# Configuration
AUTH_SERVER = "https://accounts.google.com"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
REDIRECT_URI = "https://yourapp.com/callback"

@app.route("/login")
def login():
    state = secrets.token_hex(16)
    session["oauth_state"] = state
    
    auth_url = (
        f"{AUTH_SERVER}/o/oauth2/v2/auth?"
        f"client_id={CLIENT_ID}&"
        f"redirect_uri={REDIRECT_URI}&"
        f"response_type=code&"
        f"scope=openid%20email%20profile&"
        f"state={state}"
    )
    
    return redirect(auth_url)

@app.route("/callback")
def callback():
    # Verify state
    if request.args.get("state") != session.get("oauth_state"):
        return "Invalid state", 400
    
    # Exchange code for token
    token_response = requests.post(
        f"{AUTH_SERVER}/token",
        data={
            "code": request.args.get("code"),
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "redirect_uri": REDIRECT_URI,
            "grant_type": "authorization_code"
        }
    ).json()
    
    # Get user info
    user_info = requests.get(
        f"{AUTH_SERVER}/userinfo",
        headers={"Authorization": f"Bearer {token_response['access_token']}"}
    ).json()
    
    # Create session
    session["user"] = user_info
    session["access_token"] = token_response["access_token"]
    
    return redirect("/dashboard")

token_comparison = {
    "id_token": {
        "purpose": "Authentication (who is the user)",
        "audience": "Client application",
        "format": "JWT",
        "contains": "User claims (sub, name, email)",
        "verified_by": "Signature + claims"
    },
    
    "access_token": {
        "purpose": "Authorization (what can they access)",
        "audience": "Resource server (API)",
        "format": "Opaque or JWT",
        "contains": "Scopes, user, client info",
        "verified_by": "Token introspection"
    }
}

# Example ID Token (decoded)

id_token = {
    "iss": "https://accounts.google.com",  # Issuer
    "azp": "client-id.apps.googleusercontent.com",  # Authorized party
    "aud": "client-id.apps.googleusercontent.com",  # Audience
    "sub": "1234567890",  # Subject (user ID)
    "at_hash": "access_token_hash",  # Access token hash
    "iat": 1516239022,  # Issued at
    "exp": 1516242622,  # Expiration
    
    # Custom claims
    "name": "John Doe",
    "picture": "https://...",
    "email": "[email protected]",
    "email_verified": True,
    "locale": "en"
}

# OIDC Discovery endpoint

# GET https://accounts.google.com/.well-known/openid-configuration

discovery_document = {
    "issuer": "https://accounts.google.com",
    "authorization_endpoint": "https://accounts.google.com/o/oauth2/v2/auth",
    "token_endpoint": "https://oauth2.googleapis.com/token",
    "userinfo_endpoint": "https://openidconnect.googleapis.com/v1/userinfo",
    "jwks_uri": "https://www.googleapis.com/oauth2/v3/certs",
    "response_types_supported": ["code", "token", "id_token"],
    "subject_types_supported": ["public"],
    "id_token_signing_alg_values_supported": ["RS256"]
}

# Use discovery to configure client
import requests

def get_oidc_config(issuer):
    well_known = f"{issuer}/.well-known/openid-configuration"
    return requests.get(well_known).json()

config = get_oidc_config("https://accounts.google.com")

# Access Token - Short-lived
access_token = {
    "lifetime": "15 minutes to 1 hour",
    "purpose": "Access protected resources",
    "storage": "Server memory (not localStorage!)"
}

# Refresh Token - Long-lived
refresh_token = {
    "lifetime": "Days to weeks",
    "purpose": "Get new access tokens",
    "storage": "Secure server-side storage, encrypted"
}

# ID Token - For authentication
id_token = {
    "lifetime": "15 minutes to 1 hour",
    "purpose": "Verify user identity",
    "storage": "Session storage"
}

# SECURE storage

# Frontend (SPA)
secure_storage = {
    "access_token": {
        "where": "JavaScript memory only",
        "why": "XSS can't steal from memory"
    },
    
    "refresh_token": {
        "where": "HTTP-only cookie (server-side)",
        "why": "JavaScript can't read HTTP-only cookies"
    }
}

# Backend - Always encrypt refresh tokens
import cryptography

class TokenStore:
    def __init__(self, key):
        self.cipher = cryptography.fernet.Fernet(key)
    
    def store(self, user_id, refresh_token):
        encrypted = self.cipher.encrypt(refresh_token.encode())
        db.save(user_id, encrypted)
    
    def get(self, user_id):
        encrypted = db.get(user_id)
        return self.cipher.decrypt(encrypted).decode()

# Validate JWT (ID Token or Access Token)

import jwt

def validate_token(token, jwks_uri, issuer, audience):
    # Get signing keys
    jwks = requests.get(jwks_uri).json()
    
    # Find key for token
    unverified_header = jwt.get_unverified_header(token)
    signing_key = None
    
    for key in jwks["keys"]:
        if key["kid"] == unverified_header["kid"]:
            signing_key = jwt.algorithms.RSAAlgorithm.from_jwk(key)
            break
    
    if not signing_key:
        raise ValueError("No matching signing key")
    
    # Verify and decode
    payload = jwt.decode(
        token,
        signing_key,
        algorithms=["RS256"],
        audience=audience,
        issuer=issuer
    )
    
    return payload

# Security recommendations

security_basics:
  - "Use HTTPS everywhere"
  - "Validate redirect_uri exactly"
  - "Use state parameter for CSRF"
  - "Use code verifier for PKCE"

token_security:
  - "Short-lived access tokens"
  - "Long-lived, secure refresh tokens"
  - "Never expose tokens in URLs"
  - "Implement token revocation"

client_security:
  - "Never store client_secret in frontend"
  - "Use PKCE for public clients"
  - "Validate all responses"

scopes:
  - "Request minimum necessary scopes"
  - "Don't trust scope parameter in token exchange"
  - "Validate granted scopes"

# PKCE for additional security

class PKCE:
    """PKCE implementation"""
    
    def __init__(self):
        self.code_verifier = secrets.token_urlsafe(32)
        self.code_challenge = self._generate_challenge()
    
    def _generate_challenge(self):
        # SHA256 hash of verifier, base64url encoded
        digest = hashlib.sha256(self.code_verifier.encode()).digest()
        return base64.urlsafe_b64encode(digest).decode()[:-1]
    
    def get_authorization_params(self):
        return {
            "code_challenge": self.code_challenge,
            "code_challenge_method": "S256"
        }
    
    def get_token_params(self, code):
        return {
            "code_verifier": self.code_verifier
        }

# Authorization Code + PKCE

@app.route("/login")
def login():
    pkce = PKCE()
    session["pkce"] = pkce
    
    params = {
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": "openid profile email",
        "state": secrets.token_hex(16),
        **pkce.get_authorization_params()
    }
    
    return redirect(f"{AUTH_URL}?{urllib.parse.urlencode(params)}")

@app.route("/callback")
def callback():
    pkce = session["pkce"]
    
    # Exchange code with verifier
    token_response = requests.post(
        TOKEN_URL,
        data={
            "grant_type": "authorization_code",
            "code": request.args.get("code"),
            "redirect_uri": REDIRECT_URI,
            "client_id": CLIENT_ID,
            **pkce.get_token_params(request.args.get("code"))
        }
    ).json()
    
    return token_response