# -*- coding: utf-8 -*- """ Activities Models This module defines Django models for activity logging and audit trails. These models track user actions, system events, and administrative operations for security and compliance purposes. Models: - ActivityType: Categorization of different activity types - Activity: Main activity logging model - AuditLog: Detailed audit logging with before/after states - SecurityEvent: Security-related event tracking Author: AdTlas Development Team Version: 1.0.0 Last Updated: 2024 """ from django.db import models from django.contrib.auth import get_user_model from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.fields import GenericForeignKey from django.utils.translation import gettext_lazy as _ from django.utils import timezone from django.core.serializers.json import DjangoJSONEncoder import json from apps.core.models import BaseModel User = get_user_model() class ActivityType(BaseModel): """ Model for categorizing different types of activities. This model defines categories for activities to help with filtering, reporting, and analysis of user actions. Attributes: name (str): The name of the activity type code (str): Unique code for the activity type description (str): Description of the activity type is_security_related (bool): Whether this activity type is security-related retention_days (int): How long to keep activities of this type is_active (bool): Whether this activity type is currently active """ name = models.CharField( _('Name'), max_length=100, help_text=_('Human-readable name for the activity type') ) code = models.CharField( _('Code'), max_length=50, unique=True, help_text=_('Unique code for the activity type') ) description = models.TextField( _('Description'), blank=True, help_text=_('Detailed description of the activity type') ) is_security_related = models.BooleanField( _('Security Related'), default=False, help_text=_('Whether this activity type is security-related') ) retention_days = models.PositiveIntegerField( _('Retention Days'), default=365, help_text=_('Number of days to retain activities of this type') ) is_active = models.BooleanField( _('Active'), default=True, help_text=_('Whether this activity type is currently active') ) class Meta: verbose_name = _('Activity Type') verbose_name_plural = _('Activity Types') ordering = ['name'] indexes = [ models.Index(fields=['code']), models.Index(fields=['is_security_related']), ] def __str__(self): """Return string representation of the activity type.""" return f"{self.name} ({self.code})" @classmethod def get_or_create_type(cls, code, name=None, **kwargs): """ Get or create an activity type by code. Args: code (str): The activity type code name (str, optional): The activity type name **kwargs: Additional fields for creation Returns: tuple: (ActivityType instance, created boolean) """ defaults = {'name': name or code.replace('_', ' ').title()} defaults.update(kwargs) return cls.objects.get_or_create(code=code, defaults=defaults) class ActivityManager(models.Manager): """ Custom manager for Activity model with useful query methods. """ def for_user(self, user): """ Get activities for a specific user. Args: user: The user instance Returns: QuerySet: Activities for the user """ return self.filter(user=user) def security_events(self): """ Get security-related activities. Returns: QuerySet: Security-related activities """ return self.filter(activity_type__is_security_related=True) def recent(self, days=7): """ Get recent activities within specified days. Args: days (int): Number of days to look back Returns: QuerySet: Recent activities """ since = timezone.now() - timezone.timedelta(days=days) return self.filter(created_at__gte=since) def by_action(self, action): """ Get activities by action type. Args: action (str): The action type Returns: QuerySet: Activities with the specified action """ return self.filter(action=action) class Activity(BaseModel): """ Model for logging user and system activities. This model tracks all significant actions performed by users and the system for audit trails and security monitoring. Attributes: user (User): The user who performed the activity (nullable for system activities) activity_type (ActivityType): The type/category of activity action (str): The specific action performed description (str): Human-readable description of the activity ip_address (str): IP address from which the activity originated user_agent (str): User agent string from the request session_key (str): Session key for tracking user sessions extra_data (dict): Additional data related to the activity content_type (ContentType): Content type of related object object_id (int): ID of related object content_object (Model): Generic foreign key to related object success (bool): Whether the activity was successful error_message (str): Error message if activity failed """ # Activity classification ACTION_CHOICES = [ ('login', _('Login')), ('logout', _('Logout')), ('register', _('Register')), ('password_change', _('Password Change')), ('password_reset', _('Password Reset')), ('email_verify', _('Email Verification')), ('profile_update', _('Profile Update')), ('create', _('Create')), ('update', _('Update')), ('delete', _('Delete')), ('view', _('View')), ('download', _('Download')), ('upload', _('Upload')), ('export', _('Export')), ('import', _('Import')), ('admin_action', _('Admin Action')), ('api_call', _('API Call')), ('security_event', _('Security Event')), ('system_event', _('System Event')), ('other', _('Other')), ] user = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='activities', verbose_name=_('User'), help_text=_('User who performed the activity') ) activity_type = models.ForeignKey( ActivityType, on_delete=models.CASCADE, related_name='activities', verbose_name=_('Activity Type'), help_text=_('Type/category of the activity') ) action = models.CharField( _('Action'), max_length=50, choices=ACTION_CHOICES, help_text=_('The specific action performed') ) description = models.TextField( _('Description'), help_text=_('Human-readable description of the activity') ) ip_address = models.GenericIPAddressField( _('IP Address'), null=True, blank=True, help_text=_('IP address from which the activity originated') ) user_agent = models.TextField( _('User Agent'), blank=True, help_text=_('User agent string from the request') ) session_key = models.CharField( _('Session Key'), max_length=40, blank=True, help_text=_('Session key for tracking user sessions') ) extra_data = models.JSONField( _('Extra Data'), default=dict, blank=True, encoder=DjangoJSONEncoder, help_text=_('Additional data related to the activity') ) # Generic foreign key for related objects content_type = models.ForeignKey( ContentType, on_delete=models.CASCADE, null=True, blank=True, verbose_name=_('Content Type') ) object_id = models.PositiveIntegerField( _('Object ID'), null=True, blank=True ) content_object = GenericForeignKey('content_type', 'object_id') success = models.BooleanField( _('Success'), default=True, help_text=_('Whether the activity was successful') ) error_message = models.TextField( _('Error Message'), blank=True, help_text=_('Error message if activity failed') ) objects = ActivityManager() class Meta: verbose_name = _('Activity') verbose_name_plural = _('Activities') ordering = ['-created_at'] indexes = [ models.Index(fields=['user', '-created_at']), models.Index(fields=['action', '-created_at']), models.Index(fields=['ip_address', '-created_at']), models.Index(fields=['success', '-created_at']), models.Index(fields=['content_type', 'object_id']), ] def __str__(self): """Return string representation of the activity.""" user_str = self.user.email if self.user else 'System' return f"{user_str} - {self.get_action_display()} - {self.created_at}" @classmethod def log_activity(cls, user=None, action='other', description='', ip_address=None, user_agent='', session_key='', content_object=None, extra_data=None, success=True, error_message='', activity_type_code='general'): """ Convenience method to log an activity. Args: user: The user performing the activity action (str): The action being performed description (str): Description of the activity ip_address (str): IP address of the user user_agent (str): User agent string session_key (str): Session key content_object: Related object extra_data (dict): Additional data success (bool): Whether the activity was successful error_message (str): Error message if failed activity_type_code (str): Activity type code Returns: Activity: The created activity instance """ # Get or create activity type activity_type, _ = ActivityType.get_or_create_type(activity_type_code) # Prepare content type and object id content_type = None object_id = None if content_object: content_type = ContentType.objects.get_for_model(content_object) object_id = content_object.pk return cls.objects.create( user=user, activity_type=activity_type, action=action, description=description, ip_address=ip_address, user_agent=user_agent, session_key=session_key, content_type=content_type, object_id=object_id, extra_data=extra_data or {}, success=success, error_message=error_message ) def get_related_object_display(self): """ Get a display representation of the related object. Returns: str: String representation of the related object """ if self.content_object: return str(self.content_object) return _('No related object') def to_dict(self): """ Convert activity to dictionary for API responses. Returns: dict: Activity data as dictionary """ return { 'id': self.id, 'user': self.user.email if self.user else None, 'action': self.action, 'action_display': self.get_action_display(), 'description': self.description, 'ip_address': self.ip_address, 'success': self.success, 'created_at': self.created_at.isoformat(), 'related_object': self.get_related_object_display(), 'extra_data': self.extra_data } class AuditLog(BaseModel): """ Model for detailed audit logging with before/after states. This model provides detailed audit trails for model changes, storing the before and after states of objects. Attributes: user (User): The user who made the change action (str): The type of change (create, update, delete) model_name (str): Name of the model that was changed object_id (str): ID of the object that was changed object_repr (str): String representation of the object changes (dict): Dictionary of field changes before_state (dict): State of the object before change after_state (dict): State of the object after change ip_address (str): IP address of the user user_agent (str): User agent string """ ACTION_CHOICES = [ ('create', _('Create')), ('update', _('Update')), ('delete', _('Delete')), ] user = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='audit_logs', verbose_name=_('User') ) action = models.CharField( _('Action'), max_length=10, choices=ACTION_CHOICES ) model_name = models.CharField( _('Model Name'), max_length=100 ) object_id = models.CharField( _('Object ID'), max_length=255 ) object_repr = models.CharField( _('Object Representation'), max_length=255 ) changes = models.JSONField( _('Changes'), default=dict, encoder=DjangoJSONEncoder, help_text=_('Dictionary of field changes') ) before_state = models.JSONField( _('Before State'), default=dict, encoder=DjangoJSONEncoder, help_text=_('State of the object before change') ) after_state = models.JSONField( _('After State'), default=dict, encoder=DjangoJSONEncoder, help_text=_('State of the object after change') ) ip_address = models.GenericIPAddressField( _('IP Address'), null=True, blank=True ) user_agent = models.TextField( _('User Agent'), blank=True ) class Meta: verbose_name = _('Audit Log') verbose_name_plural = _('Audit Logs') ordering = ['-created_at'] indexes = [ models.Index(fields=['user', '-created_at']), models.Index(fields=['model_name', '-created_at']), models.Index(fields=['action', '-created_at']), models.Index(fields=['object_id', '-created_at']), ] def __str__(self): """Return string representation of the audit log.""" user_str = self.user.email if self.user else 'System' return f"{user_str} {self.action} {self.model_name} {self.object_id}" class SecurityEvent(BaseModel): """ Model for tracking security-related events. This model specifically tracks security events like failed logins, suspicious activities, and security policy violations. Attributes: event_type (str): Type of security event severity (str): Severity level of the event user (User): User associated with the event (if any) ip_address (str): IP address where event originated user_agent (str): User agent string description (str): Description of the security event details (dict): Additional details about the event resolved (bool): Whether the security event has been resolved resolved_by (User): User who resolved the event resolved_at (datetime): When the event was resolved resolution_notes (str): Notes about the resolution """ EVENT_TYPE_CHOICES = [ ('failed_login', _('Failed Login')), ('account_lockout', _('Account Lockout')), ('suspicious_activity', _('Suspicious Activity')), ('unauthorized_access', _('Unauthorized Access')), ('data_breach', _('Data Breach')), ('malware_detected', _('Malware Detected')), ('policy_violation', _('Policy Violation')), ('privilege_escalation', _('Privilege Escalation')), ('other', _('Other')), ] SEVERITY_CHOICES = [ ('low', _('Low')), ('medium', _('Medium')), ('high', _('High')), ('critical', _('Critical')), ] event_type = models.CharField( _('Event Type'), max_length=50, choices=EVENT_TYPE_CHOICES ) severity = models.CharField( _('Severity'), max_length=20, choices=SEVERITY_CHOICES, default='medium' ) user = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='security_events', verbose_name=_('User') ) ip_address = models.GenericIPAddressField( _('IP Address'), null=True, blank=True ) user_agent = models.TextField( _('User Agent'), blank=True ) description = models.TextField( _('Description') ) details = models.JSONField( _('Details'), default=dict, encoder=DjangoJSONEncoder ) resolved = models.BooleanField( _('Resolved'), default=False ) resolved_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='resolved_security_events', verbose_name=_('Resolved By') ) resolved_at = models.DateTimeField( _('Resolved At'), null=True, blank=True ) resolution_notes = models.TextField( _('Resolution Notes'), blank=True ) class Meta: verbose_name = _('Security Event') verbose_name_plural = _('Security Events') ordering = ['-created_at'] indexes = [ models.Index(fields=['event_type', '-created_at']), models.Index(fields=['severity', '-created_at']), models.Index(fields=['resolved', '-created_at']), models.Index(fields=['user', '-created_at']), models.Index(fields=['ip_address', '-created_at']), ] def __str__(self): """Return string representation of the security event.""" return f"{self.get_event_type_display()} - {self.get_severity_display()}" def resolve(self, resolved_by, resolution_notes=''): """ Mark the security event as resolved. Args: resolved_by (User): User resolving the event resolution_notes (str): Notes about the resolution """ self.resolved = True self.resolved_by = resolved_by self.resolved_at = timezone.now() self.resolution_notes = resolution_notes self.save()