""" Audit reporting and analysis views """ from django.shortcuts import render, get_object_or_404 from django.contrib.auth.decorators import login_required from django.http import JsonResponse, HttpResponse from django.db.models import Count, Q, F, Avg, Max, Min from django.utils import timezone from django.core.paginator import Paginator from django.contrib import messages from django.contrib.contenttypes.models import ContentType from datetime import datetime, timedelta, time import json import csv from io import StringIO from .models import ( CustomUser, UserAccessLog, EnhancedAuditLog, PermissionChangeLog, SecurityAlert, DataAccessPattern, AuditEventType, SecurityEventSeverity ) from .decorators import module_access_required from .audit_service import AuditService from reports.enhanced_pdf_service import EnhancedPDFService @login_required @module_access_required('audit_reports') def audit_dashboard(request): """Main audit dashboard with overview metrics""" # Get date range from request days = int(request.GET.get('days', 30)) start_date = timezone.now() - timedelta(days=days) # Basic metrics total_events = EnhancedAuditLog.objects.filter(timestamp__gte=start_date).count() security_events = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, is_security_event=True ).count() # User activity metrics active_users = EnhancedAuditLog.objects.filter( timestamp__gte=start_date ).values('user').distinct().count() # Permission changes permission_changes = PermissionChangeLog.objects.filter( timestamp__gte=start_date ).count() # Security alerts open_alerts = SecurityAlert.objects.filter(is_resolved=False).count() critical_alerts = SecurityAlert.objects.filter( is_resolved=False, severity__in=[SecurityEventSeverity.HIGH, SecurityEventSeverity.CRITICAL] ).count() # Event type breakdown event_breakdown = EnhancedAuditLog.objects.filter( timestamp__gte=start_date ).values('event_type').annotate( count=Count('id') ).order_by('-count')[:10] # Recent security events recent_security_events = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, is_security_event=True ).select_related('user').order_by('-timestamp')[:10] # Top active users top_users = EnhancedAuditLog.objects.filter( timestamp__gte=start_date ).values('user__first_name', 'user__last_name', 'user__role').annotate( activity_count=Count('id') ).order_by('-activity_count')[:10] context = { 'total_events': total_events, 'security_events': security_events, 'active_users': active_users, 'permission_changes': permission_changes, 'open_alerts': open_alerts, 'critical_alerts': critical_alerts, 'event_breakdown': event_breakdown, 'recent_security_events': recent_security_events, 'top_users': top_users, 'days': days, 'start_date': start_date, } return render(request, 'users/audit_dashboard.html', context) @login_required @module_access_required('audit_reports') def audit_trail_report(request): """Detailed audit trail report with filtering and search""" # Get filter parameters user_id = request.GET.get('user_id') event_type = request.GET.get('event_type') module = request.GET.get('module') start_date = request.GET.get('start_date') end_date = request.GET.get('end_date') search_query = request.GET.get('search', '').strip() security_only = request.GET.get('security_only') == 'true' # Build query logs = EnhancedAuditLog.objects.select_related('user').all() # Apply filters if user_id: logs = logs.filter(user_id=user_id) if event_type: logs = logs.filter(event_type=event_type) if module: logs = logs.filter(module=module) if start_date: try: start_dt = datetime.strptime(start_date, '%Y-%m-%d') logs = logs.filter(timestamp__gte=start_dt) except ValueError: messages.error(request, 'Invalid start date format') if end_date: try: end_dt = datetime.strptime(end_date, '%Y-%m-%d') logs = logs.filter(timestamp__lte=end_dt) except ValueError: messages.error(request, 'Invalid end date format') if search_query: logs = logs.filter( Q(description__icontains=search_query) | Q(action__icontains=search_query) | Q(user__first_name__icontains=search_query) | Q(user__last_name__icontains=search_query) ) if security_only: logs = logs.filter(is_security_event=True) # Order by timestamp logs = logs.order_by('-timestamp') # Pagination paginator = Paginator(logs, 50) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) # Get filter options users = CustomUser.objects.filter(is_active=True).order_by('first_name', 'last_name') event_types = EnhancedAuditLog.objects.values_list('event_type', flat=True).distinct().order_by('event_type') modules = EnhancedAuditLog.objects.values_list('module', flat=True).distinct().order_by('module') context = { 'page_obj': page_obj, 'users': users, 'event_types': event_types, 'modules': modules, 'filters': { 'user_id': user_id, 'event_type': event_type, 'module': module, 'start_date': start_date, 'end_date': end_date, 'search_query': search_query, 'security_only': security_only, } } return render(request, 'users/audit_trail_report.html', context) @login_required @module_access_required('audit_reports') def user_activity_analysis(request): """User activity analysis and pattern detection""" user_id = request.GET.get('user_id') days = int(request.GET.get('days', 30)) start_date = timezone.now() - timedelta(days=days) if user_id: user = get_object_or_404(CustomUser, id=user_id) # User-specific analysis user_logs = EnhancedAuditLog.objects.filter( user=user, timestamp__gte=start_date ) # Activity summary total_activities = user_logs.count() unique_modules = user_logs.values('module').distinct().count() security_events = user_logs.filter(is_security_event=True).count() # Daily activity pattern daily_activity = user_logs.extra( select={'day': 'DATE(timestamp)'} ).values('day').annotate( count=Count('id') ).order_by('day') # Module usage module_usage = user_logs.values('module').annotate( count=Count('id') ).order_by('-count') # Time pattern analysis hourly_pattern = user_logs.extra( select={'hour': 'EXTRACT(hour FROM timestamp)'} ).values('hour').annotate( count=Count('id') ).order_by('hour') # Recent activities recent_activities = user_logs.order_by('-timestamp')[:20] # Access patterns access_pattern = DataAccessPattern.objects.filter( user=user, date__gte=start_date.date() ).order_by('-date')[:30] context = { 'analyzed_user': user, 'total_activities': total_activities, 'unique_modules': unique_modules, 'security_events': security_events, 'daily_activity': list(daily_activity), 'module_usage': module_usage, 'hourly_pattern': list(hourly_pattern), 'recent_activities': recent_activities, 'access_pattern': access_pattern, 'days': days, } return render(request, 'users/user_activity_analysis.html', context) else: # System-wide activity analysis # Most active users active_users = EnhancedAuditLog.objects.filter( timestamp__gte=start_date ).values( 'user__first_name', 'user__last_name', 'user__role', 'user_id' ).annotate( activity_count=Count('id'), security_events=Count('id', filter=Q(is_security_event=True)) ).order_by('-activity_count')[:20] # Module activity module_activity = EnhancedAuditLog.objects.filter( timestamp__gte=start_date ).values('module').annotate( count=Count('id'), unique_users=Count('user', distinct=True) ).order_by('-count') # Unusual patterns detection unusual_patterns = [] # After-hours activity after_hours = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, timestamp__time__lt=time(8, 0), # Before 8 AM ).union( EnhancedAuditLog.objects.filter( timestamp__gte=start_date, timestamp__time__gt=time(18, 0), # After 6 PM ) ).values( 'user__first_name', 'user__last_name', 'user_id' ).annotate( count=Count('id') ).order_by('-count')[:10] # Bulk operations bulk_operations = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, event_type=AuditEventType.BULK_OPERATION ).values( 'user__first_name', 'user__last_name', 'user_id', 'action' ).annotate( count=Count('id') ).order_by('-count')[:10] context = { 'active_users': active_users, 'module_activity': module_activity, 'after_hours': after_hours, 'bulk_operations': bulk_operations, 'days': days, 'users': CustomUser.objects.filter(is_active=True).order_by('first_name', 'last_name'), } return render(request, 'users/system_activity_analysis.html', context) @login_required @module_access_required('audit_reports') def security_compliance_report(request): """Security compliance and monitoring report""" days = int(request.GET.get('days', 30)) start_date = timezone.now() - timedelta(days=days) # Security metrics total_security_events = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, is_security_event=True ).count() # Security alerts by severity alerts_by_severity = SecurityAlert.objects.filter( created_at__gte=start_date ).values('severity').annotate( count=Count('id') ).order_by('severity') # Failed access attempts failed_access = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, event_type=AuditEventType.FAILED_ACCESS ).values( 'user__first_name', 'user__last_name', 'user_id', 'ip_address' ).annotate( count=Count('id') ).order_by('-count')[:20] # Permission escalations permission_escalations = PermissionChangeLog.objects.filter( timestamp__gte=start_date, old_value=False, new_value=True ).select_related('user', 'changed_by').order_by('-timestamp')[:20] # Suspicious IP addresses suspicious_ips = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, is_security_event=True ).values('ip_address').annotate( event_count=Count('id'), unique_users=Count('user', distinct=True) ).order_by('-event_count')[:10] # Data export activities data_exports = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, event_type=AuditEventType.EXPORT_DATA ).select_related('user').order_by('-timestamp')[:20] # Open security alerts open_alerts = SecurityAlert.objects.filter( is_resolved=False ).order_by('-created_at')[:20] # Compliance metrics compliance_metrics = { 'password_changes': EnhancedAuditLog.objects.filter( timestamp__gte=start_date, action__icontains='password' ).count(), 'role_changes': PermissionChangeLog.objects.filter( timestamp__gte=start_date, permission_type='role' ).count(), 'critical_permission_grants': PermissionChangeLog.objects.filter( timestamp__gte=start_date, new_value=True, permission_name__in=[ 'users.delete_user', 'users.change_permissions', 'system.admin_access', 'reports.export_all' ] ).count(), } context = { 'total_security_events': total_security_events, 'alerts_by_severity': alerts_by_severity, 'failed_access': failed_access, 'permission_escalations': permission_escalations, 'suspicious_ips': suspicious_ips, 'data_exports': data_exports, 'open_alerts': open_alerts, 'compliance_metrics': compliance_metrics, 'days': days, 'start_date': start_date, } return render(request, 'users/security_compliance_report.html', context) @login_required @module_access_required('audit_reports') def permission_usage_analytics(request): """Permission usage analytics and optimization suggestions""" days = int(request.GET.get('days', 90)) # Default to 90 days for better analysis start_date = timezone.now() - timedelta(days=days) # Permission usage frequency permission_usage = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, permission_checked__isnull=False ).values('permission_checked').annotate( total_checks=Count('id'), granted_count=Count('id', filter=Q(permission_granted=True)), denied_count=Count('id', filter=Q(permission_granted=False)), unique_users=Count('user', distinct=True) ).order_by('-total_checks') # Unused permissions (permissions that exist but are never checked) from .enhanced_permissions_models import PagePermission all_permissions = PagePermission.objects.values_list('action_code', flat=True) used_permissions = set( EnhancedAuditLog.objects.filter( timestamp__gte=start_date, permission_checked__isnull=False ).values_list('permission_checked', flat=True).distinct() ) unused_permissions = set(all_permissions) - used_permissions # Most denied permissions most_denied = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, permission_granted=False ).values('permission_checked').annotate( denied_count=Count('id'), unique_users=Count('user', distinct=True) ).order_by('-denied_count')[:20] # Permission changes by role role_permission_changes = PermissionChangeLog.objects.filter( timestamp__gte=start_date ).values('user__role').annotate( total_changes=Count('id'), grants=Count('id', filter=Q(new_value=True)), revokes=Count('id', filter=Q(new_value=False)) ).order_by('-total_changes') # Users with most permission denials users_with_denials = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, permission_granted=False ).values( 'user__first_name', 'user__last_name', 'user__role', 'user_id' ).annotate( denial_count=Count('id'), unique_permissions=Count('permission_checked', distinct=True) ).order_by('-denial_count')[:20] # Generate optimization suggestions suggestions = [] # Suggest removing unused permissions if unused_permissions: suggestions.append({ 'type': 'cleanup', 'title': 'Remove Unused Permissions', 'description': f'Consider removing {len(unused_permissions)} permissions that haven\'t been used in {days} days', 'impact': 'low', 'permissions': list(unused_permissions)[:10] # Show first 10 }) # Suggest role adjustments for frequently denied permissions for denial in most_denied[:5]: if denial['denied_count'] > 10: suggestions.append({ 'type': 'role_adjustment', 'title': f'Review Permission: {denial["permission_checked"]}', 'description': f'This permission was denied {denial["denied_count"]} times to {denial["unique_users"]} users', 'impact': 'medium', 'permission': denial['permission_checked'] }) # Suggest training for users with many denials for user in users_with_denials[:3]: if user['denial_count'] > 20: suggestions.append({ 'type': 'training', 'title': f'User Training Needed: {user["user__first_name"]} {user["user__last_name"]}', 'description': f'User has {user["denial_count"]} permission denials across {user["unique_permissions"]} different permissions', 'impact': 'high', 'user_id': user['user_id'] }) context = { 'permission_usage': permission_usage, 'unused_permissions': list(unused_permissions)[:20], # Limit display 'most_denied': most_denied, 'role_permission_changes': role_permission_changes, 'users_with_denials': users_with_denials, 'suggestions': suggestions, 'days': days, 'start_date': start_date, } return render(request, 'users/permission_usage_analytics.html', context) @login_required @module_access_required('audit_reports') def export_audit_report(request): """Export audit reports in various formats""" report_type = request.GET.get('type', 'audit_trail') format_type = request.GET.get('format', 'csv') if report_type == 'audit_trail': return _export_audit_trail(request, format_type) elif report_type == 'security_compliance': return _export_security_compliance(request, format_type) elif report_type == 'permission_usage': return _export_permission_usage(request, format_type) else: return JsonResponse({'error': 'Invalid report type'}, status=400) def _export_audit_trail(request, format_type): """Export audit trail data""" # Apply same filters as audit_trail_report view logs = EnhancedAuditLog.objects.select_related('user').all() # Apply filters (same logic as audit_trail_report) user_id = request.GET.get('user_id') event_type = request.GET.get('event_type') module = request.GET.get('module') start_date = request.GET.get('start_date') end_date = request.GET.get('end_date') if user_id: logs = logs.filter(user_id=user_id) if event_type: logs = logs.filter(event_type=event_type) if module: logs = logs.filter(module=module) if start_date: try: start_dt = datetime.strptime(start_date, '%Y-%m-%d') logs = logs.filter(timestamp__gte=start_dt) except ValueError: pass if end_date: try: end_dt = datetime.strptime(end_date, '%Y-%m-%d') logs = logs.filter(timestamp__lte=end_dt) except ValueError: pass logs = logs.order_by('-timestamp')[:1000] # Limit to 1000 records if format_type == 'csv': response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="audit_trail.csv"' writer = csv.writer(response) writer.writerow([ 'Timestamp', 'User', 'Role', 'Event Type', 'Action', 'Module', 'Permission Checked', 'Permission Granted', 'IP Address', 'Description' ]) for log in logs: writer.writerow([ log.timestamp.strftime('%Y-%m-%d %H:%M:%S'), log.user.get_full_name() if log.user else 'Anonymous', log.role_at_time or '', log.event_type, log.action, log.module, log.permission_checked or '', 'Yes' if log.permission_granted else 'No' if log.permission_granted is not None else '', log.ip_address or '', log.description ]) return response elif format_type == 'pdf': # Use enhanced PDF service pdf_service = EnhancedPDFService() # Prepare data for PDF data = [] for log in logs: data.append({ 'timestamp': log.timestamp.strftime('%Y-%m-%d %H:%M:%S'), 'user': log.user.get_full_name() if log.user else 'Anonymous', 'event_type': log.event_type, 'action': log.action, 'module': log.module, 'description': log.description[:100] + '...' if len(log.description) > 100 else log.description }) pdf_content = pdf_service.generate_audit_report(data, 'Audit Trail Report') response = HttpResponse(pdf_content, content_type='application/pdf') response['Content-Disposition'] = 'attachment; filename="audit_trail.pdf"' return response return JsonResponse({'error': 'Invalid format'}, status=400) def _export_security_compliance(request, format_type): """Export security compliance data""" days = int(request.GET.get('days', 30)) start_date = timezone.now() - timedelta(days=days) # Get security events security_events = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, is_security_event=True ).select_related('user').order_by('-timestamp')[:500] if format_type == 'csv': response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="security_compliance.csv"' writer = csv.writer(response) writer.writerow([ 'Timestamp', 'User', 'Event Type', 'Action', 'Severity', 'IP Address', 'Description', 'Requires Attention' ]) for event in security_events: writer.writerow([ event.timestamp.strftime('%Y-%m-%d %H:%M:%S'), event.user.get_full_name() if event.user else 'Anonymous', event.event_type, event.action, event.severity or '', event.ip_address or '', event.description, 'Yes' if event.requires_attention else 'No' ]) return response return JsonResponse({'error': 'Invalid format'}, status=400) def _export_permission_usage(request, format_type): """Export permission usage analytics""" days = int(request.GET.get('days', 90)) start_date = timezone.now() - timedelta(days=days) # Get permission usage data permission_usage = EnhancedAuditLog.objects.filter( timestamp__gte=start_date, permission_checked__isnull=False ).values('permission_checked').annotate( total_checks=Count('id'), granted_count=Count('id', filter=Q(permission_granted=True)), denied_count=Count('id', filter=Q(permission_granted=False)), unique_users=Count('user', distinct=True) ).order_by('-total_checks') if format_type == 'csv': response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="permission_usage.csv"' writer = csv.writer(response) writer.writerow([ 'Permission', 'Total Checks', 'Granted', 'Denied', 'Unique Users', 'Grant Rate' ]) for usage in permission_usage: grant_rate = (usage['granted_count'] / usage['total_checks'] * 100) if usage['total_checks'] > 0 else 0 writer.writerow([ usage['permission_checked'], usage['total_checks'], usage['granted_count'], usage['denied_count'], usage['unique_users'], f"{grant_rate:.1f}%" ]) return response return JsonResponse({'error': 'Invalid format'}, status=400) @login_required @module_access_required('audit_reports') def security_alerts(request): """Security alerts management""" # Filter parameters severity = request.GET.get('severity') alert_type = request.GET.get('alert_type') resolved = request.GET.get('resolved') alerts = SecurityAlert.objects.all() if severity: alerts = alerts.filter(severity=severity) if alert_type: alerts = alerts.filter(alert_type=alert_type) if resolved == 'true': alerts = alerts.filter(is_resolved=True) elif resolved == 'false': alerts = alerts.filter(is_resolved=False) alerts = alerts.select_related('user', 'resolved_by').order_by('-created_at') # Pagination paginator = Paginator(alerts, 20) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) context = { 'page_obj': page_obj, 'severity_choices': SecurityEventSeverity.choices, 'alert_type_choices': SecurityAlert.ALERT_TYPES, 'filters': { 'severity': severity, 'alert_type': alert_type, 'resolved': resolved, } } return render(request, 'users/security_alerts.html', context) @login_required @module_access_required('audit_reports') def resolve_alert(request, alert_id): """Resolve a security alert""" if request.method == 'POST': alert = get_object_or_404(SecurityAlert, id=alert_id) resolution_notes = request.POST.get('resolution_notes', '') alert.resolve(request.user, resolution_notes) # Log the resolution audit_service = AuditService() audit_service.log_user_action( request.user, f'resolve_security_alert_{alert.alert_type}', 'security', request, alert_id=str(alert_id), resolution_notes=resolution_notes ) messages.success(request, 'Security alert resolved successfully.') return JsonResponse({'success': True}) return JsonResponse({'error': 'Invalid request method'}, status=405)