Authentication Deep Dive

# jwt_create.py
import jwt
import secrets
from datetime import datetime, timezone, timedelta

# For symmetric signing (HS256), shared secret
SECRET_KEY = secrets.token_hex(32)

def create_access_token(user_id: int, roles: list[str]) -> str:
    """Create a short-lived access token."""
    now = datetime.now(timezone.utc)
    payload = {
        "sub": str(user_id),          # Subject (user ID)
        "iat": now,                    # Issued at
        "exp": now + timedelta(minutes=15),  # Expires in 15 min
        "roles": roles,
        "type": "access",
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def create_refresh_token(user_id: int) -> str:
    """Create a long-lived refresh token."""
    now = datetime.now(timezone.utc)
    payload = {
        "sub": str(user_id),
        "iat": now,
        "exp": now + timedelta(days=30),
        "type": "refresh",
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

access_token = create_access_token(42, roles=["user", "admin"])
refresh_token = create_refresh_token(42)

print(f"Access token:  {access_token[:50]}...")
print(f"Refresh token: {refresh_token[:50]}...")

# jwt_verify.py, continued
def verify_token(token: str, expected_type: str = "access") -> dict | None:
    """
    Verify a JWT. Returns payload if valid, None if invalid.
    Raises are caught internally, never let JWT errors leak to clients.
    """
    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=["HS256"],
            options={"require": ["exp", "iat", "sub"]},
        )
        if payload.get("type") != expected_type:
            return None
        return payload
    except jwt.ExpiredSignatureError:
        print("Token expired")
        return None
    except jwt.InvalidTokenError as e:
        print(f"Invalid token: {e}")
        return None

# Verify the access token
payload = verify_token(access_token, "access")
if payload:
    print(f"Authenticated: user_id={payload['sub']}, roles={payload['roles']}")

# Tampered token, signature verification FAILS
tampered = access_token[:-10] + "AAAAAAAAAA"
result = verify_token(tampered)
print(f"Tampered token: {result}")   # None

# Algorithm confusion attack, always specify algorithms explicitly
try:
    jwt.decode(access_token, SECRET_KEY, algorithms=["none"])
except Exception as e:
    print(f"Algorithm=none rejected: {type(e).__name__}")

# token_refresh.py, Token refresh + revocation via blocklist
import time

# In production: use Redis SET with TTL for the blocklist
REFRESH_TOKEN_BLOCKLIST: set[str] = set()

def refresh_access_token(refresh_token: str) -> str | None:
    """Exchange a valid refresh token for a new access token."""
    # Check blocklist first (revoked tokens)
    if refresh_token in REFRESH_TOKEN_BLOCKLIST:
        print("Refresh token revoked")
        return None

    payload = verify_token(refresh_token, expected_type="refresh")
    if payload is None:
        return None

    user_id = int(payload["sub"])
    # In production: look up user roles from database
    return create_access_token(user_id, roles=["user"])

def logout(refresh_token: str) -> None:
    """Revoke a refresh token."""
    REFRESH_TOKEN_BLOCKLIST.add(refresh_token)

# Simulate refresh flow
new_access = refresh_access_token(refresh_token)
print(f"New access token: {new_access[:40] if new_access else None}...")

# Logout, revoke refresh token
logout(refresh_token)
result = refresh_access_token(refresh_token)
print(f"After logout: {result}")   # None

# oauth2_client_credentials.py, Machine-to-machine auth
import requests

def get_oauth2_token(
    token_url: str,
    client_id: str,
    client_secret: str,
    scope: str = "",
) -> dict:
    """
    OAuth2 Client Credentials flow.
    No user interaction required, used for service accounts.
    """
    response = requests.post(
        token_url,
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
            "scope": scope,
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

def call_api_with_token(api_url: str, access_token: str) -> dict:
    """Call a protected API using the Bearer token."""
    response = requests.get(
        api_url,
        headers={"Authorization": f"Bearer {access_token}"},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

# Example usage (requires a real OAuth2 server)
# token_data = get_oauth2_token(
#     token_url="https://auth.example.com/oauth/token",
#     client_id="my-service",
#     client_secret="secret-from-secrets-manager",
#     scope="read:data write:data",
# )
# print(f"Access token type: {token_data['token_type']}")
# print(f"Expires in: {token_data['expires_in']} seconds")
# result = call_api_with_token("https://api.example.com/data", token_data["access_token"])

print("OAuth2 Client Credentials pattern: POST grant_type=client_credentials")
print("Token stored in memory (not disk), re-fetched when expired")
print()
print("OpenID Connect adds:")
print("  • id_token (JWT with user identity)")
print("  • /userinfo endpoint for profile data")
print("  • Discovery endpoint (.well-known/openid-configuration)")