# File: advanced_security_middleware.py # Enhanced security middleware for QR Attendance System from functools import wraps from flask import request, session, jsonify, current_app, g import hashlib import secrets import jwt from datetime import datetime, timedelta import re from collections import defaultdict, deque import time import hmac import base64 import os # Try to import cryptography, fallback if not available try: from cryptography.fernet import Fernet HAS_CRYPTOGRAPHY = True except ImportError: HAS_CRYPTOGRAPHY = False class SecurityManager: """ Advanced security manager for QR Attendance System """ def __init__(self, app=None, db=None, logger_handler=None): self.app = app self.db = db self.logger_handler = logger_handler # Security tracking self.failed_attempts = defaultdict(lambda: deque(maxlen=10)) self.suspicious_ips = defaultdict(int) self.session_tokens = {} # Security configuration self.max_failed_attempts = 5 # per IP + username, 15 minutes self.max_failed_attempts_per_username = 20 # per username from any IP, 15 minutes self.lockout_duration = 900 # 15 minutes self.session_timeout = 3600 # 1 hour if app: self.init_app(app, db, logger_handler) def init_app(self, app, db, logger_handler): """Initialize security manager with Flask app""" self.app = app self.db = db self.logger_handler = logger_handler # Generate encryption key for sensitive data self.setup_encryption() # Register security middleware app.before_request(self.security_check) # Register security routes self.register_security_routes() def setup_encryption(self): """Setup encryption for sensitive data. Derives a stable Fernet key from the app's SECRET_KEY so that all gunicorn workers share the same key without needing a separate ENCRYPTION_KEY env var. A random key is only generated as a last resort (dev mode without SECRET_KEY set). """ if HAS_CRYPTOGRAPHY: encryption_key = self.app.config.get('ENCRYPTION_KEY') if not encryption_key: # Derive a deterministic 32-byte key from SECRET_KEY so every # worker produces the same value — no per-worker randomness. secret = self.app.config.get('SECRET_KEY', '') derived = hashlib.sha256(secret.encode()).digest() encryption_key = base64.urlsafe_b64encode(derived) self.cipher = Fernet(encryption_key) else: self.cipher = None def security_check(self): """Comprehensive security check before each request""" client_ip = self.get_client_ip() # Check for suspicious activity if self.is_suspicious_request(): self.log_security_event('suspicious_request', { 'ip': client_ip, 'user_agent': request.headers.get('User-Agent', ''), 'endpoint': request.endpoint, 'method': request.method }) return jsonify({'error': 'Request blocked for security reasons'}), 403 # NOTE: per-worker in-memory session token validation removed. # Flask's cryptographically signed session cookie provides session # integrity; CSRF tokens handle cross-site forgery. Keeping the # validate_session_security() call here would log users out on every # gunicorn worker boundary because session_tokens is not shared. # Check for SQL injection attempts if self.detect_sql_injection(): self.log_security_event('sql_injection_attempt', { 'ip': client_ip, 'query_params': dict(request.args), 'form_data': dict(request.form) if request.form else {} }) return jsonify({'error': 'Malicious request detected'}), 403 # Rate limiting for authentication endpoints if request.endpoint in ['login', 'register', 'reset_password']: if self.is_auth_rate_limited(): return jsonify({ 'error': 'Too many attempts, please try again later' }), 429 def get_client_ip(self): """Get real client IP address""" # request.remote_addr is the real client address: app.py wraps the app in # ProxyFix(x_for=TRUSTED_PROXY_COUNT), which takes it from the entry our # own proxy (Nginx) appended to X-Forwarded-For. The FIRST entry of that # header — used here before — is whatever the client chose to send, so # changing it on every request bypassed the login rate limit. return request.remote_addr or 'unknown' def is_suspicious_request(self): """Detect suspicious request patterns""" client_ip = self.get_client_ip() user_agent = request.headers.get('User-Agent', '').lower() # Check for common attack patterns suspicious_patterns = [ r' 30: # More than 30 requests per minute return True return False def validate_session_security(self): """Validate session security and integrity""" try: user_id = session.get('user_id') session_token = session.get('security_token') if not user_id or not session_token: return False # Check if session token matches stored token stored_token = self.session_tokens.get(user_id) if not stored_token or not hmac.compare_digest(session_token, stored_token['token']): return False # Check session timeout - skip if "Remember Me" is enabled if not session.get('remember_me', False): if time.time() - stored_token['created'] > self.session_timeout: del self.session_tokens[user_id] return False # Check if session IP matches (optional security measure) if self.app.config.get('STRICT_SESSION_IP', False): if stored_token['ip'] != self.get_client_ip(): self.log_security_event('session_ip_mismatch', { 'user_id': user_id, 'original_ip': stored_token['ip'], 'current_ip': self.get_client_ip() }) return False return True except Exception as e: if self.logger_handler: self.logger_handler.logger.error(f"Session validation error: {e}") return False def detect_sql_injection(self): """Detect potential SQL injection attempts""" sql_patterns = [ r"union\s+select", r"drop\s+table", r"insert\s+into", r"delete\s+from", r"update\s+set", r"exec\s*\(", r"sp_executesql", r"xp_cmdshell", r";\s*--", r"'\s*or\s*'", r'"\s*or\s*"', r"1\s*=\s*1" ] # Check all request parameters check_data = [] check_data.extend(request.args.values()) check_data.extend(request.form.values()) # Only attempt JSON parsing when the client declared application/json. # Calling request.json without this guard raises a 415 Unsupported Media Type # on every non-JSON request (GET pages, form POSTs, favicon, etc.). if request.content_type and 'application/json' in request.content_type: try: json_body = request.get_json(silent=True, force=False) if json_body and isinstance(json_body, dict): check_data.extend( str(v) for v in json_body.values() if isinstance(v, (str, int, float)) ) except Exception: pass for data in check_data: data_str = str(data).lower() for pattern in sql_patterns: if re.search(pattern, data_str, re.IGNORECASE): return True return False def _login_attempt_keys(self, username=None): """Failed-login counter keys: this IP + username, and the username alone.""" client_ip = self.get_client_ip() name = (username or '').strip().lower() if not name: return [f"ip:{client_ip}"] return [f"ip:{client_ip}|user:{name}", f"user:{name}"] def is_auth_rate_limited(self, username=None): """ Should this login attempt be blocked? Two counters over lockout_duration (15 minutes): - this IP + username: blocked after max_failed_attempts (5) - this username from ANY IP: blocked after max_failed_attempts_per_username (20), so rotating IP addresses cannot brute-force one account Deliberately not a plain per-IP counter: users behind one shared address (office NAT, a proxy that hides client IPs) must not lock each other out. Counters are in memory per gunicorn worker. """ current_time = time.time() for key in self._login_attempt_keys(username): self.failed_attempts[key] = deque([ attempt for attempt in self.failed_attempts.get(key, ()) if current_time - attempt < self.lockout_duration ], maxlen=50) limit = (self.max_failed_attempts_per_username if key.startswith('user:') else self.max_failed_attempts) if len(self.failed_attempts[key]) >= limit: return True return False def record_failed_attempt(self, identifier): """Record a failed authentication attempt (identifier = the submitted username)""" client_ip = self.get_client_ip() current_time = time.time() keys = self._login_attempt_keys(identifier) for key in keys: if key not in self.failed_attempts: self.failed_attempts[key] = deque(maxlen=50) self.failed_attempts[key].append(current_time) self.log_security_event('authentication_failure', { 'ip': client_ip, 'identifier': identifier, 'attempts': len(self.failed_attempts[keys[0]]) }) def create_secure_session(self, user_id): """Create a secure session with additional security measures""" # Generate secure session token session_token = secrets.token_urlsafe(32) # Store session information self.session_tokens[user_id] = { 'token': session_token, 'created': time.time(), 'ip': self.get_client_ip(), 'user_agent': request.headers.get('User-Agent', '')[:200] } # Set session data session['security_token'] = session_token session['login_time'] = datetime.utcnow().isoformat() # Clear the failed-login counters for this IP + username and the username client_ip = self.get_client_ip() for key in self._login_attempt_keys(session.get('username')) + [f"ip:{client_ip}"]: self.failed_attempts.pop(key, None) self.log_security_event('secure_session_created', { 'user_id': user_id, 'ip': client_ip }) def encrypt_sensitive_data(self, data): """Encrypt sensitive data before storage""" if not self.cipher: return data # Return as-is if encryption not available try: if isinstance(data, str): data = data.encode('utf-8') encrypted_data = self.cipher.encrypt(data) return base64.b64encode(encrypted_data).decode('utf-8') except Exception as e: if self.logger_handler: self.logger_handler.logger.error(f"Encryption error: {e}") return data def decrypt_sensitive_data(self, encrypted_data): """Decrypt sensitive data""" if not self.cipher: return encrypted_data # Return as-is if encryption not available try: encrypted_bytes = base64.b64decode(encrypted_data.encode('utf-8')) decrypted_data = self.cipher.decrypt(encrypted_bytes) return decrypted_data.decode('utf-8') except Exception as e: if self.logger_handler: self.logger_handler.logger.error(f"Decryption error: {e}") return encrypted_data def log_security_event(self, event_type, details): """Log security events for monitoring""" try: security_log = { 'event_type': event_type, 'timestamp': datetime.utcnow().isoformat(), 'ip': self.get_client_ip(), 'user_agent': request.headers.get('User-Agent', ''), 'endpoint': request.endpoint, 'method': request.method, 'details': details } if self.logger_handler: self.logger_handler.log_security_event( event_type=event_type, description=f"Security event: {event_type}", additional_data=security_log ) except Exception as e: if self.logger_handler: self.logger_handler.logger.error(f"Security logging error: {e}") def register_security_routes(self): """Register security monitoring API endpoints""" @self.app.route('/api/security/status') def security_status(): """Get current security status""" try: # Admin only endpoint if not session.get('user_id') or session.get('role') != 'admin': return jsonify({'error': 'Access denied'}), 403 current_time = time.time() # Count active suspicious IPs suspicious_count = len([ ip for ip, count in self.suspicious_ips.items() if count > 3 ]) # Count recent failed attempts recent_failures = sum( len([ attempt for attempt in attempts if current_time - attempt < 300 # Last 5 minutes ]) for attempts in self.failed_attempts.values() ) # Count active sessions active_sessions = len([ token for token in self.session_tokens.values() if current_time - token['created'] < self.session_timeout ]) return jsonify({ 'suspicious_ips': suspicious_count, 'recent_failed_attempts': recent_failures, 'active_sessions': active_sessions, 'rate_limited_ips': len(self.failed_attempts), 'security_status': 'normal' if suspicious_count < 5 else 'elevated' }) except Exception as e: if self.logger_handler: self.logger_handler.logger.error(f"Security status error: {e}") return jsonify({'error': 'Failed to get security status'}), 500 @self.app.route('/api/security/clear-blocks', methods=['POST']) def clear_security_blocks(): """Clear security blocks (admin only)""" try: if not session.get('user_id') or session.get('role') != 'admin': return jsonify({'error': 'Access denied'}), 403 # Clear failed attempts cleared_ips = len(self.failed_attempts) self.failed_attempts.clear() # Clear suspicious IPs cleared_suspicious = len(self.suspicious_ips) self.suspicious_ips.clear() self.log_security_event('security_blocks_cleared', { 'admin_user': session.get('username'), 'cleared_failed_attempts': cleared_ips, 'cleared_suspicious_ips': cleared_suspicious }) return jsonify({ 'message': 'Security blocks cleared successfully', 'cleared_failed_attempts': cleared_ips, 'cleared_suspicious_ips': cleared_suspicious }) except Exception as e: if self.logger_handler: self.logger_handler.logger.error(f"Clear blocks error: {e}") return jsonify({'error': 'Failed to clear security blocks'}), 500 def enhanced_login_required(f): """ Enhanced login required decorator with security checks """ @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: return jsonify({'error': 'Authentication required'}), 401 # Additional security validation if not session.get('security_token'): session.clear() return jsonify({'error': 'Session security validation failed'}), 401 # Check session timeout - skip if "Remember Me" is enabled if not session.get('remember_me', False): login_time_str = session.get('login_time') if login_time_str: try: login_time = datetime.fromisoformat(login_time_str) if datetime.utcnow() - login_time > timedelta(hours=8): session.clear() return jsonify({'error': 'Session expired'}), 401 except ValueError: session.clear() return jsonify({'error': 'Invalid session data'}), 401 return f(*args, **kwargs) return decorated_function def csrf_protect(f): """ CSRF protection decorator """ @wraps(f) def decorated_function(*args, **kwargs): if request.method == 'POST': token = request.form.get('csrf_token') or request.headers.get('X-CSRF-Token') expected_token = session.get('csrf_token') if not token or not expected_token or not hmac.compare_digest(token, expected_token): return jsonify({'error': 'CSRF token validation failed'}), 403 return f(*args, **kwargs) return decorated_function def generate_csrf_token(): """Generate CSRF token for forms""" if 'csrf_token' not in session: session['csrf_token'] = secrets.token_urlsafe(32) return session['csrf_token']