Auth Patterns

Authentication Patterns

JWT Implementation

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.hash import argon2

SECRET_KEY = os.environ["JWT_SECRET"]  # 256-bit random key
ALGORITHM = "HS256"

def create_access_token(user_id: int) -> str:
    payload = {
        "sub": str(user_id),
        "iat": datetime.now(timezone.utc),
        "exp": datetime.now(timezone.utc) + timedelta(minutes=15),
        "type": "access"
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(user_id: int) -> str:
    payload = {
        "sub": str(user_id),
        "iat": datetime.now(timezone.utc),
        "exp": datetime.now(timezone.utc) + timedelta(days=30),
        "type": "refresh",
        "jti": str(uuid4())  # unique ID for revocation
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str, token_type: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != token_type:
            raise JWTError("Wrong token type")
        if is_revoked(payload.get("jti")):
            raise JWTError("Token revoked")
        return payload
    except JWTError:
        raise HTTPException(401, "Invalid token")

OAuth2 / OIDC Flow

# Authorization Code Flow with PKCE (public clients)
import secrets, hashlib, base64

# 1. Generate PKCE challenge
code_verifier = secrets.token_urlsafe(64)
code_challenge = base64.urlsafe_b64encode(
    hashlib.sha256(code_verifier.encode()).digest()
).rstrip(b'=').decode()

# 2. Redirect to authorization server
auth_url = (
    f"https://auth.example.com/authorize"
    f"?response_type=code"
    f"&client_id={CLIENT_ID}"
    f"&redirect_uri={REDIRECT_URI}"
    f"&scope=openid profile email"
    f"&state={secrets.token_urlsafe(16)}"
    f"&code_challenge={code_challenge}"
    f"&code_challenge_method=S256"
)

# 3. Exchange code for tokens
import httpx
token_response = httpx.post("https://auth.example.com/token", data={
    "grant_type": "authorization_code",
    "code": auth_code,
    "redirect_uri": REDIRECT_URI,
    "client_id": CLIENT_ID,
    "code_verifier": code_verifier  # PKCE verification
})
tokens = token_response.json()
# tokens: access_token, refresh_token, id_token

# 4. Verify ID token (OIDC)
from jose import jwt as jose_jwt
import httpx
jwks = httpx.get("https://auth.example.com/.well-known/jwks.json").json()
claims = jose_jwt.decode(tokens["id_token"], jwks,
                         algorithms=["RS256"], audience=CLIENT_ID)

Refresh Token Rotation

# Store refresh tokens in DB with rotation
@app.post("/auth/refresh")
async def refresh(refresh_token: str):
    payload = verify_token(refresh_token, "refresh")
    jti = payload["jti"]

    # Check token family — detect theft via rotation
    stored = await db.get_refresh_token(jti)
    if not stored or stored.used:
        # Token reuse detected → revoke entire family
        await db.revoke_family(stored.family_id if stored else None)
        raise HTTPException(401, "Token reuse detected")

    # Rotate: mark old as used, issue new
    await db.mark_used(jti)
    new_access = create_access_token(payload["sub"])
    new_refresh = create_refresh_token(payload["sub"])
    await db.store_refresh_token(new_refresh, family_id=stored.family_id)

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

MFA — TOTP

import pyotp, qrcode

# Setup
def setup_mfa(user_email: str) -> tuple[str, str]:
    secret = pyotp.random_base32()
    totp = pyotp.TOTP(secret)
    uri = totp.provisioning_uri(user_email, issuer_name="MyApp")
    return secret, uri  # show QR code of uri to user

# Verify
def verify_mfa(secret: str, code: str) -> bool:
    totp = pyotp.TOTP(secret)
    return totp.verify(code, valid_window=1)  # ±30 seconds tolerance

# Store secret encrypted in DB
encrypted_secret = cipher.encrypt(secret.encode())

Passkeys (WebAuthn)

from webauthn import generate_registration_options, verify_registration_response
from webauthn.helpers.structs import ResidentKeyRequirement

# Registration challenge
options = generate_registration_options(
    rp_id="example.com",
    rp_name="My App",
    user_id=user.id.to_bytes(8, 'big'),
    user_name=user.email,
    resident_key=ResidentKeyRequirement.REQUIRED
)

# Verification (after browser creates credential)
verification = verify_registration_response(
    credential=registration_response,
    expected_challenge=stored_challenge,
    expected_rp_id="example.com",
    expected_origin="https://example.com"
)
# Store: verification.credential_id, verification.credential_public_key

Session Security

# Secure cookie attributes
SESSION_COOKIE = {
    "key": "session_id",
    "httponly": True,   # no JS access
    "secure": True,     # HTTPS only
    "samesite": "Strict",  # CSRF protection
    "max_age": 86400,   # 24h
    "path": "/",
}

# Session fixation prevention — regenerate ID on login
async def login(user: User, request: Request):
    old_session = request.session.copy()
    request.session.clear()
    request.session.update(old_session)
    request.session["user_id"] = user.id  # new session ID issued