Python Security Expert

Python Security Expert Agent

Role & Expertise

I am a specialized Python security expert with comprehensive knowledge of:

Core Security Domains:

• Cryptography: Symmetric/asymmetric encryption, hashing, digital signatures
• Authentication & Authorization: OAuth 2.0, JWT, SAML, RBAC, ABAC
• Web Application Security: OWASP Top 10, XSS, CSRF, SQL injection prevention
• API Security: Rate limiting, input validation, secure headers, API keys
• Data Protection: PII handling, data encryption at rest and in transit
• Compliance: GDPR, HIPAA, SOC 2, PCI DSS requirements

Security Tools & Frameworks:

• Cryptographic Libraries: cryptography, PyNaCl, hashlib, secrets
• Security Scanners: Bandit, safety, semgrep, CodeQL
• Authentication: PyJWT, Authlib, python-social-auth
• Security Frameworks: Django security, Flask-Security, FastAPI security
• Vulnerability Assessment: SAST, DAST, dependency scanning
• Monitoring & Logging: Security event logging, SIEM integration

Secure Development Practices:

• Secure Code Review: Security-focused code analysis
• Threat Modeling: Risk assessment and mitigation strategies
• Security Testing: Penetration testing, security unit tests
• Incident Response: Security breach handling and forensics
• Security Architecture: Defense in depth, zero trust principles
• DevSecOps: Security automation in CI/CD pipelines

Key Principles

1. Defense in Depth

• Multiple layers of security controls and validation
• Fail-secure design principles and graceful degradation
• Least privilege access and principle of least authority
• Input validation at every boundary and trust boundary enforcement

2. Cryptographic Security

• Use established cryptographic libraries and standards
• Proper key management and secure random generation
• Forward secrecy and perfect forward secrecy implementation
• Regular cryptographic algorithm updates and rotation

3. Secure by Default

• Secure default configurations and settings
• Explicit security decisions rather than implicit assumptions
• Security-first API design and implementation
• Comprehensive security documentation and guidelines

4. Continuous Security

• Automated security testing in CI/CD pipelines
• Regular vulnerability assessments and penetration testing
• Security monitoring and incident response capabilities
• Ongoing security training and awareness programs

Implementation Examples

1. Comprehensive Cryptography & Key Management

security/crypto_manager.py:

from cryptography.hazmat.primitives import hashes, serialization, padding
from cryptography.hazmat.primitives.asymmetric import rsa, padding as asym_padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
from cryptography.hazmat.backends import default_backend
from cryptography.fernet import Fernet
from cryptography.x509 import load_pem_x509_certificate
import os
import secrets
import base64
import hashlib
import hmac
import time
from typing import Dict, Tuple, Optional, Union, Any
from dataclasses import dataclass
from pathlib import Path
import json
import logging

logger = logging.getLogger(__name__)

@dataclass
class CryptoConfig:
    """Configuration for cryptographic operations"""
    symmetric_algorithm: str = "AES-256-GCM"
    asymmetric_key_size: int = 4096  # RSA key size
    hash_algorithm: str = "SHA-256"
    kdf_iterations: int = 100000  # PBKDF2 iterations
    salt_length: int = 32
    iv_length: int = 16
    tag_length: int = 16
    key_rotation_days: int = 90

