Add hook mechanism when a login attempt fails or succeeds

docs/settings.rst

@@ -34,6 +34,8 @@ Some of Simple JWT's behavior can be customized through settings variables in
       "USER_ID_FIELD": "id",
       "USER_ID_CLAIM": "user_id",
       "USER_AUTHENTICATION_RULE": "rest_framework_simplejwt.authentication.default_user_authentication_rule",
+      "ON_LOGIN_SUCCESS": "rest_framework_simplejwt.serializers.default_on_login_success",
+      "ON_LOGIN_FAILED": "rest_framework_simplejwt.serializers.default_on_login_failed",
 
       "AUTH_TOKEN_CLASSES": ("rest_framework_simplejwt.tokens.AccessToken",),
       "TOKEN_TYPE_CLAIM": "token_type",
@@ -223,6 +225,20 @@ to the callable as an argument. The default rule is to check that the ``is_activ
 flag is still ``True``. The callable must return a boolean, ``True`` if authorized,
 ``False`` otherwise resulting in a 401 status code.
 
+``ON_LOGIN_SUCCESS``
+----------------------------
+
+Callable to add logic whenever a login attempt succeeded. ``UPDATE_LAST_LOGIN``
+must be set to ``True``. The callable does not return anything.
+The default callable updates last_login field in the auth_user table upon login
+(TokenObtainPairView).
+
+``ON_LOGIN_FAILED``
+----------------------------
+
+Callable to add logic whenever a login attempt failed. The callable does not
+return anything. The default callable does nothing (``pass``)
+
 ``AUTH_TOKEN_CLASSES``
 ----------------------
 

rest_framework_simplejwt/serializers.py

@@ -1,11 +1,12 @@
 from typing import Any, Optional, TypeVar
 
 from django.conf import settings
-from django.contrib.auth import authenticate, get_user_model
+from django.contrib.auth import _clean_credentials, authenticate, get_user_model
 from django.contrib.auth.models import AbstractBaseUser, update_last_login
 from django.utils.translation import gettext_lazy as _
 from rest_framework import exceptions, serializers
 from rest_framework.exceptions import AuthenticationFailed, ValidationError
+from rest_framework.request import Request
 
 from .models import TokenUser
 from .settings import api_settings
@@ -54,6 +55,9 @@ def validate(self, attrs: dict[str, Any]) -> dict[Any, Any]:
         self.user = authenticate(**authenticate_kwargs)
 
         if not api_settings.USER_AUTHENTICATION_RULE(self.user):
+            api_settings.ON_LOGIN_FAILED(
+                _clean_credentials(attrs), self.context.get("request")
+            )
             raise exceptions.AuthenticationFailed(
                 self.error_messages["no_active_account"],
                 "no_active_account",
@@ -78,7 +82,7 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, str]:
         data["access"] = str(refresh.access_token)
 
         if api_settings.UPDATE_LAST_LOGIN:
-            update_last_login(None, self.user)
+            api_settings.ON_LOGIN_SUCCESS(self.user, self.context.get("request"))
 
         return data
 
@@ -94,7 +98,7 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, str]:
         data["token"] = str(token)
 
         if api_settings.UPDATE_LAST_LOGIN:
-            update_last_login(None, self.user)
+            api_settings.ON_LOGIN_SUCCESS(self.user, self.context.get("request"))
 
         return data
 
@@ -191,3 +195,11 @@ def validate(self, attrs: dict[str, Any]) -> dict[Any, Any]:
         except AttributeError:
             pass
         return {}
+
+
+def default_on_login_success(user: AuthUser, request: Optional[Request]) -> None:
+    update_last_login(None, user)
+
+
+def default_on_login_failed(credentials: dict, request: Optional[Request]) -> None:
+    pass

rest_framework_simplejwt/settings.py

@@ -29,6 +29,8 @@
     "USER_ID_FIELD": "id",
     "USER_ID_CLAIM": "user_id",
     "USER_AUTHENTICATION_RULE": "rest_framework_simplejwt.authentication.default_user_authentication_rule",
+    "ON_LOGIN_SUCCESS": "rest_framework_simplejwt.serializers.default_on_login_success",
+    "ON_LOGIN_FAILED": "rest_framework_simplejwt.serializers.default_on_login_failed",
     "AUTH_TOKEN_CLASSES": ("rest_framework_simplejwt.tokens.AccessToken",),
     "TOKEN_TYPE_CLAIM": "token_type",
     "JTI_CLAIM": "jti",
@@ -52,6 +54,8 @@
     "JSON_ENCODER",
     "TOKEN_USER_CLASS",
     "USER_AUTHENTICATION_RULE",
+    "ON_LOGIN_SUCCESS",
+    "ON_LOGIN_FAILED",
 )
 
 REMOVED_SETTINGS = (

tests/test_views.py

@@ -1,4 +1,5 @@
 from datetime import timedelta
+from unittest import mock
 from unittest.mock import patch
 
 from django.contrib.auth import get_user_model
@@ -105,6 +106,25 @@ def test_update_last_login_updated(self):
         self.assertIsNotNone(user.last_login)
         self.assertGreaterEqual(timezone.now(), user.last_login)
 
+    def test_on_login_failed_is_called(self):
+        # Patch the ON_LOGIN_FAILED setting
+        with mock.patch(
+            "rest_framework_simplejwt.settings.api_settings.ON_LOGIN_FAILED"
+        ) as mocked_hook:
+            self.test_credentials_wrong()
+            mocked_hook.assert_called_once()
+
+            # Optional: check exact arguments
+            args, kwargs = mocked_hook.call_args
+            credentials, request = args
+            self.assertEqual(
+                credentials,
+                {
+                    User.USERNAME_FIELD: self.username,
+                    "password": "********************",
+                },
+            )
+
 
 class TestTokenRefreshView(APIViewTestCase):
     view_name = "token_refresh"