API Authentication

API Authentication Patterns

Comprehensive guide to implementing secure API authentication including JWT, OAuth 2.0, API keys, and session-based patterns. Covers when to use each approach, security best practices, and common vulnerabilities to avoid.

Quick Reference

When to use this skill:

• Implementing API authentication
• Choosing between auth strategies (JWT vs OAuth vs sessions)
• Securing API endpoints
• Implementing token refresh logic
• Debugging authentication issues
• Preventing auth vulnerabilities

Common triggers:

• "How should I implement authentication"
• "JWT vs OAuth vs API keys"
• "How to secure this API"
• "Implement refresh tokens"
• "Store authentication tokens securely"
• "Fix authentication vulnerability"

Prevents vulnerabilities:

• Token theft and replay attacks
• Insecure token storage
• Missing token expiration
• Weak password hashing
• CSRF attacks

Part 1: Authentication Strategy Decision Matrix

When to Use Each Pattern

Decision Tree

START: What type of client?

├─ Mobile app or SPA?
│  └─ Use JWT (stateless, works across domains)
│
├─ Third-party integration?
│  └─ Use OAuth 2.0 (delegation, scoped permissions)
│
├─ Service-to-service?
│  └─ Use API Keys (simple, rate-limitable)
│
└─ Traditional web app (same domain)?
   └─ Use Sessions (revocable, server-controlled)

Part 2: JWT (JSON Web Tokens)

JWT Structure

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

[HEADER].[PAYLOAD].[SIGNATURE]

Header (algorithm and type):

{
  "alg": "HS256",
  "typ": "JWT"
}

Payload (claims):

{
  "sub": "1234567890",    // Subject (user ID)
  "name": "John Doe",     // Custom claim
  "iat": 1516239022,      // Issued at
  "exp": 1516242622       // Expires at (required!)
}

Signature (verification):

HMACSHA256(
  base64UrlEncode(header) + "." + base64UrlEncode(payload),
  secret
)

JWT Implementation (Python)

import jwt
import datetime
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

SECRET_KEY = "your-256-bit-secret"  # Must be strong, from environment
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7

security = HTTPBearer()

def create_access_token(user_id: int) -> str:
    """Create short-lived access token."""
    expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {
        "sub": str(user_id),
        "exp": expires,
        "iat": datetime.datetime.utcnow(),
        "type": "access"
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(user_id: int) -> str:
    """Create long-lived refresh token."""
    expires = datetime.datetime.utcnow() + datetime.timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    payload = {
        "sub": str(user_id),
        "exp": expires,
        "iat": datetime.datetime.utcnow(),
        "type": "refresh"
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)) -> dict:
    """Verify and decode JWT token."""
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "access":
            raise HTTPException(status_code=401, detail="Invalid token type")
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Login endpoint
@app.post("/login")
async def login(username: str, password: str):
    user = authenticate_user(username, password)  # Your auth logic
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    access_token = create_access_token(user.id)
    refresh_token = create_refresh_token(user.id)

    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

# Protected endpoint
@app.get("/protected")
async def protected_route(payload: dict = Depends(verify_token)):
    user_id = payload["sub"]
    return {"message": f"Hello user {user_id}"}

# Refresh token endpoint
@app.post("/refresh")
async def refresh(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=401, detail="Invalid token type")

        # Generate new access token
        user_id = int(payload["sub"])
        new_access_token = create_access_token(user_id)

        return {"access_token": new_access_token, "token_type": "bearer"}
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Refresh token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

JWT Security Best Practices

✅ Do:

• Use strong secret keys (256-bit minimum)
• Always set expiration (exp claim)
• Use short-lived access tokens (15 minutes)
• Use separate refresh tokens (7 days)
• Store secret in environment variables
• Use HTTPS only
• Validate signature on every request
• Check token type (access vs refresh)

❌ Don't:

