auth-architecture

name	auth-architecture
description	Design authentication and authorization architecture covering JWT, OAuth2, RBAC, session management, and MFA. Outputs auth flow diagrams, token strategy, permission models, and implementation patterns.
argument-hint	["application type","user types","compliance requirements","existing identity provider"]
allowed-tools	Read, Write, Bash

Auth Architecture

Authentication (who are you?) and authorization (what can you do?) are the most security-critical parts of any system. Getting them wrong is catastrophic; getting them right requires deliberate design before any code is written.

Process

1. Identify identity sources — internal users, external OAuth, service accounts, API consumers.
2. Choose token strategy — JWT vs. opaque tokens vs. session cookies; trade-offs matter.
3. Design the permission model — RBAC, ABAC, or resource-level permissions.
4. Map all auth flows — login, refresh, logout, MFA, password reset, service-to-service.
5. Plan token lifecycle — expiry, refresh windows, revocation strategy.
6. Define trust boundaries — what verifies tokens, which services are trusted callers.
7. Document attack scenarios — replay, CSRF, token theft, privilege escalation.

Output Format

Auth Strategy Decision Matrix

Scenario	Recommended Approach	Why
Web app, same domain	HttpOnly session cookie	CSRF-safe, can be revoked instantly
SPA + API, different domains	Short-lived JWT (15min) + refresh token	Stateless, revocable via refresh
Mobile app	Long-lived refresh + short access token	Secure storage, offline capability
Service-to-service	mTLS or signed JWT with short expiry	No user context, machine identity
Public API (third-party)	API key + optional OAuth	Simplicity for developers
Admin panel	Session cookie + MFA	Higher security, explicit logout

JWT Token Strategy

# auth/tokens.py
import jwt
import secrets
import hashlib
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass
from typing import Optional

@dataclass
class TokenPair:
    access_token: str
    refresh_token: str
    access_expires_at: datetime
    refresh_expires_at: datetime

class TokenService:
    ACCESS_TOKEN_TTL = timedelta(minutes=15)   # Short-lived
    REFRESH_TOKEN_TTL = timedelta(days=30)     # Rotated on each use

    def __init__(self, private_key: str, public_key: str, algorithm: str = "RS256"):
        self.private_key = private_key
        self.public_key = public_key
        self.algorithm = algorithm   # RS256 (asymmetric) — not HS256

    def create_token_pair(
        self,
        user_id: str,
        roles: list[str],
        permissions: list[str],
        device_id: str = None,
    ) -> TokenPair:
        now = datetime.now(timezone.utc)
        
        # Access token — short lived, contains claims
        access_payload = {
            : user_id,
            : now,
            : now + .ACCESS_TOKEN_TTL,
            : ,
            : roles,
            : permissions,
            : secrets.token_urlsafe(),  
        }
         device_id:
            access_payload[] = device_id

        access_token = jwt.encode(access_payload, .private_key, algorithm=.algorithm)

        
        refresh_token_raw = secrets.token_urlsafe()
        refresh_token_hash = hashlib.sha256(refresh_token_raw.encode()).hexdigest()

        
        refresh_expires = now + .REFRESH_TOKEN_TTL
        ._store_refresh_token(
            token_hash=refresh_token_hash,
            user_id=user_id,
            device_id=device_id,
            expires_at=refresh_expires,
        )

         TokenPair(
            access_token=access_token,
            refresh_token=refresh_token_raw,   
            access_expires_at=now + .ACCESS_TOKEN_TTL,
            refresh_expires_at=refresh_expires,
        )

     () -> :
        
        :
            payload = jwt.decode(
                token,
                .public_key,
                algorithms=[.algorithm],
                options={: [, , , ]}
            )
             payload.get() != :
                 jwt.InvalidTokenError()
             payload
         jwt.ExpiredSignatureError:
             TokenExpiredError()
         jwt.InvalidTokenError  e:
             TokenInvalidError()

     () -> TokenPair:
        
        token_hash = hashlib.sha256(refresh_token.encode()).hexdigest()
        stored = ._get_refresh_token(token_hash)

          stored:
            
            
             ._was_recently_rotated(token_hash):
                ._revoke_all_tokens_for_user(stored[])
                 TokenReuseError()
             TokenInvalidError()

         stored[] < datetime.now(timezone.utc):
             TokenExpiredError()

        
        ._revoke_refresh_token(token_hash)
        user = ._get_user(stored[])

         .create_token_pair(
            user_id=user.,
            roles=user.roles,
            permissions=user.permissions,
            device_id=stored.get(),
        )

     ():
        
        ._revoke_all_tokens_for_user(user_id)

