Authentication & Security Patterns

This skill provides comprehensive authentication and security patterns for modern web applications in 2025, focusing on JWT tokens, OAuth2, multi-factor authentication, and zero-trust security principles that work across different frameworks and databases.

When to Use This Skill

Use this skill when you need to:

• Implement secure authentication with JWT tokens
• Set up OAuth2 social login providers
• Implement role-based access control (RBAC)
• Add multi-factor authentication (MFA)
• Secure API endpoints with proper middleware
• Handle session management and token refresh
• Implement zero-trust security patterns
• Set up WebSocket authentication
• Create audit trails and security logging

Modern Authentication Architecture

1. JWT Token Management with Refresh Tokens

# core/auth.py
import jwt
import secrets
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from passlib.context import CryptContext
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os

class TokenManager:
    """JWT token management with security best practices"""

    def __init__(self):
        self.secret_key = os.getenv("JWT_SECRET_KEY", self._generate_secret())
        self.algorithm = "HS256"
        self.access_token_expire = timedelta(minutes=15)
        self.refresh_token_expire = timedelta(days=7)
        self.pwd_context = CryptContext(
            schemes=["pbkdf2_sha256"],
            default="pbkdf2_sha256",
            pbkdf2_sha256__default_rounds=120000
        )

    def _generate_secret(self) -> str:
        """Generate cryptographically secure secret"""
        return secrets.token_urlsafe(32)

    def create_password_hash(self, password: str) -> str:
        """Create secure password hash"""
        return self.pwd_context.hash(password)

    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        """Verify password against hash"""
        return self.pwd_context.verify(plain_password, hashed_password)

    def create_access_token(
        self,
        data: Dict[str, Any],
        expires_delta: Optional[timedelta] = None
    ) -> str:
        """Create JWT access token"""
        to_encode = data.copy()

        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + self.access_token_expire

        to_encode.update({
            "exp": expire,
            "iat": datetime.utcnow(),
            "type": "access",
            "jti": secrets.token_urlsafe(16)  # JWT ID
        })

        encoded_jwt = jwt.encode(
            to_encode,
            self.secret_key,
            algorithm=self.algorithm
        )
        return encoded_jwt

    def create_refresh_token(
        self,
        user_id: str,
        device_info: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """Create secure refresh token with device binding"""
        jti = secrets.token_urlsafe(32)

        token_data = {
            "sub": user_id,
            "jti": jti,
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + self.refresh_token_expire,
            "type": "refresh",
            "device": device_info or {}
        }

        # Store refresh token in database or cache
        refresh_token = jwt.encode(
            token_data,
            self.secret_key,
            algorithm=self.algorithm
        )

        # Store token hash for revocation checking
        token_hash = self._hash_token(refresh_token)

        return {
            "token": refresh_token,
            "jti": jti,
            "expires_at": token_data["exp"],
            "token_hash": token_hash
        }

    def verify_token(self, token: str, token_type: str = "access") -> Dict[str, Any]:
        """Verify and decode JWT token"""
        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=[self.algorithm],
                options={"verify_exp": True}
            )

            if payload.get("type") != token_type:
                raise ValueError("Invalid token type")

            return payload

        except jwt.ExpiredSignatureError:
            raise ValueError("Token has expired")
        except jwt.JWTError:
            raise ValueError("Invalid token")

    def revoke_token(self, jti: str):
        """Revoke a token (add to blacklist)"""
        # Implement token blacklisting (Redis or database)
        pass

    def _hash_token(self, token: str) -> str:
        """Hash token for storage"""
        return self.pwd_context.hash(token)

# Singleton instance
token_manager = TokenManager()

2. OAuth2 Provider Integration

# core/oauth.py
from typing import Dict, Any, Optional
from abc import ABC, abstractmethod
import httpx
from urllib.parse import urlencode, parse_qs

class OAuth2Provider(ABC):
    """Base OAuth2 provider implementation"""

    def __init__(
        self,
        client_id: str,
        client_secret: str,
        redirect_uri: str,
        scopes: list
    ):
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri
        self.scopes = scopes

    @abstractmethod
    def get_authorization_url(self, state: str) -> str:
        """Get OAuth2 authorization URL"""
        pass

    @abstractmethod
    async def exchange_code_for_token(self, code: str, state: str) -> Dict[str, Any]:
        """Exchange authorization code for access token"""
        pass

    @abstractmethod
    async def get_user_info(self, access_token: str) -> Dict[str, Any]:
        """Get user information from provider"""
        pass

class GoogleOAuth2Provider(OAuth2Provider):
    """Google OAuth2 provider implementation"""

    AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
    TOKEN_URL = "https://oauth2.googleapis.com/token"
    USER_INFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"

    def get_authorization_url(self, state: str) -> str:
        params = {
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "response_type": "code",
            "scope": " ".join(self.scopes),
            "state": state,
            "access_type": "offline",  # For refresh tokens
            "prompt": "consent"
        }
        return f"{self.AUTH_URL}?{urlencode(params)}"

    async def exchange_code_for_token(self, code: str, state: str) -> Dict[str, Any]:
        async with httpx.AsyncClient() as client:
            data = {
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "code": code,
                "grant_type": "authorization_code",
                "redirect_uri": self.redirect_uri
            }

            response = await client.post(self.TOKEN_URL, data=data)
            response.raise_for_status()

            return response.json()

    async def get_user_info(self, access_token: str) -> Dict[str, Any]:
        async with httpx.AsyncClient() as client:
            headers = {"Authorization": f"Bearer {access_token}"}
            response = await client.get(
                self.USER_INFO_URL,
                headers=headers
            )
            response.raise_for_status()
            return response.json()

class GitHubOAuth2Provider(OAuth2Provider):
    """GitHub OAuth2 provider implementation"""

    AUTH_URL = "https://github.com/login/oauth/authorize"
    TOKEN_URL = "https://github.com/login/oauth/access_token"
    USER_INFO_URL = "https://api.github.com/user"

    def get_authorization_url(self, state: str) -> str:
        params = {
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "response_type": "code",
            "scope": " ".join(self.scopes),
            "state": state
        }