# -*- 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