Implementing OAuth2 without Libraries

Core Engineering Group•Oct 18, 2024

20 min read

Disclaimer

Do not use this in production. This article is for educational purposes. In production, always use battle-tested libraries like passport.js, spring-security-oauth2, or authlib.

The Authorization Code Grant Flow

The Authorization Code Grant is the most secure OAuth2 flow. Here's how it works:

Client redirects user to Authorization Server
User authenticates and grants permission
Authorization Server redirects back with a code
Client exchanges the code for tokens
Client uses tokens to access resources

sequenceDiagram
    participant User
    participant Client
    participant Auth as Auth Server

    User->>Client: Click "Login"
    Client-->>User: 302 Redirect to Auth (with PKCE Challenge)
    User->>Auth: Enter Credentials
    Auth-->>User: 302 Redirect to Callback (with Code)
    User->>Client: GET /callback?code=...
    Client->>Auth: POST /token (Code + PKCE Verifier)
    Auth-->>Client: JSON { access_token, refresh_token }
    Note right of Client: Client is now authenticated

Step 1: Authorization Request

from urllib.parse import urlencode
import secrets
 
def generate_authorization_url(client_id: str, redirect_uri: str) -> str:
    state = secrets.token_urlsafe(32)
    code_verifier = secrets.token_urlsafe(64)
 
    # PKCE: Generate code challenge
    import hashlib, base64
    code_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).decode().rstrip('=')
 
    params = {
        'response_type': 'code',
        'client_id': client_id,
        'redirect_uri': redirect_uri,
        'scope': 'openid profile email',
        'state': state,
        'code_challenge': code_challenge,
        'code_challenge_method': 'S256',
    }
 
    return f"https://auth.example.com/authorize?{urlencode(params)}"

Step 2: Token Exchange

import httpx
 
async def exchange_code_for_tokens(
    code: str,
    client_id: str,
    client_secret: str,
    redirect_uri: str,
    code_verifier: str,
) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://auth.example.com/token",
            data={
                'grant_type': 'authorization_code',
                'code': code,
                'redirect_uri': redirect_uri,
                'client_id': client_id,
                'client_secret': client_secret,
                'code_verifier': code_verifier,
            },
        )
        return response.json()

Step 3: Refreshing Tokens

Access tokens are short-lived. To stay logged in, we exchange the refresh token for a new pair of tokens:

async def refresh_access_token(
    refresh_token: str,
    client_id: str,
    client_secret: str,
) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://auth.example.com/token",
            data={
                'grant_type': 'refresh_token',
                'refresh_token': refresh_token,
                'client_id': client_id,
                'client_secret': client_secret,
            },
        )
        return response.json()

Step 4: Protected Middleware

Now we need to protect our API routes. Here is a simple middleware example using FastAPI:

from fastapi import Request, HTTPException
 
async def verify_token(request: Request):
    auth_header = request.headers.get('Authorization')
    if not auth_header or not auth_header.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing token")
    
    token = auth_header.split(' ')[1]
    # In production, verify JWT signature here!
    user_info = await fetch_user_info(token) 
    request.state.user = user_info

Security Considerations

When implementing OAuth2, these are the critical security measures:

Always use PKCE: Prevents authorization code interception
Validate the state parameter: Prevents CSRF attacks
Use short-lived access tokens: 15 minutes maximum
Rotate refresh tokens: One-time use with rotation
Validate redirect URIs: Exact match, no wildcards

Pro Tip: Store refresh tokens encrypted at rest, and implement token binding to prevent token theft from being useful.

Understanding OAuth2 internals makes you a better engineer, even if you never implement it from scratch in production. The security patterns—PKCE, state validation, token rotation—apply broadly across authentication systems.

BackendBytes Dev Team

Core Engineering Group

Building robust, scalable applications with modern best practices.

Implementing OAuth2 without Libraries

BackendBytes Dev Team

Core Engineering Group•Oct 18, 2024

20 min read

Disclaimer

Do not use this in production. This article is for educational purposes. In production, always use battle-tested libraries like passport.js, spring-security-oauth2, or authlib.

The Authorization Code Grant Flow

The Authorization Code Grant is the most secure OAuth2 flow. Here's how it works:

Client redirects user to Authorization Server
User authenticates and grants permission
Authorization Server redirects back with a code
Client exchanges the code for tokens
Client uses tokens to access resources

