OAuth 2.0: Complete Developer Guide with Examples

# oauth_server.py
# Simple OAuth 2.0 authorization server (for learning)

from flask import Flask, request, redirect, jsonify
import secrets
import json
from datetime import datetime, timedelta

app = Flask(__name__)

# In-memory storage (use DB in production)
clients = {
    'myapp': {
        'client_secret': 'secret123',
        'redirect_uris': ['http://localhost:5001/callback'],
        'name': 'My App',
    }
}
authorization_codes = {}
access_tokens = {}
refresh_tokens = {}

def generate_token():
    """Generate a secure random token."""
    return secrets.token_urlsafe(32)

@app.route('/oauth/authorize', methods=['GET'])
def authorize():
    """Step 1: User authorizes the application."""
    client_id = request.args.get('client_id')
    redirect_uri = request.args.get('redirect_uri')
    response_type = request.args.get('response_type')
    state = request.args.get('state')
    scope = request.args.get('scope', 'read')

    # Validate client
    client = clients.get(client_id)
    if not client or redirect_uri not in client['redirect_uris']:
        return jsonify({'error': 'invalid_client'}), 400

    if response_type != 'code':
        return jsonify({'error': 'unsupported_response_type'}), 400

    # In real app: show user a consent screen
    # Here we auto-approve for simplicity
    code = generate_token()
    authorization_codes[code] = {
        'client_id': client_id,
        'redirect_uri': redirect_uri,
        'scope': scope,
        'user_id': 'user_42',
        'expires_at': datetime.now() + timedelta(minutes=10),
    }

    # Redirect back to the app with the authorization code
    redirect_url = f"{redirect_uri}?code={code}&state={state}"
    print(f"Redirecting to: {redirect_url}")
    return redirect(redirect_url)

@app.route('/oauth/token', methods=['POST'])
def token():
    """Step 2: Exchange authorization code for tokens."""
    client_id = request.form.get('client_id')
    client_secret = request.form.get('client_secret')
    grant_type = request.form.get('grant_type')
    code = request.form.get('code')
    redirect_uri = request.form.get('redirect_uri')

    # Validate client credentials
    client = clients.get(client_id)
    if not client or client['client_secret'] != client_secret:
        return jsonify({'error': 'invalid_client'}), 401

    if grant_type != 'authorization_code':
        return jsonify({'error': 'unsupported_grant_type'}), 400

    # Validate authorization code
    auth_code = authorization_codes.get(code)
    if not auth_code:
        return jsonify({'error': 'invalid_grant'}), 400

    if auth_code['client_id'] != client_id:
        return jsonify({'error': 'invalid_grant'}), 400

    if datetime.now() > auth_code['expires_at']:
        return jsonify({'error': 'code_expired'}), 400

    # Generate tokens
    access_token = generate_token()
    refresh_token = generate_token()

    access_tokens[access_token] = {
        'user_id': auth_code['user_id'],
        'scope': auth_code['scope'],
        'client_id': client_id,
        'expires_at': datetime.now() + timedelta(hours=1),
    }
    refresh_tokens[refresh_token] = {
        'user_id': auth_code['user_id'],
        'client_id': client_id,
    }

    # Clean up used authorization code
    del authorization_codes[code]

    print(f"Issued access token: {access_token[:20]}...")
    return jsonify({
        'access_token': access_token,
        'token_type': 'Bearer',
        'expires_in': 3600,
        'refresh_token': refresh_token,
        'scope': auth_code['scope'],
    })

@app.route('/oauth/refresh', methods=['POST'])
def refresh():
    """Step 3: Refresh an expired access token."""
    client_id = request.form.get('client_id')
    client_secret = request.form.get('client_secret')
    refresh_token = request.form.get('refresh_token')
    grant_type = request.form.get('grant_type')

    if grant_type != 'refresh_token':
        return jsonify({'error': 'unsupported_grant_type'}), 400

    # Validate refresh token
    stored = refresh_tokens.get(refresh_token)
    if not stored:
        return jsonify({'error': 'invalid_grant'}), 400

    # Issue new access token
    new_access = generate_token()
    access_tokens[new_access] = {
        'user_id': stored['user_id'],
        'scope': 'read',
        'client_id': client_id,
        'expires_at': datetime.now() + timedelta(hours=1),
    }

    print(f"Refreshed token for user {stored['user_id']}")
    return jsonify({
        'access_token': new_access,
        'token_type': 'Bearer',
        'expires_in': 3600,
    })