• Store sensitive data in payload (it's base64, not encrypted!)
• Use symmetric signing (HS256) for public APIs (use RS256)
• Store tokens in localStorage (XSS vulnerability)
• Skip expiration validation
• Use same token for access and refresh
• Hard-code secrets

Token Storage (Client-Side)

❌ Bad (localStorage - vulnerable to XSS):

localStorage.setItem('token', token);  // XSS can steal this!

✅ Good (httpOnly cookie):

# Server sets httpOnly cookie
response.set_cookie(
    key="access_token",
    value=access_token,
    httponly=True,  # Not accessible via JavaScript
    secure=True,    # HTTPS only
    samesite="lax", # CSRF protection
    max_age=900     # 15 minutes
)

✅ Also Good (memory only for SPAs):

// Store in memory (lost on refresh, but more secure)
let accessToken = null;

async function login(username, password) {
  const response = await fetch('/login', {
    method: 'POST',
    body: JSON.stringify({ username, password })
  });
  const data = await response.json();
  accessToken = data.access_token;  // Store in memory
}

Part 3: OAuth 2.0

OAuth 2.0 Flows

Authorization Code Flow (most common, for web apps):

1. Client → Authorization Server: "User wants to log in"
2. Authorization Server → User: Login page
3. User → Authorization Server: Credentials
4. Authorization Server → Client: Authorization code
5. Client → Authorization Server: Exchange code for access token
6. Authorization Server → Client: Access token + refresh token

Client Credentials Flow (for service-to-service):

1. Service → Authorization Server: Client ID + Secret
2. Authorization Server → Service: Access token

OAuth 2.0 Implementation (Authorization Code Flow)

from fastapi import FastAPI, HTTPException
from authlib.integrations.starlette_client import OAuth
import os

app = FastAPI()

oauth = OAuth()
oauth.register(
    name='google',
    client_id=os.getenv('GOOGLE_CLIENT_ID'),
    client_secret=os.getenv('GOOGLE_CLIENT_SECRET'),
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

@app.get('/login/google')
async def login_google(request: Request):
    redirect_uri = request.url_for('auth_google')
    return await oauth.google.authorize_redirect(request, redirect_uri)

@app.get('/auth/google')
async def auth_google(request: Request):
    try:
        token = await oauth.google.authorize_access_token(request)
        user_info = token.get('userinfo')

        # Create or update user in your database
        user = get_or_create_user(
            email=user_info['email'],
            name=user_info['name']
        )

        # Create your own JWT for subsequent requests
        access_token = create_access_token(user.id)

        return {"access_token": access_token, "token_type": "bearer"}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

OAuth 2.0 Scopes

# Define scopes for your API
SCOPES = {
    "read:posts": "Read posts",
    "write:posts": "Create and edit posts",
    "delete:posts": "Delete posts",
    "read:profile": "Read user profile",
    "write:profile": "Update user profile"
}

# Include scopes in JWT
def create_access_token(user_id: int, scopes: list[str]) -> str:
    payload = {
        "sub": str(user_id),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=15),
        "scopes": scopes  # Add scopes to token
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

# Check scopes in protected endpoint
def require_scopes(required_scopes: list[str]):
    def decorator(func):
        async def wrapper(payload: dict = Depends(verify_token)):
            token_scopes = payload.get("scopes", [])
            if not all(scope in token_scopes for scope in required_scopes):
                raise HTTPException(status_code=403, detail="Insufficient permissions")
            return await func(payload)
        return wrapper
    return decorator

@app.delete("/posts/{post_id}")
@require_scopes(["delete:posts"])
async def delete_post(post_id: int, payload: dict = Depends(verify_token)):
    # User has delete:posts scope
    pass

Part 4: API Keys

API Key Implementation

import secrets
import hashlib
from datetime import datetime

# Generate API key
def generate_api_key() -> tuple[str, str]:
    """Generate API key and return (key, hashed_key)."""
    # Generate random 32-byte key
    api_key = secrets.token_urlsafe(32)

    # Hash for storage (never store plain key!)
    hashed_key = hashlib.sha256(api_key.encode()).hexdigest()

    return api_key, hashed_key

# Store in database
def create_api_key(user_id: int, name: str) -> str:
    api_key, hashed_key = generate_api_key()

    db.execute("""
        INSERT INTO api_keys (user_id, name, key_hash, created_at)
        VALUES (?, ?, ?, ?)
    """, user_id, name, hashed_key, datetime.utcnow())

    # Return plain key to user (only time they see it!)
    return api_key

# Verify API key
def verify_api_key(api_key: str) -> dict:
    """Verify API key and return user info."""
    hashed_key = hashlib.sha256(api_key.encode()).hexdigest()

    result = db.execute("""
        SELECT user_id, name, created_at, last_used_at
        FROM api_keys
        WHERE key_hash = ? AND revoked_at IS NULL
    """, hashed_key).fetchone()

    if not result:
        raise HTTPException(status_code=401, detail="Invalid API key")

    # Update last used timestamp
    db.execute("""
        UPDATE api_keys
        SET last_used_at = ?
        WHERE key_hash = ?
    """, datetime.utcnow(), hashed_key)

    return {"user_id": result[0], "key_name": result[1]}

# Middleware
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

@app.get("/api/data")
async def get_data(api_key: str = Security(api_key_header)):
    user_info = verify_api_key(api_key)
    return {"data": "protected data", "user_id": user_info["user_id"]}

API Key Best Practices

✅ Do:

• Hash keys before storing (use SHA-256 minimum)
• Generate cryptographically secure keys (secrets module)
• Allow users to name keys ("Production Server", "CI/CD")
• Track last used timestamp
• Allow key revocation
• Rate limit by API key
• Log API key usage

❌ Don't:

• Store plain text keys
• Use predictable key generation
• Expose keys in URLs (use headers)
• Share keys across environments

API Key Revocation

@app.delete("/api-keys/{key_id}")
async def revoke_api_key(key_id: int, current_user: dict = Depends(get_current_user)):
    db.execute("""
        UPDATE api_keys
        SET revoked_at = ?
        WHERE id = ? AND user_id = ?
    """, datetime.utcnow(), key_id, current_user["id"])

    return {"message": "API key revoked"}

Part 5: Session-Based Authentication

Session Implementation

from fastapi import FastAPI, Cookie, Response
import redis
import secrets
import json

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

def create_session(user_id: int) -> str:
    """Create session and return session ID."""
    session_id = secrets.token_urlsafe(32)

    session_data = {
        "user_id": user_id,
        "created_at": datetime.utcnow().isoformat()
    }

    # Store in Redis with 24-hour expiry
    redis_client.setex(
        f"session:{session_id}",
        86400,  # 24 hours
        json.dumps(session_data)
    )

    return session_id

def verify_session(session_id: str) -> dict:
    """Verify session and return user data."""
    session_data = redis_client.get(f"session:{session_id}")

    if not session_data:
        raise HTTPException(status_code=401, detail="Session expired")

    return json.loads(session_data)

@app.post("/login")
async def login(username: str, password: str, response: Response):
    user = authenticate_user(username, password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Create session
    session_id = create_session(user.id)

    # Set httpOnly cookie
    response.set_cookie(
        key="session_id",
        value=session_id,
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=86400  # 24 hours
    )

    return {"message": "Logged in successfully"}

@app.get("/protected")
async def protected_route(session_id: str = Cookie(None)):
    if not session_id:
        raise HTTPException(status_code=401, detail="Not authenticated")

    session_data = verify_session(session_id)
    user_id = session_data["user_id"]

    return {"message": f"Hello user {user_id}"}

@app.post("/logout")
async def logout(session_id: str = Cookie(None), response: Response):
    if session_id:
        redis_client.delete(f"session:{session_id}")

    response.delete_cookie("session_id")
    return {"message": "Logged out successfully"}

Part 6: Password Security

Password Hashing (Never Store Plain Text!)

import bcrypt

def hash_password(password: str) -> str:
    """Hash password with bcrypt."""
    salt = bcrypt.gensalt(rounds=12)  # 12 rounds is good balance
    hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
    return hashed.decode('utf-8')

def verify_password(password: str, hashed: str) -> bool:
    """Verify password against hash."""
    return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8'))

# When creating user
@app.post("/register")
async def register(username: str, password: str):
    # Validate password strength
    if len(password) < 12:
        raise HTTPException(status_code=400, detail="Password must be at least 12 characters")

    # Hash password
    hashed_password = hash_password(password)

    # Store hashed password (NEVER plain text!)
    db.execute("""
        INSERT INTO users (username, password_hash)
        VALUES (?, ?)
    """, username, hashed_password)

    return {"message": "User created"}

# When logging in
@app.post("/login")
async def login(username: str, password: str):
    user = db.execute("SELECT id, password_hash FROM users WHERE username = ?", username).fetchone()

    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Verify password
    if not verify_password(password, user[1]):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Create token/session
    access_token = create_access_token(user[0])
    return {"access_token": access_token}

Password Requirements

Minimum requirements:

• At least 12 characters (NIST recommendation)
• Mix of uppercase, lowercase, numbers, symbols
• Not in common password list
• Not similar to username

Implementation:

import re

def validate_password(password: str, username: str) -> tuple[bool, str]:
    """Validate password strength."""
    if len(password) < 12:
        return False, "Password must be at least 12 characters"

    if not re.search(r"[a-z]", password):
        return False, "Password must contain lowercase letter"

    if not re.search(r"[A-Z]", password):
        return False, "Password must contain uppercase letter"

    if not re.search(r"\d", password):
        return False, "Password must contain number"

    if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", password):
        return False, "Password must contain special character"

    if username.lower() in password.lower():
        return False, "Password cannot contain username"

    # Check against common passwords
    if is_common_password(password):
        return False, "Password is too common"

    return True, "Password is strong"

Part 7: Common Vulnerabilities

Vulnerability 1: Token Replay Attacks

Problem: Attacker intercepts token and reuses it

✅ Solution: Short expiration + refresh tokens

# Access token: 15 minutes
# Refresh token: 7 days, single-use

def refresh_tokens(refresh_token: str):
    # Verify refresh token
    payload = jwt.decode(refresh_token, SECRET_KEY)

    # Check if already used (store used tokens in Redis)
    if redis_client.get(f"used:{refresh_token}"):
        raise HTTPException(status_code=401, detail="Token already used")

    # Mark as used
    redis_client.setex(f"used:{refresh_token}", 604800, "1")  # 7 days

    # Generate new tokens
    user_id = int(payload["sub"])
    new_access = create_access_token(user_id)
    new_refresh = create_refresh_token(user_id)

    return {"access_token": new_access, "refresh_token": new_refresh}

Vulnerability 2: CSRF Attacks

Problem: Attacker tricks user into making authenticated request

✅ Solution: CSRF tokens + SameSite cookies

from fastapi import Cookie, Header

def verify_csrf(
    csrf_token: str = Header(None, alias="X-CSRF-Token"),
    session_id: str = Cookie(None)
):
    """Verify CSRF token matches session."""
    if not csrf_token:
        raise HTTPException(status_code=403, detail="CSRF token missing")

    session_data = verify_session(session_id)
    stored_csrf = session_data.get("csrf_token")

    if csrf_token != stored_csrf:
        raise HTTPException(status_code=403, detail="Invalid CSRF token")

@app.post("/sensitive-action")
async def sensitive_action(
    csrf_check: None = Depends(verify_csrf)
):
    # Action protected from CSRF
    pass

Vulnerability 3: Timing Attacks

Problem: Attacker uses response timing to guess credentials

✅ Solution: Constant-time comparison

import hmac

def constant_time_compare(a: str, b: str) -> bool:
    """Compare strings in constant time (prevents timing attacks)."""
    return hmac.compare_digest(a, b)

# Use for password hash comparison
if not constant_time_compare(provided_hash, stored_hash):
    raise HTTPException(status_code=401)

Part 8: Rate Limiting

from fastapi import Request
import time

# In-memory rate limiter (use Redis for production)
rate_limits = {}

def rate_limit(max_requests: int, window_seconds: int):
    """Rate limit decorator."""
    def decorator(func):
        async def wrapper(request: Request, *args, **kwargs):
            client_ip = request.client.host
            key = f"{client_ip}:{func.__name__}"

            now = time.time()

            if key not in rate_limits:
                rate_limits[key] = []

            # Remove old requests outside window
            rate_limits[key] = [
                req_time for req_time in rate_limits[key]
                if now - req_time < window_seconds
            ]

            # Check if over limit
            if len(rate_limits[key]) >= max_requests:
                raise HTTPException(
                    status_code=429,
                    detail=f"Rate limit exceeded. Try again in {window_seconds} seconds."
                )

            # Add current request
            rate_limits[key].append(now)

            return await func(request, *args, **kwargs)
        return wrapper
    return decorator

@app.post("/login")
@rate_limit(max_requests=5, window_seconds=60)  # 5 login attempts per minute
async def login(request: Request, username: str, password: str):
    # Login logic
    pass

Quick Security Checklist

Token Security:

• Always use HTTPS (never HTTP)
• Set token expiration (15 min for access, 7 days for refresh)
• Store secrets in environment variables
• Hash API keys before storing
• Use httpOnly cookies for tokens
• Never store sensitive data in JWT payload

Password Security:

• Use bcrypt or argon2 for hashing
• Enforce minimum 12 characters
• Never store plain text passwords
• Use constant-time comparison
• Implement rate limiting on login

API Security:

• Validate all inputs
• Implement rate limiting
• Use CORS restrictions
• Add CSRF protection for state-changing operations
• Log authentication events
• Monitor for suspicious patterns

Resources

JWT:

• JWT.io: https://jwt.io/
• RFC 7519: https://tools.ietf.org/html/rfc7519

OAuth 2.0:

• OAuth 2.0 RFC: https://tools.ietf.org/html/rfc6749
• OAuth 2.0 Playground: https://www.oauth.com/playground/

Security:

• OWASP Auth Cheatsheet: https://cheatsheetseries.owasp.org/cheatsheets/Authentication_Cheat_Sheet.html
• NIST Password Guidelines: https://pages.nist.gov/800-63-3/

Libraries:

• PyJWT: https://pyjwt.readthedocs.io/
• Authlib: https://docs.authlib.org/
• Passlib: https://passlib.readthedocs.io/