sequenceDiagram
    participant User
    participant Client
    participant Auth as Auth Server

    User->>Client: Click "Login"
    Client-->>User: 302 Redirect to Auth (with PKCE Challenge)
    User->>Auth: Enter Credentials
    Auth-->>User: 302 Redirect to Callback (with Code)
    User->>Client: GET /callback?code=...
    Client->>Auth: POST /token (Code + PKCE Verifier)
    Auth-->>Client: JSON { access_token, refresh_token }
    Note right of Client: Client is now authenticated

Step 1: Authorization Request

from urllib.parse import urlencode
import secrets
 
def generate_authorization_url(client_id: str, redirect_uri: str) -> str:
    state = secrets.token_urlsafe(32)
    code_verifier = secrets.token_urlsafe(64)
 
    # PKCE: Generate code challenge
    import hashlib, base64
    code_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).decode().rstrip('=')
 
    params = {
        'response_type': 'code',
        'client_id': client_id,
        'redirect_uri': redirect_uri,
        'scope': 'openid profile email',
        'state': state,
        'code_challenge': code_challenge,
        'code_challenge_method': 'S256',
    }
 
    return f"https://auth.example.com/authorize?{urlencode(params)}"

Step 2: Token Exchange

import httpx
 
async def exchange_code_for_tokens(
    code: str,
    client_id: str,
    client_secret: str,
    redirect_uri: str,
    code_verifier: str,
) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://auth.example.com/token",
            data={
                'grant_type': 'authorization_code',
                'code': code,
                'redirect_uri': redirect_uri,
                'client_id': client_id,
                'client_secret': client_secret,
                'code_verifier': code_verifier,
            },
        )
        return response.json()

Step 3: Refreshing Tokens

Access tokens are short-lived. To stay logged in, we exchange the refresh token for a new pair of tokens:

async def refresh_access_token(
    refresh_token: str,
    client_id: str,
    client_secret: str,
) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://auth.example.com/token",
            data={
                'grant_type': 'refresh_token',
                'refresh_token': refresh_token,
                'client_id': client_id,
                'client_secret': client_secret,
            },
        )
        return response.json()

Step 4: Protected Middleware

Now we need to protect our API routes. Here is a simple middleware example using FastAPI:

from fastapi import Request, HTTPException
 
async def verify_token(request: Request):
    auth_header = request.headers.get('Authorization')
    if not auth_header or not auth_header.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing token")
    
    token = auth_header.split(' ')[1]
    # In production, verify JWT signature here!
    user_info = await fetch_user_info(token) 
    request.state.user = user_info

Security Considerations

When implementing OAuth2, these are the critical security measures:

Always use PKCE: Prevents authorization code interception
Validate the state parameter: Prevents CSRF attacks
Use short-lived access tokens: 15 minutes maximum
Rotate refresh tokens: One-time use with rotation
Validate redirect URIs: Exact match, no wildcards

Pro Tip: Store refresh tokens encrypted at rest, and implement token binding to prevent token theft from being useful.

Conclusion

BackendBytes Dev Team

Core Engineering Group

Building robust, scalable applications with modern best practices.

Implementing OAuth2 without Libraries

Disclaimer

The Authorization Code Grant Flow

Step 1: Authorization Request

Step 2: Token Exchange

Step 3: Refreshing Tokens

Step 4: Protected Middleware

Security Considerations

Conclusion

Read Next

HTTP/1.1 vs HTTP/2: A Visual Deep Dive

PostgreSQL Query Planner Internals: How Postgres Optimizes Your Queries

Zero-Downtime Database Migrations at Scale

HTTP/1.1 vs HTTP/2: A Visual Deep Dive

PostgreSQL Query Planner Internals: How Postgres Optimizes Your Queries

Zero-Downtime Database Migrations at Scale

Implementing OAuth2 without Libraries

Disclaimer

The Authorization Code Grant Flow

Step 1: Authorization Request

Step 2: Token Exchange

Step 3: Refreshing Tokens

Step 4: Protected Middleware

Security Considerations

Conclusion

Read Next

HTTP/1.1 vs HTTP/2: A Visual Deep Dive

PostgreSQL Query Planner Internals: How Postgres Optimizes Your Queries

Zero-Downtime Database Migrations at Scale

HTTP/1.1 vs HTTP/2: A Visual Deep Dive

PostgreSQL Query Planner Internals: How Postgres Optimizes Your Queries

Zero-Downtime Database Migrations at Scale