@app.route('/oauth/resource', methods=['GET'])
def resource():
    """Protected resource — requires valid access token."""
    auth_header = request.headers.get('Authorization', '')
    if not auth_header.startswith('Bearer '):
        return jsonify({'error': 'missing_token'}), 401

    token = auth_header[7:]
    token_data = access_tokens.get(token)

    if not token_data:
        return jsonify({'error': 'invalid_token'}), 401

    if datetime.now() > token_data['expires_at']:
        return jsonify({'error': 'token_expired'}), 401

    return jsonify({
        'user_id': token_data['user_id'],
        'scope': token_data['scope'],
        'message': 'Access granted to protected resource',
    })

if __name__ == '__main__':
    print("OAuth 2.0 Authorization Server running on :5000")
    app.run(port=5000, debug=True)

# Step 1: Get authorization code
curl "http://localhost:5000/oauth/authorize?client_id=myapp&redirect_uri=http://localhost:5001/callback&response_type=code&state=xyz"

# Step 2: Exchange code for tokens
curl -X POST http://localhost:5000/oauth/token \
  -d "client_id=myapp" \
  -d "client_secret=secret123" \
  -d "grant_type=authorization_code" \
  -d "code=AUTHORIZATION_CODE_FROM_STEP_1" \
  -d "redirect_uri=http://localhost:5001/callback"

# Step 3: Access protected resource
curl http://localhost:5000/oauth/resource \
  -H "Authorization: Bearer ACCESS_TOKEN_FROM_STEP_2"

{
  "access_token": "abc123def456...",
  "token_type": "Bearer",
  "expires_in": 3600,
  "refresh_token": "ghi789jkl012...",
  "scope": "read"
}

# google_oauth.py
# Sign in with Google using OAuth 2.0

import requests
from flask import Flask, request, redirect, session
from google.oauth2 import id_token
from google.auth.transport import requests as google_requests
import os
import secrets

app = Flask(__name__)
app.secret_key = os.environ.get('FLASK_SECRET_KEY', secrets.token_hex(32))

GOOGLE_CLIENT_ID = os.environ['GOOGLE_CLIENT_ID']
GOOGLE_CLIENT_SECRET = os.environ['GOOGLE_CLIENT_SECRET']
GOOGLE_REDIRECT_URI = 'http://localhost:5000/login/google/callback'

@app.route('/login/google')
def login_google():
    """Step 1: Redirect user to Google's OAuth consent screen."""
    params = {
        'client_id': GOOGLE_CLIENT_ID,
        'redirect_uri': GOOGLE_REDIRECT_URI,
        'response_type': 'code',
        'scope': 'openid email profile',
        'access_type': 'offline',  # Get refresh token
        'state': secrets.token_urlsafe(16),
    }
    auth_url = 'https://accounts.google.com/o/oauth2/v2/auth'
    url = f"{auth_url}?{'&'.join(f'{k}={v}' for k, v in params.items())}"
    print(f"Redirecting to Google: {url[:60]}...")
    return redirect(url)

@app.route('/login/google/callback')
def google_callback():
    """Step 2: Handle Google's callback with authorization code."""
    code = request.args.get('code')
    state = request.args.get('state')

    # Exchange authorization code for tokens
    token_url = 'https://oauth2.googleapis.com/token'
    token_data = {
        'code': code,
        'client_id': GOOGLE_CLIENT_ID,
        'client_secret': GOOGLE_CLIENT_SECRET,
        'redirect_uri': GOOGLE_REDIRECT_URI,
        'grant_type': 'authorization_code',
    }

    response = requests.post(token_url, data=token_data)
    tokens = response.json()

    print(f"Token response: {tokens.get('token_type', 'error')}")

    # Verify the ID token
    try:
        info = id_token.verify_oauth2_token(
            tokens['id_token'],
            google_requests.Request(),
            GOOGLE_CLIENT_ID,
        )
        user_id = info['sub']
        email = info['email']
        name = info.get('name', 'Unknown')

        print(f"Authenticated user: {email} ({name})")
        print(f"User ID: {user_id}")

        # Store user info in session
        session['user'] = {
            'id': user_id,
            'email': email,
            'name': name,
            'picture': info.get('picture', ''),
        }

        return f"Welcome, {name}! You're signed in with {email}"

    except ValueError as e:
        print(f"Token verification failed: {e}")
        return "Authentication failed", 401

