# -*- coding: utf-8 -*- """ Adtlas Accounts Signals This module contains Django signals for the accounts app to handle automatic actions when user-related events occur. Features: - Auto-create user profiles - Log user activities - Manage user sessions - Handle role assignments - Password change tracking Author: Adtlas Development Team Version: 2.0.0 Last Updated: 2025-07-08 """ from django.db.models.signals import post_save, pre_save, post_delete from django.contrib.auth.signals import user_logged_in, user_logged_out, user_login_failed from django.dispatch import receiver from django.utils import timezone from django.contrib.sessions.models import Session from django.contrib.auth import get_user_model from .models import User, Profile, UserActivity, UserSession, UserRole User = get_user_model() @receiver(post_save, sender=User) def create_user_profile(sender, instance, created, **kwargs): """ Create a Profile when a new user is created. """ if created: Profile.objects.get_or_create(user=instance) @receiver(post_save, sender=User) def save_user_profile(sender, instance, **kwargs): """ Save the Profile when the user is saved. """ if hasattr(instance, 'profile'): instance.profile.save() @receiver(pre_save, sender=User) def track_password_change(sender, instance, **kwargs): """ Track when a user's password is changed. """ if instance.pk: try: old_instance = User.objects.get(pk=instance.pk) if old_instance.password != instance.password: instance.password_changed_at = timezone.now() # Reset failed login attempts on password change instance.failed_login_attempts = 0 instance.account_locked_until = None except User.DoesNotExist: pass @receiver(user_logged_in) def log_user_login(sender, request, user, **kwargs): """ Log user login activity and create/update session tracking. """ # Get client IP address x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip_address = x_forwarded_for.split(',')[0] else: ip_address = request.META.get('REMOTE_ADDR') # Get user agent user_agent = request.META.get('HTTP_USER_AGENT', '') # Update user's last activity user.last_activity = timezone.now() user.failed_login_attempts = 0 # Reset on successful login user.save(update_fields=['last_activity', 'failed_login_attempts']) # Create user activity log UserActivity.objects.create( user=user, action='login', ip_address=ip_address, user_agent=user_agent, details={'login_method': 'web'} ) # Create or update session tracking if hasattr(request, 'session'): session_key = request.session.session_key if session_key: session = Session.objects.filter(session_key=session_key).first() expires_at = session.expire_date if session else timezone.now() + timezone.timedelta(days=1) UserSession.objects.update_or_create( session_key=session_key, defaults={ 'user': user, 'ip_address': ip_address, 'user_agent': user_agent, 'is_active': True, 'last_activity': timezone.now(), 'expires_at': expires_at } ) @receiver(user_logged_out) def log_user_logout(sender, request, user, **kwargs): """ Log user logout activity and update session tracking. """ if user: # Get client IP address x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip_address = x_forwarded_for.split(',')[0] else: ip_address = request.META.get('REMOTE_ADDR') # Get user agent user_agent = request.META.get('HTTP_USER_AGENT', '') # Create user activity log UserActivity.objects.create( user=user, action='logout', ip_address=ip_address, user_agent=user_agent, details={'logout_method': 'web'} ) # Mark session as inactive if hasattr(request, 'session') and request.session.session_key: UserSession.objects.filter( session_key=request.session.session_key ).update(is_active=False) @receiver(user_login_failed) def log_failed_login(sender, credentials, request, **kwargs): """ Log failed login attempts and implement account lockout. """ # Get client IP address x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip_address = x_forwarded_for.split(',')[0] else: ip_address = request.META.get('REMOTE_ADDR') # Get user agent user_agent = request.META.get('HTTP_USER_AGENT', '') # Try to find the user by email/username email = credentials.get('email') or credentials.get('username') if email: try: user = User.objects.get(email=email) # Increment failed login attempts user.failed_login_attempts += 1 # Lock account after 5 failed attempts for 30 minutes if user.failed_login_attempts >= 5: user.account_locked_until = timezone.now() + timezone.timedelta(minutes=30) user.save(update_fields=['failed_login_attempts', 'account_locked_until']) # Create user activity log UserActivity.objects.create( user=user, action='login', ip_address=ip_address, user_agent=user_agent, details={ 'success': False, 'reason': 'invalid_credentials', 'failed_attempts': user.failed_login_attempts } ) except User.DoesNotExist: # Skip logging for unknown emails since UserActivity requires a valid user # This prevents IntegrityError when user=None is passed # Consider implementing a separate SecurityLog model for tracking # failed attempts with non-existent emails if needed pass @receiver(post_save, sender=UserRole) def log_role_assignment(sender, instance, created, **kwargs): """ Log when roles are assigned to users. """ if created: UserActivity.objects.create( user=instance.user, action='admin', object_type='UserRole', object_id=str(instance.id), object_repr=str(instance), details={ 'action': 'role_assigned', 'role': instance.role.name, 'role_code': instance.role.code, 'assigned_by': instance.assigned_by.email if instance.assigned_by else None } ) @receiver(post_delete, sender=UserRole) def log_role_removal(sender, instance, **kwargs): """ Log when roles are removed from users. """ UserActivity.objects.create( user=instance.user, action='admin', object_type='UserRole', object_id=str(instance.id), object_repr=str(instance), details={ 'action': 'role_removed', 'role': instance.role.name, 'role_code': instance.role.code } ) def cleanup_expired_sessions(): """ Utility function to clean up expired sessions. This can be called by a management command or celery task. """ expired_sessions = UserSession.objects.filter( expires_at__lt=timezone.now(), is_active=True ) count = expired_sessions.count() expired_sessions.update(is_active=False) return count def cleanup_old_activities(days=90): """ Utility function to clean up old activity logs. Keep only activities from the last X days. """ cutoff_date = timezone.now() - timezone.timedelta(days=days) deleted_count, _ = UserActivity.objects.filter( created_at__lt=cutoff_date ).delete() return deleted_count