RBAC Permission Model

# auth/permissions.py
from enum import Enum
from functools import wraps
from fastapi import Depends, HTTPException, status

# Define permissions as fine-grained strings
class Permission(str, Enum):
    # Orders
    ORDERS_READ = "orders:read"
    ORDERS_CREATE = "orders:create"
    ORDERS_UPDATE = "orders:update"
    ORDERS_DELETE = "orders:delete"
    ORDERS_CANCEL = "orders:cancel"
    # Users
    USERS_READ = "users:read"
    USERS_CREATE = "users:create"
    USERS_UPDATE = "users:update"
    USERS_DELETE = "users:delete"
    # Admin
    ADMIN_FULL = "admin:*"

# Role → Permission mappings
ROLE_PERMISSIONS: dict[str, set[str]] = {
    "customer": {
        Permission.ORDERS_READ,
        Permission.ORDERS_CREATE,
        Permission.ORDERS_CANCEL,
    },
    "support": {
        Permission.ORDERS_READ,
        Permission.ORDERS_UPDATE,
        Permission.USERS_READ,
    },
    "ops": {
        Permission.ORDERS_READ,
        Permission.ORDERS_UPDATE,
        Permission.ORDERS_DELETE,
        Permission.USERS_READ,
        Permission.USERS_UPDATE,
    },
    "admin": {
        Permission.ADMIN_FULL,
    },
}

def get_effective_permissions(roles: list[str]) -> set[str]:
    """Resolve effective permissions from roles — union of all role permissions."""
    perms = ()
     role  roles:
        perms.update(ROLE_PERMISSIONS.get(role, ()))
    
    
     Permission.ADMIN_FULL  perms:
         {p.value  p  Permission}
    
     perms


 ():
    
     ():
        user_permissions = (current_user.get(, []))
        
        missing = [p  p  permissions  p   user_permissions]
         missing:
             HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=
            )
         current_user
     dependency


 ():
    
     ():
        is_owner = current_user[] == resource_owner_id
        user_permissions = (current_user.get(, []))
        has_permission = (p  user_permissions  p  fallback_permissions)
        
          (is_owner  has_permission):
             HTTPException(status_code=, detail=)
         current_user
     dependency




  ():
    order =  order_service.get(order_id)
    
        current_user[]  order.user_id != current_user[]:
         HTTPException(status_code=, detail=)
     order


  ():
     order_service.delete(order_id)

OAuth2 Integration

# auth/oauth.py — Social login (Google, GitHub)
from fastapi import APIRouter, Request
from authlib.integrations.starlette_client import OAuth

router = APIRouter()
oauth = OAuth()

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

@router.get('/auth/google')
async def google_login(request: Request):
    redirect_uri = request.url_for('google_callback')
    return await oauth.google.authorize_redirect(request, redirect_uri)

@router.get('/auth/google/callback')
async def google_callback(request: Request):
    token = await oauth.google.authorize_access_token(request)
    userinfo = token.get('userinfo')
    
    if not userinfo or not userinfo.get('email_verified'):
        raise HTTPException(400, "Google login failed or email not verified")
    
    # Find or create user
    user = await user_service.find_or_create_oauth_user(
        provider="google",
        provider_id=userinfo["sub"],
        email=userinfo["email"],
        name=userinfo.get("name"),
    )
    
    
    tokens = token_service.create_token_pair(
        user_id=user.,
        roles=user.roles,
        permissions=get_effective_permissions(user.roles),
    )
    
    response = RedirectResponse(url=)
    response.set_cookie(
        key=,
        value=tokens.refresh_token,
        httponly=,        
        secure=,          
        samesite=,       
        max_age= *  * ,
    )
     response