if __name__ == '__main__':
    app.run(port=5000, debug=True)

# github_oauth.py
# Sign in with GitHub using OAuth 2.0

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

app = Flask(__name__)
app.secret_key = os.environ.get('FLASK_SECRET_KEY', secrets.token_hex(32))

GITHUB_CLIENT_ID = os.environ['GITHUB_CLIENT_ID']
GITHUB_CLIENT_SECRET = os.environ['GITHUB_CLIENT_SECRET']

@app.route('/login/github')
def login_github():
    """Redirect to GitHub OAuth authorization."""
    params = {
        'client_id': GITHUB_CLIENT_ID,
        'redirect_uri': 'http://localhost:5000/login/github/callback',
        'scope': 'read:user user:email',
        'state': secrets.token_urlsafe(16),
    }
    url = f"https://github.com/login/oauth/authorize?{'&'.join(f'{k}={v}' for k, v in params.items())}"
    print(f"Redirecting to GitHub OAuth")
    return redirect(url)

@app.route('/login/github/callback')
def github_callback():
    """Handle GitHub OAuth callback."""
    code = request.args.get('code')

    # Exchange code for access token
    token_response = requests.post(
        'https://github.com/login/oauth/access_token',
        headers={'Accept': 'application/json'},
        data={
            'client_id': GITHUB_CLIENT_ID,
            'client_secret': GITHUB_CLIENT_SECRET,
            'code': code,
        }
    )
    access_token = token_response.json().get('access_token')
    print(f"GitHub access token obtained: {access_token[:10]}...")

    # Fetch user data
    user_response = requests.get(
        'https://api.github.com/user',
        headers={
            'Authorization': f'token {access_token}',
            'Accept': 'application/json',
        }
    )
    user_data = user_response.json()

    # Fetch email (may be separate endpoint)
    email_response = requests.get(
        'https://api.github.com/user/emails',
        headers={'Authorization': f'token {access_token}'}
    )
    primary_email = next(
        (e['email'] for e in email_response.json() if e['primary']),
        'no-email@github.com'
    )

    print(f"GitHub user: {user_data.get('login')} ({primary_email})")
    print(f"Repos: {user_data.get('public_repos')}")
    print(f"Followers: {user_data.get('followers')}")

    session['user'] = {
        'id': user_data['id'],
        'login': user_data['login'],
        'name': user_data.get('name', user_data['login']),
        'email': primary_email,
        'avatar': user_data.get('avatar_url'),
    }

    return f"Signed in as GitHub user: {user_data.get('login')}"

# pkce_flow.py
# OAuth 2.0 with PKCE for mobile/web apps that can't store secrets

import hashlib
import base64
import secrets
import requests

def generate_pkce_pair():
    """Generate code_verifier and code_challenge for PKCE."""
    # Step 1: Generate random code_verifier
    code_verifier = secrets.token_urlsafe(64)[:128]

    # Step 2: SHA256 hash and base64url encode to get code_challenge
    code_challenge = hashlib.sha256(
        code_verifier.encode('ascii')
    ).digest()
    code_challenge = base64.urlsafe_b64encode(code_challenge).rstrip(b'=').decode('ascii')

    print(f"Code Verifier: {code_verifier[:20]}...")
    print(f"Code Challenge: {code_challenge[:20]}...")

    return code_verifier, code_challenge