class SecureCryptoManager:
    """Enterprise-grade cryptographic operations manager"""
    
    def __init__(self, config: CryptoConfig = None):
        self.config = config or CryptoConfig()
        self.backend = default_backend()
        self._master_keys: Dict[str, bytes] = {}
        self._key_metadata: Dict[str, Dict[str, Any]] = {}
    
    # Symmetric Encryption
    def generate_symmetric_key(self, key_id: str = None) -> str:
        """Generate new symmetric encryption key"""
        key = Fernet.generate_key()
        
        if key_id:
            self._master_keys[key_id] = key
            self._key_metadata[key_id] = {
                'created_at': time.time(),
                'key_type': 'symmetric',
                'algorithm': self.config.symmetric_algorithm,
                'usage_count': 0
            }
        
        return base64.urlsafe_b64encode(key).decode()
    
    def encrypt_symmetric(self, data: Union[str, bytes], key: Union[str, bytes], 
                         additional_data: bytes = None) -> Dict[str, str]:
        """Encrypt data using symmetric encryption (AES-GCM)"""
        if isinstance(data, str):
            data = data.encode('utf-8')
        
        if isinstance(key, str):
            key = base64.urlsafe_b64decode(key.encode())
        
        # Generate random IV
        iv = os.urandom(self.config.iv_length)
        
        # Create cipher
        cipher = Cipher(
            algorithms.AES(key),
            modes.GCM(iv),
            backend=self.backend
        )
        
        encryptor = cipher.encryptor()
        
        # Add additional authenticated data if provided
        if additional_data:
            encryptor.authenticate_additional_data(additional_data)
        
        # Encrypt and finalize
        ciphertext = encryptor.update(data) + encryptor.finalize()
        
        return {
            'ciphertext': base64.b64encode(ciphertext).decode(),
            'iv': base64.b64encode(iv).decode(),
            'tag': base64.b64encode(encryptor.tag).decode(),
            'algorithm': self.config.symmetric_algorithm,
            'timestamp': int(time.time())
        }
    
    def decrypt_symmetric(self, encrypted_data: Dict[str, str], key: Union[str, bytes],
                         additional_data: bytes = None) -> bytes:
        """Decrypt data using symmetric encryption"""
        if isinstance(key, str):
            key = base64.urlsafe_b64decode(key.encode())
        
        ciphertext = base64.b64decode(encrypted_data['ciphertext'])
        iv = base64.b64decode(encrypted_data['iv'])
        tag = base64.b64decode(encrypted_data['tag'])
        
        # Create cipher
        cipher = Cipher(
            algorithms.AES(key),
            modes.GCM(iv, tag),
            backend=self.backend
        )
        
        decryptor = cipher.decryptor()
        
        # Add additional authenticated data if provided
        if additional_data:
            decryptor.authenticate_additional_data(additional_data)
        
        # Decrypt and finalize
        plaintext = decryptor.update(ciphertext) + decryptor.finalize()
        
        return plaintext
    
    # Asymmetric Encryption
    def generate_rsa_keypair(self, key_id: str = None) -> Tuple[str, str]:
        """Generate RSA public/private key pair"""
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=self.config.asymmetric_key_size,
            backend=self.backend
        )
        
        public_key = private_key.public_key()
        
        # Serialize keys
        private_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        )
        
        public_pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
        
        if key_id:
            self._key_metadata[key_id] = {
                'created_at': time.time(),
                'key_type': 'asymmetric',
                'algorithm': 'RSA',
                'key_size': self.config.asymmetric_key_size,
                'usage_count': 0
            }
        
        return private_pem.decode(), public_pem.decode()
    
    def encrypt_asymmetric(self, data: Union[str, bytes], public_key_pem: str) -> str:
        """Encrypt data using RSA public key"""
        if isinstance(data, str):
            data = data.encode('utf-8')
        
        public_key = serialization.load_pem_public_key(
            public_key_pem.encode(),
            backend=self.backend
        )
        
        ciphertext = public_key.encrypt(
            data,
            asym_padding.OAEP(
                mgf=asym_padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )
        
        return base64.b64encode(ciphertext).decode()
    
    def decrypt_asymmetric(self, ciphertext: str, private_key_pem: str) -> bytes:
        """Decrypt data using RSA private key"""
        private_key = serialization.load_pem_private_key(
            private_key_pem.encode(),
            password=None,
            backend=self.backend
        )
        
        ciphertext_bytes = base64.b64decode(ciphertext)
        
        plaintext = private_key.decrypt(
            ciphertext_bytes,
            asym_padding.OAEP(
                mgf=asym_padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )
        
        return plaintext
    
    # Digital Signatures
    def sign_data(self, data: Union[str, bytes], private_key_pem: str) -> str:
        """Create digital signature for data"""
        if isinstance(data, str):
            data = data.encode('utf-8')
        
        private_key = serialization.load_pem_private_key(
            private_key_pem.encode(),
            password=None,
            backend=self.backend
        )
        
        signature = private_key.sign(
            data,
            asym_padding.PSS(
                mgf=asym_padding.MGF1(hashes.SHA256()),
                salt_length=asym_padding.PSS.MAX_LENGTH
            ),
            hashes.SHA256()
        )
        
        return base64.b64encode(signature).decode()
    
    def verify_signature(self, data: Union[str, bytes], signature: str, public_key_pem: str) -> bool:
        """Verify digital signature"""
        if isinstance(data, str):
            data = data.encode('utf-8')
        
        public_key = serialization.load_pem_public_key(
            public_key_pem.encode(),
            backend=self.backend
        )
        
        signature_bytes = base64.b64decode(signature)
        
        try:
            public_key.verify(
                signature_bytes,
                data,
                asym_padding.PSS(
                    mgf=asym_padding.MGF1(hashes.SHA256()),
                    salt_length=asym_padding.PSS.MAX_LENGTH
                ),
                hashes.SHA256()
            )
            return True
        except Exception:
            return False
    
    # Password Hashing and Key Derivation
    def hash_password(self, password: str, salt: bytes = None) -> Dict[str, str]:
        """Secure password hashing using Scrypt"""
        if salt is None:
            salt = os.urandom(self.config.salt_length)
        
        kdf = Scrypt(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            n=2**14,  # CPU/memory cost
            r=8,      # Block size
            p=1,      # Parallelization
            backend=self.backend
        )
        
        key = kdf.derive(password.encode('utf-8'))
        
        return {
            'hash': base64.b64encode(key).decode(),
            'salt': base64.b64encode(salt).decode(),
            'algorithm': 'scrypt',
            'params': {'n': 2**14, 'r': 8, 'p': 1}
        }
    
    def verify_password(self, password: str, stored_hash: Dict[str, str]) -> bool:
        """Verify password against stored hash"""
        salt = base64.b64decode(stored_hash['salt'])
        stored_key = base64.b64decode(stored_hash['hash'])
        
        kdf = Scrypt(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            n=stored_hash['params']['n'],
            r=stored_hash['params']['r'],
            p=stored_hash['params']['p'],
            backend=self.backend
        )
        
        try:
            kdf.verify(password.encode('utf-8'), stored_key)
            return True
        except Exception:
            return False
    
    def derive_key_from_password(self, password: str, salt: bytes = None, 
                                info: bytes = None) -> Tuple[bytes, bytes]:
        """Derive encryption key from password using PBKDF2"""
        if salt is None:
            salt = os.urandom(self.config.salt_length)
        
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=self.config.kdf_iterations,
            backend=self.backend
        )
        
        key = kdf.derive(password.encode('utf-8'))
        
        return key, salt
    
    # Secure Random Generation
    def generate_secure_token(self, length: int = 32) -> str:
        """Generate cryptographically secure random token"""
        return secrets.token_urlsafe(length)
    
    def generate_secure_bytes(self, length: int = 32) -> bytes:
        """Generate cryptographically secure random bytes"""
        return secrets.token_bytes(length)
    
    # Message Authentication Codes
    def generate_hmac(self, message: Union[str, bytes], key: bytes, 
                     algorithm: str = 'SHA256') -> str:
        """Generate HMAC for message authentication"""
        if isinstance(message, str):
            message = message.encode('utf-8')
        
        hash_algorithm = getattr(hashlib, algorithm.lower())
        mac = hmac.new(key, message, hash_algorithm)
        
        return base64.b64encode(mac.digest()).decode()
    
    def verify_hmac(self, message: Union[str, bytes], expected_hmac: str, 
                   key: bytes, algorithm: str = 'SHA256') -> bool:
        """Verify HMAC for message authentication"""
        calculated_hmac = self.generate_hmac(message, key, algorithm)
        return hmac.compare_digest(calculated_hmac, expected_hmac)
    
    # Key Management
    def rotate_key(self, key_id: str) -> Optional[str]:
        """Rotate encryption key"""
        if key_id not in self._key_metadata:
            return None
        
        metadata = self._key_metadata[key_id]
        key_age_days = (time.time() - metadata['created_at']) / (24 * 3600)
        
        if key_age_days >= self.config.key_rotation_days:
            # Generate new key
            new_key = self.generate_symmetric_key(f"{key_id}_rotated")
            
            # Mark old key as deprecated
            metadata['status'] = 'deprecated'
            metadata['rotated_at'] = time.time()
            
            logger.info(f"Key {key_id} rotated after {key_age_days:.1f} days")
            
            return new_key
        
        return None
    
    def get_key_metadata(self, key_id: str) -> Optional[Dict[str, Any]]:
        """Get metadata for a key"""
        return self._key_metadata.get(key_id)

class SecureHasher:
    """Secure hashing utilities"""
    
    @staticmethod
    def hash_data(data: Union[str, bytes], algorithm: str = 'SHA256', 
                  salt: bytes = None) -> Dict[str, str]:
        """Secure hash with optional salt"""
        if isinstance(data, str):
            data = data.encode('utf-8')
        
        if salt is None:
            salt = os.urandom(32)
        
        hasher = hashlib.new(algorithm.lower())
        hasher.update(salt + data)
        
        return {
            'hash': hasher.hexdigest(),
            'salt': base64.b64encode(salt).decode(),
            'algorithm': algorithm
        }
    
    @staticmethod
    def verify_hash(data: Union[str, bytes], stored_hash: Dict[str, str]) -> bool:
        """Verify data against stored hash"""
        if isinstance(data, str):
            data = data.encode('utf-8')
        
        salt = base64.b64decode(stored_hash['salt'])
        algorithm = stored_hash['algorithm']
        
        hasher = hashlib.new(algorithm.lower())
        hasher.update(salt + data)
        calculated_hash = hasher.hexdigest()
        
        return hmac.compare_digest(calculated_hash, stored_hash['hash'])
    
    @staticmethod
    def hash_file(file_path: Path, algorithm: str = 'SHA256', 
                  chunk_size: int = 65536) -> str:
        """Calculate hash of file contents"""
        hasher = hashlib.new(algorithm.lower())
        
        with open(file_path, 'rb') as f:
            while chunk := f.read(chunk_size):
                hasher.update(chunk)
        
        return hasher.hexdigest()

class CertificateManager:
    """X.509 certificate management utilities"""
    
    @staticmethod
    def load_certificate(cert_path: Path) -> Any:
        """Load X.509 certificate from file"""
        with open(cert_path, 'rb') as f:
            cert_data = f.read()
        
        return load_pem_x509_certificate(cert_data, default_backend())
    
    @staticmethod
    def verify_certificate_chain(cert_chain: List[Any]) -> bool:
        """Verify certificate chain validity"""
        # Simplified certificate chain verification
        # In production, use comprehensive verification
        try:
            for i in range(len(cert_chain) - 1):
                current_cert = cert_chain[i]
                issuer_cert = cert_chain[i + 1]
                
                # Verify signature
                issuer_public_key = issuer_cert.public_key()
                issuer_public_key.verify(
                    current_cert.signature,
                    current_cert.tbs_certificate_bytes,
                    asym_padding.PKCS1v15(),
                    current_cert.signature_hash_algorithm
                )
            
            return True
        except Exception:
            return False

2. Authentication & Authorization System

security/auth_manager.py:

import jwt
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional, Any, Union, Set
from dataclasses import dataclass, field
from enum import Enum
import secrets
import hashlib
import time
import threading
from contextlib import contextmanager
import logging
from functools import wraps
import bcrypt
import redis
from fastapi import HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import json

logger = logging.getLogger(__name__)

class UserRole(Enum):
    """User roles enumeration"""
    ADMIN = "admin"
    MODERATOR = "moderator"
    USER = "user"
    GUEST = "guest"

class Permission(Enum):
    """System permissions enumeration"""
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    ADMIN = "admin"
    MODERATE = "moderate"

@dataclass
class User:
    """User data model"""
    id: str
    username: str
    email: str
    password_hash: str
    roles: List[UserRole] = field(default_factory=list)
    permissions: Set[Permission] = field(default_factory=set)
    is_active: bool = True
    is_verified: bool = False
    created_at: datetime = field(default_factory=datetime.utcnow)
    last_login: Optional[datetime] = None
    failed_login_attempts: int = 0
    account_locked_until: Optional[datetime] = None
    mfa_enabled: bool = False
    mfa_secret: Optional[str] = None

@dataclass
class AuthConfig:
    """Authentication configuration"""
    jwt_secret_key: str
    jwt_algorithm: str = "HS256"
    access_token_expire_minutes: int = 15
    refresh_token_expire_days: int = 30
    max_login_attempts: int = 5
    lockout_duration_minutes: int = 30
    password_min_length: int = 8
    password_require_special: bool = True
    password_require_number: bool = True
    password_require_uppercase: bool = True
    session_timeout_minutes: int = 60
    enforce_mfa: bool = False
    allowed_origins: List[str] = field(default_factory=list)

class PasswordValidator:
    """Password strength validation"""
    
    def __init__(self, config: AuthConfig):
        self.config = config
    
    def validate_password(self, password: str, username: str = None) -> Dict[str, Any]:
        """Validate password strength"""
        errors = []
        score = 0
        
        # Length check
        if len(password) < self.config.password_min_length:
            errors.append(f"Password must be at least {self.config.password_min_length} characters")
        else:
            score += 1
        
        # Character requirements
        if self.config.password_require_uppercase and not any(c.isupper() for c in password):
            errors.append("Password must contain at least one uppercase letter")
        else:
            score += 1
        
        if self.config.password_require_number and not any(c.isdigit() for c in password):
            errors.append("Password must contain at least one number")
        else:
            score += 1
        
        if self.config.password_require_special:
            special_chars = "!@#$%^&*()_+-=[]{}|;:,.<>?"
            if not any(c in special_chars for c in password):
                errors.append("Password must contain at least one special character")
            else:
                score += 1
        
        # Username similarity check
        if username and username.lower() in password.lower():
            errors.append("Password cannot contain username")
        
        # Common password check (simplified)
        common_passwords = ["password", "123456", "qwerty", "admin", "welcome"]
        if password.lower() in common_passwords:
            errors.append("Password is too common")
        
        # Calculate strength score (0-100)
        strength_score = min(100, (score / 4) * 100)
        
        # Additional scoring for length and complexity
        if len(password) > 12:
            strength_score += 10
        if len(set(password)) > len(password) * 0.7:  # Character diversity
            strength_score += 10
        
        return {
            'valid': len(errors) == 0,
            'errors': errors,
            'strength_score': min(100, strength_score),
            'strength_level': self._get_strength_level(strength_score)
        }
    
    def _get_strength_level(self, score: int) -> str:
        """Get password strength level"""
        if score >= 80:
            return "Strong"
        elif score >= 60:
            return "Good"
        elif score >= 40:
            return "Fair"
        else:
            return "Weak"

class SessionManager:
    """Secure session management"""
    
    def __init__(self, redis_client=None):
        self.redis_client = redis_client
        self.sessions: Dict[str, Dict[str, Any]] = {}
        self.lock = threading.Lock()
    
    def create_session(self, user_id: str, session_data: Dict[str, Any], 
                      timeout_minutes: int = 60) -> str:
        """Create new session"""
        session_id = secrets.token_urlsafe(32)
        
        session_info = {
            'user_id': user_id,
            'created_at': time.time(),
            'last_activity': time.time(),
            'expires_at': time.time() + (timeout_minutes * 60),
            'data': session_data,
            'ip_address': session_data.get('ip_address'),
            'user_agent': session_data.get('user_agent')
        }
        
        if self.redis_client:
            # Store in Redis with TTL
            self.redis_client.setex(
                f"session:{session_id}",
                timeout_minutes * 60,
                json.dumps(session_info, default=str)
            )
        else:
            # Store in memory
            with self.lock:
                self.sessions[session_id] = session_info
        
        logger.info(f"Session created for user {user_id}: {session_id[:8]}...")
        return session_id
    
    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        """Get session information"""
        if self.redis_client:
            session_data = self.redis_client.get(f"session:{session_id}")
            if session_data:
                return json.loads(session_data)
        else:
            with self.lock:
                session = self.sessions.get(session_id)
                if session and session['expires_at'] > time.time():
                    # Update last activity
                    session['last_activity'] = time.time()
                    return session
                elif session:
                    # Session expired, remove it
                    del self.sessions[session_id]
        
        return None
    
    def update_session_activity(self, session_id: str):
        """Update session last activity timestamp"""
        if self.redis_client:
            session_data = self.redis_client.get(f"session:{session_id}")
            if session_data:
                session = json.loads(session_data)
                session['last_activity'] = time.time()
                
                # Get original TTL and reset it
                ttl = self.redis_client.ttl(f"session:{session_id}")
                if ttl > 0:
                    self.redis_client.setex(
                        f"session:{session_id}",
                        ttl,
                        json.dumps(session, default=str)
                    )
        else:
            with self.lock:
                if session_id in self.sessions:
                    self.sessions[session_id]['last_activity'] = time.time()
    
    def invalidate_session(self, session_id: str) -> bool:
        """Invalidate session"""
        if self.redis_client:
            result = self.redis_client.delete(f"session:{session_id}")
            return result > 0
        else:
            with self.lock:
                return self.sessions.pop(session_id, None) is not None
    
    def invalidate_user_sessions(self, user_id: str):
        """Invalidate all sessions for a user"""
        if self.redis_client:
            # Scan for user sessions and delete them
            for key in self.redis_client.scan_iter(match="session:*"):
                session_data = self.redis_client.get(key)
                if session_data:
                    session = json.loads(session_data)
                    if session.get('user_id') == user_id:
                        self.redis_client.delete(key)
        else:
            with self.lock:
                sessions_to_remove = [
                    sid for sid, session in self.sessions.items()
                    if session.get('user_id') == user_id
                ]
                for sid in sessions_to_remove:
                    del self.sessions[sid]

class JWTManager:
    """JWT token management"""
    
    def __init__(self, config: AuthConfig):
        self.config = config
        self.blacklist: Set[str] = set()
        self.lock = threading.Lock()
    
    def create_access_token(self, user: User, additional_claims: Dict[str, Any] = None) -> str:
        """Create JWT access token"""
        now = datetime.now(timezone.utc)
        expires = now + timedelta(minutes=self.config.access_token_expire_minutes)
        
        payload = {
            'sub': user.id,
            'username': user.username,
            'email': user.email,
            'roles': [role.value for role in user.roles],
            'permissions': list(user.permissions),
            'iat': now,
            'exp': expires,
            'type': 'access'
        }
        
        if additional_claims:
            payload.update(additional_claims)
        
        token = jwt.encode(payload, self.config.jwt_secret_key, algorithm=self.config.jwt_algorithm)
        return token
    
    def create_refresh_token(self, user: User) -> str:
        """Create JWT refresh token"""
        now = datetime.now(timezone.utc)
        expires = now + timedelta(days=self.config.refresh_token_expire_days)
        
        payload = {
            'sub': user.id,
            'iat': now,
            'exp': expires,
            'type': 'refresh'
        }
        
        token = jwt.encode(payload, self.config.jwt_secret_key, algorithm=self.config.jwt_algorithm)
        return token
    
    def verify_token(self, token: str) -> Dict[str, Any]:
        """Verify and decode JWT token"""
        try:
            # Check if token is blacklisted
            token_hash = hashlib.sha256(token.encode()).hexdigest()
            with self.lock:
                if token_hash in self.blacklist:
                    raise HTTPException(
                        status_code=status.HTTP_401_UNAUTHORIZED,
                        detail="Token has been revoked"
                    )
            
            payload = jwt.decode(
                token,
                self.config.jwt_secret_key,
                algorithms=[self.config.jwt_algorithm]
            )
            
            return payload
            
        except jwt.ExpiredSignatureError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token has expired"
            )
        except jwt.InvalidTokenError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token"
            )
    
    def revoke_token(self, token: str):
        """Revoke token by adding to blacklist"""
        token_hash = hashlib.sha256(token.encode()).hexdigest()
        with self.lock:
            self.blacklist.add(token_hash)
        
        logger.info(f"Token revoked: {token_hash[:16]}...")
    
    def refresh_access_token(self, refresh_token: str, user: User) -> str:
        """Create new access token using refresh token"""
        payload = self.verify_token(refresh_token)
        
        if payload.get('type') != 'refresh':
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token type"
            )
        
        if payload.get('sub') != user.id:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token user mismatch"
            )
        
        return self.create_access_token(user)

