# -*- coding: utf-8 -*- """ Enhanced Authentication Views for Stream Processor This module provides views for enhanced authentication features including: - Enhanced login with 2FA support - 2FA setup and management - Password change with strength validation - Security settings management - Session management Author: Stream Processor Development Team Version: 1.0.0 Created: 2025 """ from django.shortcuts import render, redirect, get_object_or_404 from django.contrib.auth import login, logout, authenticate from django.contrib.auth.decorators import login_required from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib.auth.models import User from django.contrib import messages from django.views.generic import TemplateView, FormView from django.views.decorators.cache import never_cache from django.views.decorators.csrf import csrf_protect from django.views.decorators.http import require_http_methods from django.utils.decorators import method_decorator from django.urls import reverse_lazy, reverse from django.http import JsonResponse, HttpResponseRedirect from django.core.exceptions import ValidationError from django.utils.translation import gettext_lazy as _ from django.utils import timezone from django.conf import settings from django.contrib.auth import update_session_auth_hash from django.db import transaction import json from .authentication import UserProfile, LoginAttempt, UserSession from .forms import ( EnhancedLoginForm, TwoFactorForm, EnhancedPasswordChangeForm, Setup2FAForm, Disable2FAForm, SecuritySettingsForm, BackupCodesForm ) @method_decorator(csrf_protect, name='dispatch') @method_decorator(never_cache, name='dispatch') class EnhancedLoginView(FormView): """ Enhanced login view with 2FA support. """ template_name = 'authentication/login.html' form_class = EnhancedLoginForm success_url = reverse_lazy('monitoring:dashboard') def dispatch(self, request, *args, **kwargs): # Redirect if already authenticated if request.user.is_authenticated: return redirect(self.get_success_url()) return super().dispatch(request, *args, **kwargs) def get_form_kwargs(self): kwargs = super().get_form_kwargs() kwargs['request'] = self.request return kwargs def form_valid(self, form): user = form.get_user() # Check if 2FA is enabled try: profile = UserProfile.objects.get(user=user) if profile.is_2fa_enabled: # Store user in session for 2FA verification self.request.session['pre_2fa_user_id'] = user.id self.request.session['pre_2fa_backend'] = user.backend return redirect('core:two_factor_verify') except UserProfile.DoesNotExist: # Create profile if it doesn't exist UserProfile.objects.create(user=user) # Complete login login(self.request, user) # Handle remember me if form.cleaned_data.get('remember_me'): self.request.session.set_expiry(settings.SESSION_COOKIE_AGE) else: self.request.session.set_expiry(0) # Browser session messages.success(self.request, _("Welcome back, {}!").format(user.get_full_name() or user.username)) return super().form_valid(form) def form_invalid(self, form): messages.error(self.request, _("Login failed. Please check your credentials and try again.")) return super().form_invalid(form) @method_decorator(csrf_protect, name='dispatch') @method_decorator(never_cache, name='dispatch') class TwoFactorVerifyView(FormView): """ View for 2FA token verification. """ template_name = 'authentication/two_factor_verify.html' form_class = TwoFactorForm success_url = reverse_lazy('monitoring:dashboard') def dispatch(self, request, *args, **kwargs): # Check if user is in 2FA verification state if 'pre_2fa_user_id' not in request.session: return redirect('core:login') return super().dispatch(request, *args, **kwargs) def get_form_kwargs(self): kwargs = super().get_form_kwargs() user_id = self.request.session.get('pre_2fa_user_id') kwargs['user'] = get_object_or_404(User, id=user_id) return kwargs def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) user_id = self.request.session.get('pre_2fa_user_id') user = get_object_or_404(User, id=user_id) context['user'] = user return context def form_valid(self, form): user_id = self.request.session.get('pre_2fa_user_id') backend = self.request.session.get('pre_2fa_backend') user = get_object_or_404(User, id=user_id) # Set backend for login user.backend = backend # Complete login login(self.request, user) # Clean up session del self.request.session['pre_2fa_user_id'] del self.request.session['pre_2fa_backend'] messages.success(self.request, _("Welcome back, {}!").format(user.get_full_name() or user.username)) return super().form_valid(form) def form_invalid(self, form): messages.error(self.request, _("Invalid verification code. Please try again.")) return super().form_invalid(form) class BackupCodeVerifyView(FormView): """ View for backup code verification. """ template_name = 'authentication/backup_code_verify.html' form_class = BackupCodesForm success_url = reverse_lazy('monitoring:dashboard') def dispatch(self, request, *args, **kwargs): # Check if user is in 2FA verification state if 'pre_2fa_user_id' not in request.session: return redirect('core:login') return super().dispatch(request, *args, **kwargs) def get_form_kwargs(self): kwargs = super().get_form_kwargs() user_id = self.request.session.get('pre_2fa_user_id') kwargs['user'] = get_object_or_404(User, id=user_id) return kwargs def form_valid(self, form): user_id = self.request.session.get('pre_2fa_user_id') backend = self.request.session.get('pre_2fa_backend') user = get_object_or_404(User, id=user_id) # Remove used backup code profile = UserProfile.objects.get(user=user) backup_code = form.cleaned_data['backup_code'] profile.backup_codes.remove(backup_code) profile.save(update_fields=['backup_codes']) # Set backend for login user.backend = backend # Complete login login(self.request, user) # Clean up session del self.request.session['pre_2fa_user_id'] del self.request.session['pre_2fa_backend'] messages.success(self.request, _("Welcome back! You have {} backup codes remaining.").format(len(profile.backup_codes))) return super().form_valid(form) @method_decorator(login_required, name='dispatch') class Setup2FAView(TemplateView): """ View for setting up 2FA. """ template_name = 'authentication/setup_2fa.html' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) # Get or create user profile profile, created = UserProfile.objects.get_or_create(user=self.request.user) # Generate 2FA setup data secret, qr_code_url = profile.setup_2fa() context.update({ 'secret': secret, 'qr_code_url': qr_code_url, 'form': Setup2FAForm(user=self.request.user) }) return context def post(self, request, *args, **kwargs): form = Setup2FAForm(user=request.user, data=request.POST) if form.is_valid(): # Enable 2FA and generate backup codes profile = UserProfile.objects.get(user=request.user) backup_codes = profile.enable_2fa() # Store backup codes in session to display them request.session['new_backup_codes'] = backup_codes messages.success(request, _("Two-factor authentication has been enabled successfully!")) return redirect('core:show_backup_codes') # If form is invalid, show the setup page again with errors context = self.get_context_data(**kwargs) context['form'] = form return render(request, self.template_name, context) @method_decorator(login_required, name='dispatch') class ShowBackupCodesView(TemplateView): """ View for displaying backup codes after 2FA setup. """ template_name = 'authentication/show_backup_codes.html' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) # Get backup codes from session backup_codes = self.request.session.get('new_backup_codes', []) context['backup_codes'] = backup_codes return context def post(self, request, *args, **kwargs): # Clear backup codes from session after user confirms they've saved them if 'new_backup_codes' in request.session: del request.session['new_backup_codes'] messages.success(request, _("Two-factor authentication setup is complete!")) return redirect('core:security_settings') @method_decorator(login_required, name='dispatch') class Disable2FAView(FormView): """ View for disabling 2FA. """ template_name = 'authentication/disable_2fa.html' form_class = Disable2FAForm success_url = reverse_lazy('core:security_settings') def get_form_kwargs(self): kwargs = super().get_form_kwargs() kwargs['user'] = self.request.user return kwargs def form_valid(self, form): # Disable 2FA try: profile = UserProfile.objects.get(user=self.request.user) profile.disable_2fa() messages.success(self.request, _("Two-factor authentication has been disabled.")) except UserProfile.DoesNotExist: messages.error(self.request, _("2FA is not enabled for your account.")) return super().form_valid(form) @method_decorator(login_required, name='dispatch') class SecuritySettingsView(TemplateView): """ View for managing security settings. """ template_name = 'authentication/security_settings.html' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) # Get or create user profile profile, created = UserProfile.objects.get_or_create(user=self.request.user) # Get recent login attempts recent_attempts = LoginAttempt.objects.filter( username=self.request.user.username ).order_by('-created_at')[:10] # Get active sessions active_sessions = UserSession.objects.filter( user=self.request.user, is_active=True ).order_by('-last_activity') context.update({ 'profile': profile, 'security_form': SecuritySettingsForm(instance=profile), 'password_form': EnhancedPasswordChangeForm(user=self.request.user), 'recent_attempts': recent_attempts, 'active_sessions': active_sessions, 'current_session_key': self.request.session.session_key }) return context def post(self, request, *args, **kwargs): profile = UserProfile.objects.get(user=request.user) if 'update_security' in request.POST: form = SecuritySettingsForm(data=request.POST, instance=profile) if form.is_valid(): form.save() messages.success(request, _("Security settings updated successfully.")) else: messages.error(request, _("Please correct the errors below.")) elif 'change_password' in request.POST: form = EnhancedPasswordChangeForm(user=request.user, data=request.POST) if form.is_valid(): user = form.save() # Update password change timestamp profile.last_password_change = timezone.now() profile.password_reset_required = False profile.save(update_fields=['last_password_change', 'password_reset_required']) # Keep user logged in after password change update_session_auth_hash(request, user) messages.success(request, _("Password changed successfully.")) else: messages.error(request, _("Please correct the password errors below.")) return redirect('core:security_settings') @login_required @require_http_methods(["POST"]) def terminate_session(request): """ Terminate a specific user session. """ session_key = request.POST.get('session_key') if not session_key: return JsonResponse({'error': 'Session key required'}, status=400) try: # Find the session user_session = UserSession.objects.get( user=request.user, session_key=session_key, is_active=True ) # Don't allow terminating current session if session_key == request.session.session_key: return JsonResponse({'error': 'Cannot terminate current session'}, status=400) # Deactivate the session user_session.is_active = False user_session.save() # Delete the actual Django session from django.contrib.sessions.models import Session try: Session.objects.get(session_key=session_key).delete() except Session.DoesNotExist: pass return JsonResponse({'success': True}) except UserSession.DoesNotExist: return JsonResponse({'error': 'Session not found'}, status=404) @login_required @require_http_methods(["POST"]) def terminate_all_sessions(request): """ Terminate all other user sessions except current one. """ current_session_key = request.session.session_key # Get all active sessions except current other_sessions = UserSession.objects.filter( user=request.user, is_active=True ).exclude(session_key=current_session_key) # Deactivate sessions session_keys = list(other_sessions.values_list('session_key', flat=True)) other_sessions.update(is_active=False) # Delete actual Django sessions from django.contrib.sessions.models import Session Session.objects.filter(session_key__in=session_keys).delete() count = len(session_keys) messages.success(request, _("Terminated {} other session(s).").format(count)) return JsonResponse({'success': True, 'terminated_count': count}) @login_required def regenerate_backup_codes(request): """ Regenerate 2FA backup codes. """ if request.method == 'POST': try: profile = UserProfile.objects.get(user=request.user) if not profile.is_2fa_enabled: messages.error(request, _("2FA is not enabled for your account.")) return redirect('core:security_settings') # Generate new backup codes import secrets backup_codes = [secrets.token_hex(4).upper() for _ in range(10)] profile.backup_codes = backup_codes profile.save(update_fields=['backup_codes']) # Store in session to display request.session['new_backup_codes'] = backup_codes messages.success(request, _("New backup codes have been generated.")) return redirect('core:show_backup_codes') except UserProfile.DoesNotExist: messages.error(request, _("Security profile not found.")) return redirect('core:security_settings') def enhanced_logout(request): """ Enhanced logout view with session cleanup. """ if request.user.is_authenticated: # Mark session as inactive try: user_session = UserSession.objects.get( user=request.user, session_key=request.session.session_key ) user_session.is_active = False user_session.save() except UserSession.DoesNotExist: pass logout(request) messages.success(request, _("You have been logged out successfully.")) return redirect('core:login')