def pkce_authorization_flow():
    """Complete PKCE authorization flow."""
    verifier, challenge = generate_pkce_pair()

    # Step 1: Redirect user to authorization URL with code_challenge
    auth_params = {
        'response_type': 'code',
        'client_id': 'my_mobile_app',
        'redirect_uri': 'myapp://callback',
        'code_challenge': challenge,
        'code_challenge_method': 'S256',
        'state': secrets.token_urlsafe(16),
        'scope': 'openid profile',
    }

    auth_url = 'https://authserver.com/oauth/authorize'
    full_url = f"{auth_url}?{'&'.join(f'{k}={v}' for k, v in auth_params.items())}"
    print(f"Redirect user to: {full_url}")
    print("(User authenticates and authorizes)")

    # Step 2: Exchange code for tokens (includes code_verifier)
    code = input("Enter authorization code: ")

    token_params = {
        'grant_type': 'authorization_code',
        'code': code,
        'redirect_uri': 'myapp://callback',
        'client_id': 'my_mobile_app',
        'code_verifier': verifier,  # Proves we created the challenge
    }

    response = requests.post(
        'https://authserver.com/oauth/token',
        data=token_params
    )
    tokens = response.json()
    print(f"Access token obtained: {tokens.get('access_token', 'error')[:20]}...")
    print(f"Refresh token: {tokens.get('refresh_token', 'none')[:20]}...")

pkce_authorization_flow()

# scopes.py
# Working with OAuth scopes for granular permissions

# Common scope patterns
SCOPES = {
    'google': {
        'openid': 'Authenticate user identity',
        'email': 'View email address',
        'profile': 'View basic profile info',
        'https://www.googleapis.com/auth/drive.file':
            'Create and access files you create with this app',
        'https://www.googleapis.com/auth/calendar':
            'View and manage calendar events',
        'https://mail.google.com/': 'Read, compose, send emails',
    },
    'github': {
        'repo': 'Full control of private repositories',
        'repo:status': 'Access commit status',
        'user': 'Read-only access to profile',
        'user:email': 'Read email addresses',
        'admin:repo_hook': 'Manage repository webhooks',
    }
}

def request_minimal_scopes():
    """Request only the scopes you need — principle of least privilege."""
    print("Recommended scopes by use case:")
    print()
    print("Login only:")
    print("  Google: openid email profile")
    print("  GitHub: read:user user:email")
    print()
    print("Cloud storage access:")
    print("  Google: openid email https://www.googleapis.com/auth/drive.file")
    print()
    print("Calendar integration:")
    print("  Google: openid email https://www.googleapis.com/auth/calendar.readonly")
    print()

    # Validate scopes on the resource server
    def check_scope(required_scope, token_scopes):
        """Check if a token has the required scope."""
        token_scope_list = token_scopes.split()
        if required_scope not in token_scope_list:
            raise PermissionError(
                f"Token missing required scope: {required_scope}"
            )
        print(f"✅ Token has required scope: {required_scope}")

    # Example
    token_scopes = "openid email profile"
    check_scope('email', token_scopes)  # OK
    # check_scope('https://www.googleapis.com/auth/drive.file', token_scopes)  # Would fail

request_minimal_scopes()

# token_inspector.py
# Decode and inspect JWT access tokens

import jwt
import json
from datetime import datetime

def inspect_token(token, public_key=None):
    """Decode and display JWT token contents."""
    print("═" * 50)
    print("Token Inspector")
    print("═" * 50)

    # Decode header
    header = jwt.get_unverified_header(token)
    print(f"\nHeader:")
    print(f"  Algorithm: {header.get('alg', 'unknown')}")
    print(f"  Type: {header.get('typ', 'JWT')}")
    print(f"  Key ID: {header.get('kid', 'none')}")

    # Decode payload (without verification)
    payload = jwt.decode(token, options={"verify_signature": False})
    print(f"\nPayload:")

    now = datetime.now().timestamp()

    for key, value in payload.items():
        if key in ('exp', 'iat', 'nbf', 'auth_time'):
            dt = datetime.fromtimestamp(value)
            is_expired = value < now if key == 'exp' else False
            status = "⚠️ EXPIRED" if is_expired else "✅ Valid"
            print(f"  {key}: {dt} ({status})")
        elif key == 'scope':
            print(f"  {key}: {value}")
        else:
            print(f"  {key}: {value}")

    # Verify if public key is provided
    if public_key:
        try:
            verified = jwt.decode(token, public_key, algorithms=['RS256'])
            print(f"\n✅ Signature: VALID (verified with public key)")
        except jwt.InvalidSignatureError:
            print(f"\n❌ Signature: INVALID")

    print("═" * 50)

# Test with a sample token
# inspect_token("eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...")