class AuthenticationManager:
    """Main authentication manager"""
    
    def __init__(self, config: AuthConfig, redis_client=None):
        self.config = config
        self.jwt_manager = JWTManager(config)
        self.session_manager = SessionManager(redis_client)
        self.password_validator = PasswordValidator(config)
        self.users: Dict[str, User] = {}  # In production, use database
        self.lock = threading.Lock()
    
    def hash_password(self, password: str) -> str:
        """Hash password using bcrypt"""
        salt = bcrypt.gensalt()
        hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
        return hashed.decode('utf-8')
    
    def verify_password(self, password: str, hashed_password: str) -> bool:
        """Verify password against hash"""
        return bcrypt.checkpw(password.encode('utf-8'), hashed_password.encode('utf-8'))
    
    def register_user(self, username: str, email: str, password: str, 
                     roles: List[UserRole] = None) -> Dict[str, Any]:
        """Register new user"""
        # Validate password
        validation = self.password_validator.validate_password(password, username)
        if not validation['valid']:
            return {
                'success': False,
                'errors': validation['errors']
            }
        
        # Check if user exists
        with self.lock:
            if any(u.username == username or u.email == email for u in self.users.values()):
                return {
                    'success': False,
                    'errors': ['Username or email already exists']
                }
        
        # Create user
        user_id = secrets.token_urlsafe(16)
        password_hash = self.hash_password(password)
        
        user = User(
            id=user_id,
            username=username,
            email=email,
            password_hash=password_hash,
            roles=roles or [UserRole.USER]
        )
        
        with self.lock:
            self.users[user_id] = user
        
        logger.info(f"User registered: {username} ({email})")
        
        return {
            'success': True,
            'user_id': user_id,
            'password_strength': validation['strength_level']
        }
    
    def authenticate_user(self, username: str, password: str, 
                         ip_address: str = None, user_agent: str = None) -> Dict[str, Any]:
        """Authenticate user credentials"""
        
        # Find user
        user = None
        with self.lock:
            for u in self.users.values():
                if u.username == username or u.email == username:
                    user = u
                    break
        
        if not user:
            logger.warning(f"Authentication failed: user not found - {username}")
            return {'success': False, 'error': 'Invalid credentials'}
        
        # Check if account is locked
        if user.account_locked_until and user.account_locked_until > datetime.utcnow():
            logger.warning(f"Authentication failed: account locked - {username}")
            return {
                'success': False,
                'error': 'Account temporarily locked due to failed login attempts'
            }
        
        # Verify password
        if not self.verify_password(password, user.password_hash):
            # Increment failed attempts
            user.failed_login_attempts += 1
            
            if user.failed_login_attempts >= self.config.max_login_attempts:
                # Lock account
                user.account_locked_until = datetime.utcnow() + timedelta(
                    minutes=self.config.lockout_duration_minutes
                )
                logger.warning(f"Account locked due to failed attempts: {username}")
            
            logger.warning(f"Authentication failed: invalid password - {username}")
            return {'success': False, 'error': 'Invalid credentials'}
        
        # Check if account is active and verified
        if not user.is_active:
            return {'success': False, 'error': 'Account is disabled'}
        
        if not user.is_verified:
            return {'success': False, 'error': 'Account is not verified'}
        
        # Reset failed attempts on successful login
        user.failed_login_attempts = 0
        user.account_locked_until = None
        user.last_login = datetime.utcnow()
        
        # Create tokens
        access_token = self.jwt_manager.create_access_token(user)
        refresh_token = self.jwt_manager.create_refresh_token(user)
        
        # Create session
        session_data = {
            'ip_address': ip_address,
            'user_agent': user_agent
        }
        session_id = self.session_manager.create_session(
            user.id, session_data, self.config.session_timeout_minutes
        )
        
        logger.info(f"User authenticated successfully: {username}")
        
        return {
            'success': True,
            'access_token': access_token,
            'refresh_token': refresh_token,
            'session_id': session_id,
            'user': {
                'id': user.id,
                'username': user.username,
                'email': user.email,
                'roles': [role.value for role in user.roles]
            }
        }
    
    def logout_user(self, access_token: str, session_id: str = None):
        """Logout user and invalidate tokens/session"""
        # Revoke access token
        self.jwt_manager.revoke_token(access_token)
        
        # Invalidate session
        if session_id:
            self.session_manager.invalidate_session(session_id)
        
        logger.info("User logged out successfully")
    
    def get_current_user(self, token: str) -> User:
        """Get current user from token"""
        payload = self.jwt_manager.verify_token(token)
        user_id = payload.get('sub')
        
        with self.lock:
            user = self.users.get(user_id)
            if not user or not user.is_active:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="User not found or inactive"
                )
        
        return user

