login/logout first pass

api/audit/tasks.py

@@ -173,3 +173,82 @@ def create_segment_priorities_changed_audit_log(
         related_object_id=feature.pk,
         related_object_type=RelatedObjectType.FEATURE.name,
     )
+
+
+@register_task_handler()
+def create_audit_log_user_logged_in(user_id: int):
+    if not (user := get_user_model().objects.filter(id=user_id).first()):
+        logger.warning(
+            f"User with id {user_id} not found. Audit log for user logged in not created."
+        )
+        return
+
+    user = typing.cast(_AbstractBaseAuditableModel, user)
+    log_message = (
+        f"{RelatedObjectType.USER.value} logged in: {user.get_audit_log_identity()}"
+    )
+
+    audit_logs = [
+        AuditLog(
+            organisation=organisation,
+            related_object_id=user.pk,
+            related_object_type=RelatedObjectType.USER.name,
+            log=log_message,
+            is_system_event=True,
+        )
+        for organisation in user._get_organisations()
+    ]
+    AuditLog.objects.bulk_create(audit_logs)
+
+
+@register_task_handler()
+def create_audit_log_user_logged_out(user_id: int):
+    if not (user := get_user_model().objects.filter(id=user_id).first()):
+        logger.warning(
+            f"User with id {user_id} not found. Audit log for user logged out not created."
+        )
+        return
+
+    user = typing.cast(_AbstractBaseAuditableModel, user)
+    log_message = (
+        f"{RelatedObjectType.USER.value} logged out: {user.get_audit_log_identity()}"
+    )
+
+    audit_logs = [
+        AuditLog(
+            organisation=organisation,
+            related_object_id=user.pk,
+            related_object_type=RelatedObjectType.USER.name,
+            log=log_message,
+            is_system_event=True,
+        )
+        for organisation in user._get_organisations()
+    ]
+    AuditLog.objects.bulk_create(audit_logs)
+
+
+@register_task_handler()
+def create_audit_log_user_login_failed(
+    credentials: dict, codes: list[str] | None = None
+):
+    if not (username := credentials.get("username")):
+        return
+    if not (user := get_user_model().objects.get_by_natural_key(username)):
+        return
+    if not isinstance(user, _AbstractBaseAuditableModel):
+        return
+
+    reason = ",".join(codes) if codes else "password"
+    log_message = f"{RelatedObjectType.USER.value} login failed ({reason}): {user.get_audit_log_identity()}"
+
+    audit_logs = [
+        AuditLog(
+            organisation=organisation,
+            related_object_id=user.pk,
+            related_object_type=RelatedObjectType.USER.name,
+            log=log_message,
+            is_system_event=True,
+        )
+        for organisation in user._get_organisations()
+    ]
+    AuditLog.objects.bulk_create(audit_logs)

api/core/models.py

@@ -164,6 +164,7 @@ def _get_environment(self) -> Environment | None:
         return None
 
 
+# TODO #2797 later: get IP address from request
 def get_history_user(
     instance: typing.Any, request: HttpRequest
 ) -> typing.Optional["FFAdminUser"]:

api/custom_auth/views.py

@@ -1,4 +1,4 @@
-from django.contrib.auth import user_logged_out
+from django.contrib.auth import user_logged_out, user_login_failed
 from django.utils.decorators import method_decorator
 from djoser.views import UserViewSet
 from drf_yasg.utils import swagger_auto_schema
@@ -7,6 +7,7 @@
 from rest_framework.decorators import action, api_view, permission_classes
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
+from rest_framework.serializers import ValidationError
 from rest_framework.throttling import ScopedRateThrottle
 from trench.views.authtoken import (
     AuthTokenLoginOrRequestMFACode,
@@ -19,6 +20,22 @@
 from .models import UserPasswordResetRequest
 
 
+class CustomCodeLoginSerializer(AuthTokenLoginWithMFACode.serializer_class):
+    def validate(self, attrs):
+        try:
+            return super().validate(attrs)
+        except ValidationError as e:
+            # signal non-password login failure
+            if user := getattr(self, "user", None):
+                user_login_failed.send(
+                    self.__module__,
+                    credentials={"username": user.natural_key()},
+                    request=self.context["request"],
+                    codes=e.get_codes(),
+                )
+            raise
+
+
 class CustomAuthTokenLoginOrRequestMFACode(AuthTokenLoginOrRequestMFACode):
     """
     Class to handle throttling for login requests
@@ -33,6 +50,7 @@ class CustomAuthTokenLoginWithMFACode(AuthTokenLoginWithMFACode):
     Override class to add throttling
     """
 
+    serializer_class = CustomCodeLoginSerializer
     throttle_classes = [ScopedRateThrottle]
     throttle_scope = "mfa_code"
 

api/users/signals.py

@@ -1,10 +1,20 @@
 import warnings
 
 from django.conf import settings
+from django.contrib.auth.signals import (
+    user_logged_in,
+    user_logged_out,
+    user_login_failed,
+)
 from django.db.models.signals import post_migrate, post_save
 from django.dispatch import receiver
 from django.urls import reverse
 
+from audit.tasks import (
+    create_audit_log_user_logged_in,
+    create_audit_log_user_logged_out,
+    create_audit_log_user_login_failed,
+)
 from integrations.lead_tracking.pipedrive.lead_tracker import (
     PipedriveLeadTracker,
 )
@@ -49,3 +59,23 @@ def send_warning_email(sender, instance, created, **kwargs):
                 instance._initial_state["email"],
             )
         )
+
+
+@receiver(user_logged_in, sender=FFAdminUser)
+def signal_audit_log_user_logged_in(signal, sender, user, request):
+    # TODO #2797 later: get IP address from request
+    create_audit_log_user_logged_in.delay(args=(user.pk,))
+
+
+@receiver(user_logged_out, sender=FFAdminUser)
+def signal_audit_log_user_logged_out(signal, sender, user, request):
+    # TODO #2797 later: get IP address from request
+    create_audit_log_user_logged_out.delay(args=(user.pk,))
+
+
+@receiver(user_login_failed)
+def signal_audit_log_user_login_failed(signal, sender, credentials, request, **kwargs):
+    # TODO #2797 later: get IP address from request
+    create_audit_log_user_login_failed.delay(
+        args=(credentials, kwargs.get("codes", None))
+    )