MFA Implementation

# auth/mfa.py
import pyotp
import qrcode
import io
import base64

class MFAService:
    def setup_totp(self, user_id: str, email: str) -> dict:
        """Generate TOTP secret and provisioning URI."""
        secret = pyotp.random_base32()
        
        # Store encrypted secret (don't store plaintext)
        encrypted = encrypt(secret, key=settings.MFA_ENCRYPTION_KEY)
        db.store_mfa_secret(user_id, encrypted)
        
        # Generate QR code
        totp = pyotp.TOTP(secret)
        uri = totp.provisioning_uri(email, issuer_name="MyApp")
        
        qr = qrcode.make(uri)
        buffer = io.BytesIO()
        qr.save(buffer, format="PNG")
        qr_b64 = base64.b64encode(buffer.getvalue()).decode()
        
        return {
            "secret": secret,   # Show once for manual entry
            "qr_code": f"data:image/png;base64,{qr_b64}",
            "uri": uri,
        }
    
    def verify_totp(self, user_id: str, code: str) -> bool:
        """Verify TOTP code — accept 1 window before/after for clock skew."""
        encrypted_secret = db.get_mfa_secret(user_id)
        secret = decrypt(encrypted_secret, key=settings.MFA_ENCRYPTION_KEY)
        
        totp = pyotp.TOTP(secret)
        return totp.verify(code, valid_window=1)
    
    def generate_backup_codes(self, user_id: ) -> []:
        
        codes = [secrets.token_hex()  _  ()]
        hashed = [hashlib.sha256(c.encode()).hexdigest()  c  codes]
        db.store_backup_codes(user_id, hashed)
         codes  

Auth Flow Diagram

Login Flow:
  Client → POST /auth/login {email, password}
    → Verify credentials
    → [if MFA enabled] → Return {mfa_required: true, temp_token}
      → Client → POST /auth/mfa {code, temp_token}
    → Issue access_token (15min) + refresh_token (30d, httponly cookie)
    → Return {access_token, expires_at}

Authenticated Request:
  Client → GET /api/orders
    Authorization: Bearer <access_token>
  → Gateway verifies JWT signature + expiry
  → Extract claims (user_id, roles, permissions)
  → Forward to service with X-User-ID header

Token Refresh:
  Client → POST /auth/refresh
    Cookie: refresh_token=<opaque>
  → Verify refresh token in DB
  → Rotate: invalidate old, issue new pair
  → Return new {access_token}

Logout:
  Client → POST /auth/logout
    Cookie: refresh_token=<opaque>
  → Revoke refresh token in DB
  → Clear cookie
  
Force Logout All Devices:
  Client → POST /auth/logout-all
    Authorization: Bearer <access_token>
  → Revoke ALL refresh tokens for user

Rules

• Use RS256 (asymmetric) not HS256 — services can verify tokens without knowing the signing secret.
• Short access token TTL — 15 minutes maximum; longer TTLs can't be revoked.
• Rotate refresh tokens on every use — detect stolen tokens by watching for reuse.
• HttpOnly + Secure cookies for refresh tokens — never store refresh tokens in localStorage.
• Permissions in token, roles for management — embed fine-grained permissions in JWT, not just role names.
• Resource-level auth checks in code — JWT proves identity; code must still check ownership.
• Never return 403 with details — "not found" is safer than "forbidden" for resources users shouldn't know exist.
• Revocation on security events — force logout all sessions on password change, account compromise.
• MFA backup codes — always provide account recovery; store hashes, not plaintext.
• Audit all auth events — logins, failures, logouts, token revocations, permission denials.