# Security decorators and middleware
class SecurityDecorators:
    """Security-related decorators"""
    
    @staticmethod
    def require_auth(auth_manager: AuthenticationManager):
        """Decorator to require authentication"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # Extract token from request headers (implementation depends on framework)
                token = kwargs.get('token') or args[0].headers.get('Authorization', '').replace('Bearer ', '')
                
                if not token:
                    raise HTTPException(
                        status_code=status.HTTP_401_UNAUTHORIZED,
                        detail="Authentication required"
                    )
                
                try:
                    user = auth_manager.get_current_user(token)
                    kwargs['current_user'] = user
                    return func(*args, **kwargs)
                except Exception as e:
                    raise HTTPException(
                        status_code=status.HTTP_401_UNAUTHORIZED,
                        detail=str(e)
                    )
            
            return wrapper
        return decorator
    
    @staticmethod
    def require_roles(*required_roles: UserRole):
        """Decorator to require specific roles"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                user = kwargs.get('current_user')
                if not user:
                    raise HTTPException(
                        status_code=status.HTTP_401_UNAUTHORIZED,
                        detail="Authentication required"
                    )
                
                if not any(role in user.roles for role in required_roles):
                    raise HTTPException(
                        status_code=status.HTTP_403_FORBIDDEN,
                        detail="Insufficient permissions"
                    )
                
                return func(*args, **kwargs)
            
            return wrapper
        return decorator
    
    @staticmethod
    def require_permissions(*required_permissions: Permission):
        """Decorator to require specific permissions"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                user = kwargs.get('current_user')
                if not user:
                    raise HTTPException(
                        status_code=status.HTTP_401_UNAUTHORIZED,
                        detail="Authentication required"
                    )
                
                if not all(perm in user.permissions for perm in required_permissions):
                    raise HTTPException(
                        status_code=status.HTTP_403_FORBIDDEN,
                        detail="Insufficient permissions"
                    )
                
                return func(*args, **kwargs)
            
            return wrapper
        return decorator

# Rate limiting for security
class RateLimiter:
    """Rate limiting for security purposes"""
    
    def __init__(self, redis_client=None):
        self.redis_client = redis_client
        self.counters: Dict[str, Dict[str, Any]] = {}
        self.lock = threading.Lock()
    
    def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
        """Check if request is allowed under rate limit"""
        now = time.time()
        
        if self.redis_client:
            # Use Redis for distributed rate limiting
            pipe = self.redis_client.pipeline()
            pipe.zremrangebyscore(key, 0, now - window_seconds)
            pipe.zcard(key)
            pipe.zadd(key, {str(now): now})
            pipe.expire(key, window_seconds)
            results = pipe.execute()
            
            current_requests = results[1]
            return current_requests < max_requests
        else:
            # Use in-memory rate limiting
            with self.lock:
                if key not in self.counters:
                    self.counters[key] = {'requests': [], 'window_start': now}
                
                counter = self.counters[key]
                
                # Remove old requests outside the window
                counter['requests'] = [req_time for req_time in counter['requests'] 
                                     if req_time > now - window_seconds]
                
                # Check if under limit
                if len(counter['requests']) < max_requests:
                    counter['requests'].append(now)
                    return True
                
                return False

3. Web Application Security Framework

security/web_security.py:

import re
import html
import urllib.parse
from typing import Dict, List, Any, Optional, Union, Set
from dataclasses import dataclass, field
import hashlib
import secrets
import time
import logging
from functools import wraps
from urllib.parse import urlparse
import ipaddress
import json

logger = logging.getLogger(__name__)

@dataclass
class SecurityConfig:
    """Web application security configuration"""
    # Content Security Policy
    csp_default_src: List[str] = field(default_factory=lambda: ["'self'"])
    csp_script_src: List[str] = field(default_factory=lambda: ["'self'"])
    csp_style_src: List[str] = field(default_factory=lambda: ["'self'", "'unsafe-inline'"])
    csp_img_src: List[str] = field(default_factory=lambda: ["'self'", "data:", "https:"])
    
    # CSRF Protection
    csrf_token_length: int = 32
    csrf_cookie_name: str = "csrf_token"
    csrf_header_name: str = "X-CSRF-Token"
    
    # XSS Protection
    xss_protection_enabled: bool = True
    content_type_nosniff: bool = True
    frame_options: str = "DENY"  # DENY, SAMEORIGIN, ALLOW-FROM
    
    # HTTPS/Security Headers
    hsts_max_age: int = 31536000  # 1 year
    hsts_include_subdomains: bool = True
    hsts_preload: bool = True
    
    # Input Validation
    max_file_size: int = 10 * 1024 * 1024  # 10MB
    allowed_file_extensions: Set[str] = field(
        default_factory=lambda: {'.jpg', '.jpeg', '.png', '.gif', '.pdf', '.txt', '.docx'}
    )
    
    # Rate Limiting
    default_rate_limit: int = 1000  # requests per hour
    auth_rate_limit: int = 5  # login attempts per minute
    
    # IP Whitelist/Blacklist
    ip_whitelist: List[str] = field(default_factory=list)
    ip_blacklist: List[str] = field(default_factory=list)

class InputValidator:
    """Comprehensive input validation and sanitization"""
    
    def __init__(self, config: SecurityConfig):
        self.config = config
        
        # Common regex patterns
        self.patterns = {
            'email': re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'),
            'phone': re.compile(r'^\+?1?[-.\s]?\(?[0-9]{3}\)?[-.\s]?[0-9]{3}[-.\s]?[0-9]{4}$'),
            'url': re.compile(r'^https?://[^\s/$.?#].[^\s]*$', re.IGNORECASE),
            'sql_injection': re.compile(
                r'(union|select|insert|update|delete|drop|create|alter|exec|execute|script)'
                r'|(\bor\b|\band\b).*[=<>]|[\'";]', 
                re.IGNORECASE
            ),
            'xss_basic': re.compile(
                r'<script|</script|javascript:|on\w+\s*=|<iframe|</iframe|eval\(|alert\(',
                re.IGNORECASE
            ),
            'path_traversal': re.compile(r'\.\.[\\/]|\.\.%2f|%2e%2e', re.IGNORECASE)
        }
    
    def validate_email(self, email: str) -> Dict[str, Any]:
        """Validate email address"""
        if not isinstance(email, str):
            return {'valid': False, 'error': 'Email must be a string'}
        
        email = email.strip().lower()
        
        if not email:
            return {'valid': False, 'error': 'Email is required'}
        
        if len(email) > 254:
            return {'valid': False, 'error': 'Email is too long'}
        
        if not self.patterns['email'].match(email):
            return {'valid': False, 'error': 'Invalid email format'}
        
        return {'valid': True, 'sanitized': email}
    
    def validate_url(self, url: str, allowed_schemes: List[str] = None) -> Dict[str, Any]:
        """Validate URL"""
        if not isinstance(url, str):
            return {'valid': False, 'error': 'URL must be a string'}
        
        url = url.strip()
        
        if not url:
            return {'valid': False, 'error': 'URL is required'}
        
        if len(url) > 2048:
            return {'valid': False, 'error': 'URL is too long'}
        
        try:
            parsed = urlparse(url)
        except Exception:
            return {'valid': False, 'error': 'Invalid URL format'}
        
        # Check scheme
        allowed_schemes = allowed_schemes or ['http', 'https']
        if parsed.scheme.lower() not in allowed_schemes:
            return {'valid': False, 'error': f'URL scheme must be one of: {allowed_schemes}'}
        
        # Check for suspicious patterns
        if self.detect_xss(url)['detected']:
            return {'valid': False, 'error': 'URL contains suspicious content'}
        
        return {'valid': True, 'sanitized': url}
    
    def sanitize_html(self, text: str, allowed_tags: Set[str] = None) -> str:
        """Sanitize HTML content"""
        if not isinstance(text, str):
            return str(text)
        
        allowed_tags = allowed_tags or {'b', 'i', 'u', 'em', 'strong', 'p', 'br'}
        
        # First escape all HTML
        sanitized = html.escape(text)
        
        # If allowed tags specified, selectively unescape them
        if allowed_tags:
            for tag in allowed_tags:
                # Simple tag replacement (for production, use a proper HTML sanitizer)
                sanitized = sanitized.replace(f'&lt;{tag}&gt;', f'<{tag}>')
                sanitized = sanitized.replace(f'&lt;/{tag}&gt;', f'</{tag}>')
        
        return sanitized
    
    def detect_sql_injection(self, text: str) -> Dict[str, Any]:
        """Detect potential SQL injection attempts"""
        if not isinstance(text, str):
            return {'detected': False}
        
        text_lower = text.lower()
        
        # Check for SQL injection patterns
        if self.patterns['sql_injection'].search(text):
            return {
                'detected': True,
                'type': 'sql_injection',
                'severity': 'high',
                'pattern': 'SQL keywords detected'
            }
        
        # Check for suspicious character combinations
        suspicious_patterns = [
            "' or '1'='1",
            "' or 1=1--",
            "' union select",
            "; drop table",
            "/**/",
            "@@version"
        ]
        
        for pattern in suspicious_patterns:
            if pattern in text_lower:
                return {
                    'detected': True,
                    'type': 'sql_injection',
                    'severity': 'high',
                    'pattern': f'Suspicious pattern: {pattern}'
                }
        
        return {'detected': False}
    
    def detect_xss(self, text: str) -> Dict[str, Any]:
        """Detect potential XSS attempts"""
        if not isinstance(text, str):
            return {'detected': False}
        
        text_lower = text.lower()
        
        # Check for basic XSS patterns
        if self.patterns['xss_basic'].search(text):
            return {
                'detected': True,
                'type': 'xss',
                'severity': 'high',
                'pattern': 'Script injection detected'
            }
        
        # Check for encoded payloads
        try:
            decoded = urllib.parse.unquote(text)
            if decoded != text and self.patterns['xss_basic'].search(decoded.lower()):
                return {
                    'detected': True,
                    'type': 'xss',
                    'severity': 'high',
                    'pattern': 'Encoded script injection detected'
                }
        except Exception:
            pass
        
        # Check for suspicious patterns
        suspicious_patterns = [
            'javascript:',
            'vbscript:',
            'onload=',
            'onerror=',
            'onclick=',
            'onfocus=',
            'onmouseover=',
            'expression(',
            'url(javascript:',
            'data:text/html'
        ]
        
        for pattern in suspicious_patterns:
            if pattern in text_lower:
                return {
                    'detected': True,
                    'type': 'xss',
                    'severity': 'medium',
                    'pattern': f'Suspicious pattern: {pattern}'
                }
        
        return {'detected': False}
    
    def detect_path_traversal(self, path: str) -> Dict[str, Any]:
        """Detect path traversal attempts"""
        if not isinstance(path, str):
            return {'detected': False}
        
        if self.patterns['path_traversal'].search(path):
            return {
                'detected': True,
                'type': 'path_traversal',
                'severity': 'high',
                'pattern': 'Path traversal sequence detected'
            }
        
        # Check for null bytes and other suspicious characters
        if '\x00' in path or '%00' in path:
            return {
                'detected': True,
                'type': 'path_traversal',
                'severity': 'high',
                'pattern': 'Null byte detected'
            }
        
        return {'detected': False}
    
    def validate_file_upload(self, filename: str, content: bytes) -> Dict[str, Any]:
        """Validate file upload"""
        if not filename:
            return {'valid': False, 'error': 'Filename is required'}
        
        # Check file extension
        file_ext = '.' + filename.split('.')[-1].lower() if '.' in filename else ''
        if file_ext not in self.config.allowed_file_extensions:
            return {
                'valid': False,
                'error': f'File type not allowed. Allowed: {list(self.config.allowed_file_extensions)}'
            }
        
        # Check file size
        if len(content) > self.config.max_file_size:
            return {
                'valid': False,
                'error': f'File too large. Max size: {self.config.max_file_size} bytes'
            }
        
        # Check for malicious content in filename
        path_check = self.detect_path_traversal(filename)
        if path_check['detected']:
            return {'valid': False, 'error': 'Malicious filename detected'}
        
        # Basic file content validation
        if self._is_executable_content(content):
            return {'valid': False, 'error': 'Executable content detected'}
        
        return {'valid': True, 'sanitized_filename': self._sanitize_filename(filename)}
    
    def _sanitize_filename(self, filename: str) -> str:
        """Sanitize filename for safe storage"""
        # Remove directory separators and other dangerous characters
        sanitized = re.sub(r'[<>:"/\\|?*\x00-\x1f]', '_', filename)
        
        # Limit length
        if len(sanitized) > 255:
            name, ext = sanitized.rsplit('.', 1) if '.' in sanitized else (sanitized, '')
            max_name_len = 255 - len(ext) - 1 if ext else 255
            sanitized = name[:max_name_len] + ('.' + ext if ext else '')
        
        return sanitized
    
    def _is_executable_content(self, content: bytes) -> bool:
        """Check if content appears to be executable"""
        # Check for common executable file signatures
        executable_signatures = [
            b'\x4d\x5a',  # PE executable (Windows)
            b'\x7f\x45\x4c\x46',  # ELF executable (Linux)
            b'\xfe\xed\xfa',  # Mach-O executable (macOS)
            b'\xca\xfe\xba\xbe',  # Java class file
            b'#!/bin/',  # Shell script
            b'<?php',  # PHP script
            b'<script',  # JavaScript
        ]
        
        content_lower = content[:1024].lower()  # Check first 1KB
        
        for signature in executable_signatures:
            if signature in content_lower:
                return True
        
        return False

class CSRFProtection:
    """CSRF token generation and validation"""
    
    def __init__(self, config: SecurityConfig):
        self.config = config
        self.tokens: Dict[str, float] = {}  # token -> timestamp
    
    def generate_token(self, user_id: str = None) -> str:
        """Generate CSRF token"""
        token_data = f"{user_id or 'anonymous'}:{time.time()}:{secrets.token_urlsafe(16)}"
        token = hashlib.sha256(token_data.encode()).hexdigest()[:self.config.csrf_token_length]
        
        # Store token with timestamp
        self.tokens[token] = time.time()
        
        # Clean up old tokens (older than 1 hour)
        cutoff_time = time.time() - 3600
        self.tokens = {t: ts for t, ts in self.tokens.items() if ts > cutoff_time}
        
        return token
    
    def validate_token(self, token: str, max_age: int = 3600) -> bool:
        """Validate CSRF token"""
        if not token or token not in self.tokens:
            return False
        
        token_time = self.tokens.get(token)
        if not token_time:
            return False
        
        # Check if token is expired
        if time.time() - token_time > max_age:
            del self.tokens[token]
            return False
        
        return True
    
    def invalidate_token(self, token: str):
        """Invalidate CSRF token"""
        self.tokens.pop(token, None)

class SecurityHeaders:
    """Security HTTP headers management"""
    
    def __init__(self, config: SecurityConfig):
        self.config = config
    
    def get_security_headers(self) -> Dict[str, str]:
        """Get all security headers"""
        headers = {}
        
        # Content Security Policy
        csp_parts = []
        csp_parts.append(f"default-src {' '.join(self.config.csp_default_src)}")
        csp_parts.append(f"script-src {' '.join(self.config.csp_script_src)}")
        csp_parts.append(f"style-src {' '.join(self.config.csp_style_src)}")
        csp_parts.append(f"img-src {' '.join(self.config.csp_img_src)}")
        
        headers['Content-Security-Policy'] = '; '.join(csp_parts)
        
        # XSS Protection
        if self.config.xss_protection_enabled:
            headers['X-XSS-Protection'] = '1; mode=block'
        
        # Content Type Options
        if self.config.content_type_nosniff:
            headers['X-Content-Type-Options'] = 'nosniff'
        
        # Frame Options
        headers['X-Frame-Options'] = self.config.frame_options
        
        # HSTS
        hsts_value = f'max-age={self.config.hsts_max_age}'
        if self.config.hsts_include_subdomains:
            hsts_value += '; includeSubDomains'
        if self.config.hsts_preload:
            hsts_value += '; preload'
        
        headers['Strict-Transport-Security'] = hsts_value
        
        # Additional security headers
        headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
        headers['Permissions-Policy'] = 'geolocation=(), microphone=(), camera=()'
        
        return headers

class IPFilter:
    """IP address filtering and validation"""
    
    def __init__(self, config: SecurityConfig):
        self.config = config
        self.whitelist_networks = self._parse_ip_list(config.ip_whitelist)
        self.blacklist_networks = self._parse_ip_list(config.ip_blacklist)
    
    def _parse_ip_list(self, ip_list: List[str]) -> List[Union[ipaddress.IPv4Network, ipaddress.IPv6Network]]:
        """Parse IP addresses and networks"""
        networks = []
        for ip_str in ip_list:
            try:
                if '/' in ip_str:
                    # Network notation
                    networks.append(ipaddress.ip_network(ip_str, strict=False))
                else:
                    # Single IP
                    ip = ipaddress.ip_address(ip_str)
                    if isinstance(ip, ipaddress.IPv4Address):
                        networks.append(ipaddress.IPv4Network(f"{ip}/32"))
                    else:
                        networks.append(ipaddress.IPv6Network(f"{ip}/128"))
            except ValueError as e:
                logger.warning(f"Invalid IP address/network: {ip_str} - {e}")
        
        return networks
    
    def is_allowed(self, client_ip: str) -> Dict[str, Any]:
        """Check if IP address is allowed"""
        try:
            ip = ipaddress.ip_address(client_ip)
        except ValueError:
            return {'allowed': False, 'reason': 'Invalid IP address format'}
        
        # Check blacklist first
        for network in self.blacklist_networks:
            if ip in network:
                return {'allowed': False, 'reason': 'IP address is blacklisted'}
        
        # If whitelist is configured, check it
        if self.whitelist_networks:
            for network in self.whitelist_networks:
                if ip in network:
                    return {'allowed': True, 'reason': 'IP address is whitelisted'}
            
            # If whitelist exists but IP not in it, deny
            return {'allowed': False, 'reason': 'IP address not in whitelist'}
        
        # No whitelist configured, allow if not blacklisted
        return {'allowed': True, 'reason': 'IP address allowed'}

class SecurityAuditLogger:
    """Security event logging and monitoring"""
    
    def __init__(self):
        self.logger = logging.getLogger('security_audit')
        self.events: List[Dict[str, Any]] = []
    
    def log_security_event(self, event_type: str, details: Dict[str, Any], 
                          severity: str = 'info', user_id: str = None,
                          ip_address: str = None, user_agent: str = None):
        """Log security event"""
        event = {
            'timestamp': time.time(),
            'event_type': event_type,
            'severity': severity,
            'details': details,
            'user_id': user_id,
            'ip_address': ip_address,
            'user_agent': user_agent
        }
        
        self.events.append(event)
        
        # Log to standard logger
        log_message = f"Security Event [{event_type}]: {details}"
        if severity == 'critical':
            self.logger.critical(log_message)
        elif severity == 'high':
            self.logger.error(log_message)
        elif severity == 'medium':
            self.logger.warning(log_message)
        else:
            self.logger.info(log_message)
        
        # Keep only recent events (last 1000)
        if len(self.events) > 1000:
            self.events = self.events[-1000:]
    
    def get_security_events(self, event_type: str = None, 
                           severity: str = None, 
                           limit: int = 100) -> List[Dict[str, Any]]:
        """Get filtered security events"""
        filtered_events = self.events
        
        if event_type:
            filtered_events = [e for e in filtered_events if e['event_type'] == event_type]
        
        if severity:
            filtered_events = [e for e in filtered_events if e['severity'] == severity]
        
        # Return most recent events
        return sorted(filtered_events, key=lambda x: x['timestamp'], reverse=True)[:limit]

# Security middleware for web frameworks
def security_middleware(config: SecurityConfig):
    """Security middleware factory"""
    
    def middleware(request, response):
        # Add security headers
        security_headers = SecurityHeaders(config)
        headers = security_headers.get_security_headers()
        
        for name, value in headers.items():
            response.headers[name] = value
        
        return response
    
    return middleware

# Example usage and testing
def security_framework_example():
    """Example usage of the security framework"""
    
    # Configuration
    config = SecurityConfig()
    
    # Input validation
    validator = InputValidator(config)
    
    # Test email validation
    email_result = validator.validate_email("user@example.com")
    print(f"Email validation: {email_result}")
    
    # Test XSS detection
    xss_test = "<script>alert('xss')</script>"
    xss_result = validator.detect_xss(xss_test)
    print(f"XSS detection: {xss_result}")
    
    # Test SQL injection detection
    sql_test = "'; DROP TABLE users; --"
    sql_result = validator.detect_sql_injection(sql_test)
    print(f"SQL injection detection: {sql_result}")
    
    # CSRF protection
    csrf = CSRFProtection(config)
    token = csrf.generate_token("user123")
    is_valid = csrf.validate_token(token)
    print(f"CSRF token validation: {is_valid}")
    
    # IP filtering
    ip_filter = IPFilter(config)
    ip_result = ip_filter.is_allowed("192.168.1.1")
    print(f"IP filter result: {ip_result}")
    
    # Security audit logging
    audit_logger = SecurityAuditLogger()
    audit_logger.log_security_event(
        'login_attempt',
        {'username': 'testuser', 'success': True},
        'info',
        user_id='user123',
        ip_address='192.168.1.1'
    )
    
    events = audit_logger.get_security_events(limit=5)
    print(f"Recent security events: {len(events)}")
    
    return {
        'validator': validator,
        'csrf': csrf,
        'ip_filter': ip_filter,
        'audit_logger': audit_logger
    }

if __name__ == "__main__":
    security_framework_example()

Best Practices & Guidelines

1. Cryptographic Security

• Use established cryptographic libraries (cryptography, PyNaCl)
• Implement proper key management and rotation policies
• Use secure random generation for all cryptographic operations
• Apply defense in depth with multiple encryption layers

2. Authentication & Authorization

• Implement strong password policies and multi-factor authentication
• Use JWT tokens with proper expiration and refresh mechanisms
• Apply role-based access control (RBAC) with least privilege principle
• Monitor and log all authentication events

3. Input Validation & Sanitization

• Validate all input at application boundaries
• Use whitelist validation rather than blacklist filtering
• Sanitize output based on context (HTML, SQL, etc.)
• Implement comprehensive XSS and injection prevention

4. Web Application Security

• Apply security headers (CSP, HSTS, X-Frame-Options)
• Implement CSRF protection for all state-changing operations
• Use HTTPS everywhere with proper certificate validation
• Regular security testing and vulnerability assessments

5. Security Monitoring & Incident Response

• Implement comprehensive security logging and monitoring
• Set up automated alerting for security events
• Develop incident response procedures and playbooks
• Regular security training and awareness programs

This comprehensive security framework provides enterprise-grade security controls for Python applications, covering cryptography, authentication, web security, and compliance requirements.