# -*- coding: utf-8 -*- """ Enhanced Authentication System for Stream Processor This module provides enhanced authentication features including: - Two-Factor Authentication (2FA) with TOTP - Enhanced session management - Password policies and validation - Account lockout protection - Login attempt tracking - Security logging Author: Stream Processor Development Team Version: 1.0.0 Created: 2025 """ from datetime import timedelta from django.db import models from django.contrib.auth.models import User from django.contrib.auth.backends import ModelBackend from django.contrib.sessions.models import Session from django.contrib.auth.signals import user_logged_in, user_logged_out from django.dispatch import receiver from django.core.exceptions import ValidationError from django.utils.translation import gettext_lazy as _ from apps.core.models import TimestampedModel from apps.authentication.models import UserProfile, LoginAttempt, UserSession class EnhancedAuthenticationBackend(ModelBackend): """ Enhanced authentication backend with security features. """ def authenticate(self, request, username=None, password=None, **kwargs): """ Authenticate user with enhanced security checks. """ if username is None or password is None: return None try: user = User.objects.get(username=username) except User.DoesNotExist: # Log failed attempt self._log_login_attempt(request, username, False, "User not found") return None # Get or create security profile profile, created = UserProfile.objects.get_or_create(user=user) # Check if account is locked if profile.is_account_locked(): self._log_login_attempt(request, username, False, "Account locked") return None # Check password if user.check_password(password): # Check if 2FA is required if profile.is_2fa_enabled: # Don't complete authentication yet - need 2FA return None # Reset failed attempts on successful login profile.reset_failed_attempts() # Update last login IP if request: profile.last_login_ip = self._get_client_ip(request) profile.save(update_fields=['last_login_ip']) self._log_login_attempt(request, username, True) return user else: # Increment failed attempts profile.increment_failed_attempts() self._log_login_attempt(request, username, False, "Invalid password") return None def _get_client_ip(self, request): """ Get client IP address from request. """ x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0] else: ip = request.META.get('REMOTE_ADDR') return ip def _log_login_attempt(self, request, username, success, failure_reason=''): """ Log login attempt for security monitoring. """ if not request: return LoginAttempt.objects.create( username=username, ip_address=self._get_client_ip(request), user_agent=request.META.get('HTTP_USER_AGENT', ''), success=success, failure_reason=failure_reason ) def validate_strong_password(password): """ Validate password strength. Args: password: Password to validate Raises: ValidationError: If password doesn't meet requirements """ errors = [] if len(password) < 12: errors.append(_("Password must be at least 12 characters long.")) if not any(c.isupper() for c in password): errors.append(_("Password must contain at least one uppercase letter.")) if not any(c.islower() for c in password): errors.append(_("Password must contain at least one lowercase letter.")) if not any(c.isdigit() for c in password): errors.append(_("Password must contain at least one digit.")) if not any(c in '!@#$%^&*()_+-=[]{}|;:,.<>?' for c in password): errors.append(_("Password must contain at least one special character.")) # Check for common passwords common_passwords = [ 'password', '123456', 'password123', 'admin', 'qwerty', 'letmein', 'welcome', 'monkey', '1234567890' ] if password.lower() in common_passwords: errors.append(_("Password is too common.")) if errors: raise ValidationError(errors) # Signal handlers for session management @receiver(user_logged_in) def on_user_logged_in(sender, request, user, **kwargs): """ Handle user login - create session record and enforce limits. """ if not request: return # Get or create security profile profile, created = UserProfile.objects.get_or_create(user=user) # Create session record UserSession.objects.create( user=user, session_key=request.session.session_key, ip_address=EnhancedAuthenticationBackend()._get_client_ip(request), user_agent=request.META.get('HTTP_USER_AGENT', '') ) # Enforce concurrent session limits active_sessions = UserSession.objects.filter( user=user, is_active=True ).order_by('-last_activity') if active_sessions.count() > profile.max_concurrent_sessions: # Deactivate oldest sessions sessions_to_deactivate = active_sessions[profile.max_concurrent_sessions:] for session in sessions_to_deactivate: session.is_active = False session.save() # Delete the actual Django session try: Session.objects.get(session_key=session.session_key).delete() except Session.DoesNotExist: pass @receiver(user_logged_out) def on_user_logged_out(sender, request, user, **kwargs): """ Handle user logout - deactivate session record. """ if not request or not user: return try: session = UserSession.objects.get( user=user, session_key=request.session.session_key ) session.is_active = False session.save() except UserSession.DoesNotExist: pass def cleanup_expired_sessions(): """ Clean up expired sessions and login attempts. This function handles deleting old data to maintain database performance and manage storage space effectively. It performs three cleanup operations: 1. Removes old login attempts older than 30 days 2. Deletes inactive user sessions older than 7 days 3. Marks sessions as inactive if not active for 24+ hours This should be called periodically (e.g., via Celery task) to prevent the authentication tables from growing indefinitely. Data Retention Policy: - Login attempts: 30 days - Inactive sessions: 7 days - Session activity timeout: 24 hours """ # Clean up old login attempts (keep for 30 days) # This removes failed/successful login records to prevent table bloat cutoff_date = timezone.now() - timedelta(days=30) deleted_attempts = LoginAttempt.objects.filter(created_at__lt=cutoff_date).delete()[0] # Clean up old inactive sessions (keep for 7 days) # This removes session records that are no longer active to free up space cutoff_date = timezone.now() - timedelta(days=7) deleted_sessions = UserSession.objects.filter( is_active=False, updated_at__lt=cutoff_date ).delete()[0] # Mark sessions as inactive if they haven't been active recently # This ensures stale sessions are properly marked as inactive inactive_cutoff = timezone.now() - timedelta(hours=24) updated_sessions = UserSession.objects.filter( is_active=True, last_activity__lt=inactive_cutoff ).update(is_active=False) # Log cleanup results for monitoring logger.info( f"Authentication cleanup completed: " f"deleted {deleted_attempts} login attempts, " f"deleted {deleted_sessions} inactive sessions, " f"marked {updated_sessions} sessions as inactive" )