""" Enhanced audit service for comprehensive logging and security monitoring """ from django.utils import timezone from django.contrib.auth import get_user_model from django.db.models import Count, Q, F from django.core.cache import cache from datetime import datetime, timedelta, time import json import logging from typing import Optional, Dict, Any, List from .models import ( EnhancedAuditLog, PermissionChangeLog, SecurityAlert, DataAccessPattern, AuditEventType, SecurityEventSeverity ) User = get_user_model() logger = logging.getLogger(__name__) class AuditService: """ Comprehensive audit service for logging and monitoring user activities """ def __init__(self): self.cache_timeout = 300 # 5 minutes def log_user_action(self, user, action, module, request=None, **kwargs): """ Log a general user action with enhanced context """ try: # Determine event type based on action event_type = self._determine_event_type(action) # Create enhanced audit log audit_log = EnhancedAuditLog.objects.create( user=user, event_type=event_type, action=action, module=module, role_at_time=user.role if user else None, branch_context=self._get_branch_context(user), ip_address=self._get_client_ip(request) if request else None, user_agent=request.META.get('HTTP_USER_AGENT', '') if request else None, session_id=request.session.session_key if request and hasattr(request, 'session') else None, request_method=request.method if request else None, request_path=request.path if request else None, request_params=self._sanitize_request_params(request) if request else {}, description=kwargs.get('description', f"User performed {action} in {module}"), additional_data=kwargs ) # Update access patterns self._update_access_pattern(user, request) # Check for security anomalies self._check_security_anomalies(user, action, request) return audit_log except Exception as e: logger.error(f"Failed to log user action: {e}") return None def log_permission_check(self, user, permission, granted, request=None, **kwargs): """ Log permission check with detailed context """ try: audit_log = EnhancedAuditLog.log_permission_check( user=user, permission=permission, granted=granted, request=request, **kwargs ) # If permission denied, check for security concerns if not granted: self._handle_permission_denial(user, permission, request) return audit_log except Exception as e: logger.error(f"Failed to log permission check: {e}") return None def log_permission_change(self, user, changed_by, permission_name, old_value, new_value, reason="", request=None): """ Log permission changes with approval tracking """ try: # Determine if approval is required for this change approval_required = self._requires_approval(permission_name, old_value, new_value) change_log = PermissionChangeLog.objects.create( user=user, changed_by=changed_by, permission_type='custom', permission_name=permission_name, old_value=old_value, new_value=new_value, reason=reason, approval_required=approval_required, ip_address=self._get_client_ip(request) if request else None, session_id=request.session.session_key if request and hasattr(request, 'session') else None ) # Log as audit event self.log_user_action( user=changed_by, action='change_permission', module='permissions', request=request, target_user=user.get_full_name(), permission=permission_name, old_value=old_value, new_value=new_value, approval_required=approval_required ) # Create security alert for critical permission changes if self._is_critical_permission(permission_name): self._create_security_alert( alert_type='critical_permission_grant', user=user, severity=SecurityEventSeverity.HIGH, title=f"Critical Permission Changed: {permission_name}", description=f"User {user.get_full_name()} had critical permission '{permission_name}' changed from {old_value} to {new_value} by {changed_by.get_full_name()}", request=request, additional_context={ 'permission': permission_name, 'old_value': old_value, 'new_value': new_value, 'changed_by': changed_by.get_full_name() } ) return change_log except Exception as e: logger.error(f"Failed to log permission change: {e}") return None def log_data_access(self, user, model_class, object_id, action, request=None, **kwargs): """ Log data access with object tracking """ try: audit_log = EnhancedAuditLog.log_data_access( user=user, object_type=model_class, object_id=object_id, action=action, request=request, **kwargs ) # Check for bulk data access patterns self._check_bulk_access_pattern(user, model_class, action) return audit_log except Exception as e: logger.error(f"Failed to log data access: {e}") return None def log_security_event(self, user, action, severity, description, request=None, **kwargs): """ Log security events and create alerts """ try: # Log the security event audit_log = EnhancedAuditLog.log_security_event( user=user, action=action, severity=severity, description=description, request=request, **kwargs ) # Create security alert for medium+ severity events if severity in [SecurityEventSeverity.MEDIUM, SecurityEventSeverity.HIGH, SecurityEventSeverity.CRITICAL]: self._create_security_alert( alert_type='security_violation', user=user, severity=severity, title=f"Security Event: {action}", description=description, request=request, additional_context=kwargs ) return audit_log except Exception as e: logger.error(f"Failed to log security event: {e}") return None def get_user_activity_summary(self, user, days=30): """ Get user activity summary for the specified period """ try: end_date = timezone.now() start_date = end_date - timedelta(days=days) # Get activity counts activities = EnhancedAuditLog.objects.filter( user=user, timestamp__gte=start_date ).values('event_type').annotate(count=Count('id')) # Get permission denials permission_denials = EnhancedAuditLog.objects.filter( user=user, timestamp__gte=start_date, permission_granted=False ).count() # Get security events security_events = EnhancedAuditLog.objects.filter( user=user, timestamp__gte=start_date, is_security_event=True ).count() # Get access patterns access_patterns = DataAccessPattern.objects.filter( user=user, date__gte=start_date.date() ).aggregate( avg_requests=models.Avg('total_requests'), total_exports=models.Sum('data_exports'), anomalous_days=Count('id', filter=Q(is_anomalous=True)) ) return { 'user': user, 'period_days': days, 'activities': {item['event_type']: item['count'] for item in activities}, 'permission_denials': permission_denials, 'security_events': security_events, 'access_patterns': access_patterns, 'last_activity': EnhancedAuditLog.objects.filter(user=user).first() } except Exception as e: logger.error(f"Failed to get user activity summary: {e}") return {} def detect_anomalies(self, user=None, days=7): """ Detect anomalous user behavior patterns """ try: end_date = timezone.now().date() start_date = end_date - timedelta(days=days) query = DataAccessPattern.objects.filter(date__gte=start_date) if user: query = query.filter(user=user) anomalies = [] for pattern in query.filter(is_anomalous=True): anomalies.append({ 'user': pattern.user, 'date': pattern.date, 'anomaly_score': pattern.anomaly_score, 'reasons': pattern.anomaly_reasons, 'metrics': { 'total_requests': pattern.total_requests, 'failed_attempts': pattern.failed_access_attempts, 'after_hours_access': pattern.after_hours_access, 'unique_ips': pattern.unique_ip_addresses } }) return anomalies except Exception as e: logger.error(f"Failed to detect anomalies: {e}") return [] def get_security_alerts(self, resolved=None, severity=None, days=30): """ Get security alerts with filtering options """ try: end_date = timezone.now() start_date = end_date - timedelta(days=days) query = SecurityAlert.objects.filter(created_at__gte=start_date) if resolved is not None: query = query.filter(is_resolved=resolved) if severity: query = query.filter(severity=severity) return query.select_related('user', 'resolved_by').order_by('-created_at') except Exception as e: logger.error(f"Failed to get security alerts: {e}") return SecurityAlert.objects.none() # Private helper methods def _determine_event_type(self, action): """Determine event type based on action""" action_lower = action.lower() if 'login' in action_lower: return AuditEventType.LOGIN elif 'logout' in action_lower: return AuditEventType.LOGOUT elif 'permission' in action_lower: return AuditEventType.PERMISSION_CHANGE elif 'export' in action_lower or 'download' in action_lower: return AuditEventType.EXPORT_DATA elif 'report' in action_lower: return AuditEventType.REPORT_GENERATION elif 'bulk' in action_lower: return AuditEventType.BULK_OPERATION elif any(word in action_lower for word in ['create', 'update', 'delete', 'modify']): return AuditEventType.DATA_MODIFICATION else: return AuditEventType.DATA_ACCESS def _get_branch_context(self, user): """Get branch context for user""" if user and hasattr(user, 'branch') and user.branch: return user.branch.name return None def _get_client_ip(self, request): """Get client IP address from request""" if not request: return None 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 _sanitize_request_params(self, request): """Sanitize request parameters for logging""" if not request: return {} # Get query parameters params = dict(request.GET.items()) # Remove sensitive parameters sensitive_keys = ['password', 'token', 'api_key', 'secret'] for key in list(params.keys()): if any(sensitive in key.lower() for sensitive in sensitive_keys): params[key] = '[REDACTED]' return params def _update_access_pattern(self, user, request): """Update daily access pattern for user""" if not user: return try: today = timezone.now().date() current_time = timezone.now().time() pattern, created = DataAccessPattern.objects.get_or_create( user=user, date=today, defaults={ 'first_access': current_time, 'last_access': current_time, 'primary_ip': self._get_client_ip(request) if request else None } ) # Update metrics pattern.total_requests = F('total_requests') + 1 pattern.last_access = current_time if request: ip = self._get_client_ip(request) if ip and ip != pattern.primary_ip: pattern.unique_ip_addresses = F('unique_ip_addresses') + 1 # Check if after hours (before 7 AM or after 7 PM) if current_time < time(7, 0) or current_time > time(19, 0): pattern.after_hours_access = F('after_hours_access') + 1 pattern.save() # Check for anomalies self._check_access_anomalies(pattern) except Exception as e: logger.error(f"Failed to update access pattern: {e}") def _check_security_anomalies(self, user, action, request): """Check for security anomalies in user behavior""" if not user: return try: # Check for multiple failed login attempts if 'login' in action.lower() and 'failed' in action.lower(): self._check_failed_login_attempts(user, request) # Check for unusual access patterns self._check_unusual_access_patterns(user, request) except Exception as e: logger.error(f"Failed to check security anomalies: {e}") def _check_failed_login_attempts(self, user, request): """Check for multiple failed login attempts""" try: # Count failed login attempts in last hour one_hour_ago = timezone.now() - timedelta(hours=1) failed_attempts = EnhancedAuditLog.objects.filter( user=user, action__icontains='failed_login', timestamp__gte=one_hour_ago ).count() if failed_attempts >= 5: # Threshold for suspicious activity self._create_security_alert( alert_type='multiple_failed_logins', user=user, severity=SecurityEventSeverity.HIGH, title=f"Multiple Failed Login Attempts", description=f"User {user.get_full_name()} has {failed_attempts} failed login attempts in the last hour", request=request, additional_context={'failed_attempts': failed_attempts} ) except Exception as e: logger.error(f"Failed to check failed login attempts: {e}") def _check_unusual_access_patterns(self, user, request): """Check for unusual access patterns""" try: today = timezone.now().date() pattern = DataAccessPattern.objects.filter(user=user, date=today).first() if not pattern: return # Get user's typical patterns (last 30 days) thirty_days_ago = today - timedelta(days=30) historical_patterns = DataAccessPattern.objects.filter( user=user, date__gte=thirty_days_ago, date__lt=today ) if historical_patterns.exists(): avg_requests = historical_patterns.aggregate(avg=models.Avg('total_requests'))['avg'] or 0 # If today's requests are significantly higher than average if pattern.total_requests > avg_requests * 3: # 3x normal activity self._create_security_alert( alert_type='unusual_access_pattern', user=user, severity=SecurityEventSeverity.MEDIUM, title=f"Unusual Access Pattern Detected", description=f"User {user.get_full_name()} has unusually high activity today ({pattern.total_requests} requests vs {avg_requests:.1f} average)", request=request, additional_context={ 'current_requests': pattern.total_requests, 'average_requests': avg_requests } ) except Exception as e: logger.error(f"Failed to check unusual access patterns: {e}") def _handle_permission_denial(self, user, permission, request): """Handle permission denial events""" try: # Count recent permission denials one_hour_ago = timezone.now() - timedelta(hours=1) recent_denials = EnhancedAuditLog.objects.filter( user=user, permission_granted=False, timestamp__gte=one_hour_ago ).count() if recent_denials >= 10: # Threshold for suspicious activity self._create_security_alert( alert_type='permission_escalation', user=user, severity=SecurityEventSeverity.HIGH, title=f"Possible Permission Escalation Attempt", description=f"User {user.get_full_name()} has {recent_denials} permission denials in the last hour", request=request, additional_context={ 'recent_denials': recent_denials, 'latest_permission': permission } ) except Exception as e: logger.error(f"Failed to handle permission denial: {e}") def _check_bulk_access_pattern(self, user, model_class, action): """Check for bulk data access patterns""" try: # Count recent access to same model type one_hour_ago = timezone.now() - timedelta(hours=1) recent_access = EnhancedAuditLog.objects.filter( user=user, event_type=AuditEventType.DATA_ACCESS, timestamp__gte=one_hour_ago, additional_data__model_type=model_class.__name__ ).count() if recent_access >= 100: # Threshold for bulk access self._create_security_alert( alert_type='bulk_data_access', user=user, severity=SecurityEventSeverity.MEDIUM, title=f"Bulk Data Access Detected", description=f"User {user.get_full_name()} has accessed {recent_access} {model_class.__name__} records in the last hour", additional_context={ 'model_type': model_class.__name__, 'access_count': recent_access, 'action': action } ) except Exception as e: logger.error(f"Failed to check bulk access pattern: {e}") def _requires_approval(self, permission_name, old_value, new_value): """Check if permission change requires approval""" # Critical permissions that require approval critical_permissions = [ 'admin_access', 'delete_users', 'modify_permissions', 'system_settings', 'financial_data_access' ] return any(critical in permission_name.lower() for critical in critical_permissions) def _is_critical_permission(self, permission_name): """Check if permission is critical""" critical_permissions = [ 'admin', 'delete', 'system', 'financial', 'export_all' ] return any(critical in permission_name.lower() for critical in critical_permissions) def _create_security_alert(self, alert_type, user, severity, title, description, request=None, additional_context=None): """Create a security alert""" try: SecurityAlert.objects.create( alert_type=alert_type, user=user, severity=severity, title=title, description=description, ip_address=self._get_client_ip(request) if request else None, user_agent=request.META.get('HTTP_USER_AGENT', '') if request else None, additional_context=additional_context or {} ) except Exception as e: logger.error(f"Failed to create security alert: {e}") def _check_access_anomalies(self, pattern): """Check access pattern for anomalies""" try: anomaly_score = 0.0 anomaly_reasons = [] # Check for high request volume if pattern.total_requests > 1000: # High threshold anomaly_score += 0.3 anomaly_reasons.append('High request volume') # Check for after hours access if pattern.after_hours_access > 50: anomaly_score += 0.2 anomaly_reasons.append('Excessive after-hours access') # Check for multiple IP addresses if pattern.unique_ip_addresses > 5: anomaly_score += 0.2 anomaly_reasons.append('Multiple IP addresses') # Check for failed access attempts if pattern.failed_access_attempts > 20: anomaly_score += 0.3 anomaly_reasons.append('High failed access attempts') # Update pattern with anomaly information pattern.anomaly_score = anomaly_score pattern.anomaly_reasons = anomaly_reasons pattern.is_anomalous = anomaly_score > 0.5 pattern.save() # Create alert for high anomaly scores if anomaly_score > 0.7: self._create_security_alert( alert_type='unusual_access_pattern', user=pattern.user, severity=SecurityEventSeverity.HIGH, title=f"Anomalous Access Pattern Detected", description=f"User {pattern.user.get_full_name()} shows anomalous access pattern on {pattern.date}", additional_context={ 'anomaly_score': anomaly_score, 'reasons': anomaly_reasons, 'date': pattern.date.isoformat() } ) except Exception as e: logger.error(f"Failed to check access anomalies: {e}") # Global audit service instance audit_service = AuditService()