from django.shortcuts import render from django.http import HttpResponse from django.conf import settings from utils.models import SystemSetting from users.models import Branch import time import os import mimetypes from django.http import HttpResponse, Http404 from django.core.exceptions import PermissionDenied from django.contrib.auth.decorators import login_required from django.utils.decorators import method_decorator from django.views.decorators.csrf import csrf_exempt class MaintenanceModeMiddleware: """Middleware to handle maintenance mode""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): # Check if maintenance mode is enabled maintenance_mode = SystemSetting.get_bool('maintenance_mode', False) if maintenance_mode: # Allow access to maintenance page and admin if request.path in ['/maintenance/', '/admin/', '/admin/login/'] or request.user.is_staff: return self.get_response(request) # Show maintenance page for all other requests return render(request, 'utils/maintenance.html', { 'maintenance_message': SystemSetting.get_setting('maintenance_message', 'System is under maintenance. Please try again later.') }) return self.get_response(request) class SessionTimeoutMiddleware: """Middleware to handle session timeout based on settings""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if request.user.is_authenticated: # Get session timeout from settings session_timeout_minutes = SystemSetting.get_int('session_timeout_minutes', 30) # Check if session has expired last_activity = request.session.get('last_activity') if last_activity: timeout_seconds = session_timeout_minutes * 60 if time.time() - last_activity > timeout_seconds: # Session expired, logout user from django.contrib.auth import logout logout(request) from django.contrib import messages messages.warning(request, 'Your session has expired. Please log in again.') from django.shortcuts import redirect return redirect('users:login') # Update last activity request.session['last_activity'] = time.time() return self.get_response(request) class MediaFileMiddleware_DISABLED: """ Middleware to serve media files in production when web server fails to serve them """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): # DISABLED: Let web server handle media files directly # Media file serving is now handled by Apache/web server return self.get_response(request) def serve_media_file(self, request): """Serve media files with security checks""" # Extract file path from URL file_path = request.path.replace('/media/', '') # Security: Prevent directory traversal if '..' in file_path or file_path.startswith('/'): raise Http404("Invalid file path") # Construct full path full_path = os.path.join(settings.MEDIA_ROOT, file_path) # Security: Ensure file is within MEDIA_ROOT if not os.path.abspath(full_path).startswith(os.path.abspath(settings.MEDIA_ROOT)): raise Http404("File not found") # Check if file exists if not os.path.exists(full_path) or not os.path.isfile(full_path): raise Http404("File not found") # Basic permission check for KYC documents if 'kyc/' in file_path and not request.user.is_authenticated: raise PermissionDenied("Authentication required") # Determine content type content_type, _ = mimetypes.guess_type(full_path) if content_type is None: content_type = 'application/octet-stream' # Serve the file try: with open(full_path, 'rb') as f: response = HttpResponse(f.read(), content_type=content_type) response['Content-Disposition'] = f'inline; filename="{os.path.basename(full_path)}"' return response except IOError: raise Http404("File not found") class BranchMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): # Set default branch for authenticated users if request.user.is_authenticated: # Handle users with NULL branch_id if request.user.branch_id is None: try: main_branch = Branch.objects.get(is_main_branch=True) request.user.branch_id = main_branch.id request.user.save() except Branch.DoesNotExist: pass # Get selected branch from session selected_branch_id = request.session.get('selected_branch_id') if selected_branch_id: # User has a selected branch - use it if they have access try: branch = Branch.objects.get(id=selected_branch_id) # Check if user has access to this branch if (request.user.role == 'admin' or request.user.branch_id == selected_branch_id or branch in request.user.accessible_branches.all()): request.branch = branch else: # User doesn't have access - fallback to their assigned branch request.branch = request.user.branch if request.user.branch: request.session['selected_branch_id'] = str(request.user.branch.id) else: # No assigned branch - use main branch main_branch = Branch.objects.filter(is_main_branch=True, is_active=True).first() if main_branch: request.branch = main_branch request.session['selected_branch_id'] = str(main_branch.id) except (Branch.DoesNotExist, ValueError): # Invalid branch ID - fallback to user's assigned branch request.branch = request.user.branch if request.user.branch: request.session['selected_branch_id'] = str(request.user.branch.id) else: # No assigned branch - use main branch main_branch = Branch.objects.filter(is_main_branch=True, is_active=True).first() if main_branch: request.branch = main_branch request.session['selected_branch_id'] = str(main_branch.id) else: # No selected branch - set default based on user type if request.user.role == 'admin': # Admin users - use their assigned branch or main branch if request.user.branch: request.branch = request.user.branch request.session['selected_branch_id'] = str(request.user.branch.id) else: main_branch = Branch.objects.filter(is_main_branch=True, is_active=True).first() if main_branch: request.branch = main_branch request.session['selected_branch_id'] = str(main_branch.id) else: # Non-admin users - use their assigned branch request.branch = request.user.branch if request.user.branch: request.session['selected_branch_id'] = str(request.user.branch.id) else: request.branch = None response = self.get_response(request) return response from django.utils.deprecation import MiddlewareMixin class BranchFilteringMiddleware(MiddlewareMixin): """ Middleware to ensure selected_branch_id is available in session only when needed """ def process_request(self, request): if request.user.is_authenticated: import uuid as _uuid selected_branch_id = request.session.get('selected_branch_id') # Validate UUID — clear stale branch name strings from old sessions if selected_branch_id: try: _uuid.UUID(str(selected_branch_id)) except (ValueError, AttributeError): request.session['selected_branch_id'] = None request.session.modified = True selected_branch_id = None # Only set default branch if no branch is selected AND it's the first login # This prevents auto-switching branches during navigation if not selected_branch_id and not request.session.get('branch_initialized'): # Set default branch only on first access if hasattr(request.user, 'branch') and request.user.branch: default_branch = request.user.branch else: default_branch = Branch.objects.filter(is_main_branch=True).first() if default_branch: request.session['selected_branch_id'] = str(default_branch.id) request.session['branch_initialized'] = True return None