Backend: Enhance password reset logic in CustomPasswordResetSerializer (#5054)

* Enhance password reset logic in CustomPasswordResetSerializer to include checks for unverified emails and bounced email addresses. Update unit tests to cover new validation scenarios, ensuring robust error handling for password reset requests.

* Refactor password reset logic in CustomPasswordResetSerializer to utilize get_user_by_email for case-insensitive email lookup. Update unit tests to ensure proper validation and error handling for email variations. Enhance get_user_by_email utility to support case-insensitive queries and add tests for handling duplicate emails with differing cases.

Author: RishabhJain2018

apps/accounts/serializers.py

@@ -345,15 +345,30 @@ class CustomPasswordResetSerializer(PasswordResetSerializer):
     def get_email_options(self):
         try:
             user = get_user_by_email(self.data["email"])
-            if not user.is_active:
-                raise ValidationError(
-                    {
-                        "details": "Account is not active. Please contact the administrator."
-                    }
-                )
-            else:
-                return super().get_email_options()
         except User.DoesNotExist:
             raise ValidationError(
                 {"details": "User with the given email does not exist."}
             )
+        if not user.is_active:
+            raise ValidationError(
+                {
+                    "details": "Account is not active. Please contact the administrator."
+                }
+            )
+        if hasattr(user, "profile") and user.profile.email_bounced:
+            raise ValidationError(
+                {
+                    "details": "This email address has bounced and cannot receive password reset emails."
+                }
+            )
+        if not EmailAddress.objects.filter(
+            user=user,
+            email__iexact=self.data["email"],
+            verified=True,
+        ).exists():
+            raise ValidationError(
+                {
+                    "details": "Email address is not verified. Please verify your email before resetting password."
+                }
+            )
+        return super().get_email_options()

apps/base/utils.py

@@ -32,7 +32,11 @@ def get_user_by_email(email):
     If duplicate-email users still exist (pre-merge), the oldest account
     (earliest date_joined) is returned deterministically.
     """
-    user = User.objects.filter(email=email).order_by("date_joined").first()
+    user = (
+        User.objects.filter(email__iexact=email)
+        .order_by("date_joined")
+        .first()
+    )
     if user is None:
         raise User.DoesNotExist("User matching query does not exist.")
     return user

tests/unit/accounts/test_serializers.py

@@ -5,6 +5,7 @@
     CustomPasswordResetSerializer,
     ProfileSerializer,
 )
+from allauth.account.models import EmailAddress
 from challenges.models import Challenge
 from django.contrib.auth import get_user_model
 from django.test import TestCase
@@ -22,22 +23,46 @@ def setUp(self):
             email="active@example.com",
             password="password",
         )
+        self.unverified_user = self.user_model.objects.create_user(
+            username="unverified_user",
+            email="unverified@example.com",
+            password="password",
+        )
         self.inactive_user = self.user_model.objects.create_user(
             username="inactive_user",
             email="inactive@example.com",
             password="password",
         )
         self.inactive_user.is_active = False
         self.inactive_user.save()
+        EmailAddress.objects.create(
+            user=self.active_user,
+            email=self.active_user.email,
+            primary=True,
+            verified=True,
+        )
+        EmailAddress.objects.create(
+            user=self.unverified_user,
+            email=self.unverified_user.email,
+            primary=True,
+            verified=False,
+        )
+        EmailAddress.objects.create(
+            user=self.inactive_user,
+            email=self.inactive_user.email,
+            primary=True,
+            verified=True,
+        )
 
     def test_get_email_options_try_block(self):
         serializer = CustomPasswordResetSerializer(
             data={"email": self.active_user.email}
         )
         self.assertEqual(serializer.is_valid(), True)
-        expected_email_options = super(
-            CustomPasswordResetSerializer, serializer
-        ).get_email_options()
+        base_class = serializer.__class__.__mro__[1]
+        base_serializer = base_class(data={"email": self.active_user.email})
+        self.assertEqual(base_serializer.is_valid(), True)
+        expected_email_options = base_serializer.get_email_options()
         self.assertEqual(
             serializer.get_email_options(), expected_email_options
         )
@@ -68,6 +93,49 @@ def test_get_email_options_inactive_user(self):
             },
         )
 
+    def test_get_email_options_unverified_user(self):
+        serializer = CustomPasswordResetSerializer(
+            data={"email": self.unverified_user.email}
+        )
+        serializer.is_valid()  # Ensure data is validated
+        with self.assertRaises(ValidationError) as e:
+            serializer.get_email_options()
+        self.assertEqual(
+            e.exception.detail,
+            {
+                "details": "Email address is not verified. Please verify your email before resetting password."
+            },
+        )
+
+    def test_get_email_options_bounced_email_user(self):
+        self.active_user.profile.email_bounced = True
+        self.active_user.profile.save(update_fields=["email_bounced"])
+        serializer = CustomPasswordResetSerializer(
+            data={"email": self.active_user.email}
+        )
+        serializer.is_valid()  # Ensure data is validated
+        with self.assertRaises(ValidationError) as e:
+            serializer.get_email_options()
+        self.assertEqual(
+            e.exception.detail,
+            {
+                "details": "This email address has bounced and cannot receive password reset emails."
+            },
+        )
+
+    def test_get_email_options_case_insensitive_email_lookup(self):
+        serializer = CustomPasswordResetSerializer(
+            data={"email": "ACTIVE@EXAMPLE.COM"}
+        )
+        self.assertEqual(serializer.is_valid(), True)
+        base_class = serializer.__class__.__mro__[1]
+        base_serializer = base_class(data={"email": "ACTIVE@EXAMPLE.COM"})
+        self.assertEqual(base_serializer.is_valid(), True)
+        expected_email_options = base_serializer.get_email_options()
+        self.assertEqual(
+            serializer.get_email_options(), expected_email_options
+        )
+
 
 class TestProfileSerializer(TestCase):
     def setUp(self):

tests/unit/accounts/test_views.py

@@ -321,3 +321,114 @@ def test_integrity_error_non_email_reraised(self):
                     },
                     format="json",
                 )
+
+
+class PasswordResetViewTest(APITestCase):
+    def setUp(self):
+        self.client = APIClient(enforce_csrf_checks=True)
+        self.url = "/api/auth/password/reset/"
+
+        self.verified_user = User.objects.create_user(
+            username="verified_user",
+            email="verified@example.com",
+            password="password",
+        )
+        EmailAddress.objects.create(
+            user=self.verified_user,
+            email=self.verified_user.email,
+            primary=True,
+            verified=True,
+        )
+
+        self.unverified_user = User.objects.create_user(
+            username="unverified_user",
+            email="unverified@example.com",
+            password="password",
+        )
+        EmailAddress.objects.create(
+            user=self.unverified_user,
+            email=self.unverified_user.email,
+            primary=True,
+            verified=False,
+        )
+
+        self.inactive_user = User.objects.create_user(
+            username="inactive_user",
+            email="inactive@example.com",
+            password="password",
+            is_active=False,
+        )
+        EmailAddress.objects.create(
+            user=self.inactive_user,
+            email=self.inactive_user.email,
+            primary=True,
+            verified=True,
+        )
+
+    @patch("django.contrib.auth.forms.PasswordResetForm.save")
+    def test_password_reset_verified_active_user_success(self, mock_save):
+        response = self.client.post(
+            self.url, {"email": self.verified_user.email}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        mock_save.assert_called_once()
+
+    @patch("django.contrib.auth.forms.PasswordResetForm.save")
+    def test_password_reset_verified_active_user_case_insensitive_email(
+        self, mock_save
+    ):
+        response = self.client.post(
+            self.url, {"email": "VERIFIED@EXAMPLE.COM"}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        mock_save.assert_called_once()
+
+    @patch("django.contrib.auth.forms.PasswordResetForm.save")
+    def test_password_reset_unverified_user_returns_400(self, mock_save):
+        response = self.client.post(
+            self.url, {"email": self.unverified_user.email}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data["details"],
+            "Email address is not verified. Please verify your email before resetting password.",
+        )
+        mock_save.assert_not_called()
+
+    @patch("django.contrib.auth.forms.PasswordResetForm.save")
+    def test_password_reset_bounced_email_returns_400(self, mock_save):
+        self.verified_user.profile.email_bounced = True
+        self.verified_user.profile.save(update_fields=["email_bounced"])
+        response = self.client.post(
+            self.url, {"email": self.verified_user.email}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data["details"],
+            "This email address has bounced and cannot receive password reset emails.",
+        )
+        mock_save.assert_not_called()
+
+    @patch("django.contrib.auth.forms.PasswordResetForm.save")
+    def test_password_reset_inactive_user_returns_400(self, mock_save):
+        response = self.client.post(
+            self.url, {"email": self.inactive_user.email}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data["details"],
+            "Account is not active. Please contact the administrator.",
+        )
+        mock_save.assert_not_called()
+
+    @patch("django.contrib.auth.forms.PasswordResetForm.save")
+    def test_password_reset_nonexistent_user_returns_400(self, mock_save):
+        response = self.client.post(
+            self.url, {"email": "nonexistent@example.com"}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data["details"],
+            "User with the given email does not exist.",
+        )
+        mock_save.assert_not_called()

tests/unit/base/test_get_user_by_email.py

@@ -33,6 +33,18 @@ def test_returns_user_for_unique_email(self):
         result = get_user_by_email("alice@example.com")
         self.assertEqual(result.pk, user.pk)
 
+    def test_returns_user_case_insensitive(self):
+        user = User.objects.create_user(
+            username="carol", email="carol@example.com", password="password"
+        )
+        for variant in (
+            "CAROL@EXAMPLE.COM",
+            "Carol@Example.Com",
+            "carol@EXAMPLE.com",
+        ):
+            result = get_user_by_email(variant)
+            self.assertEqual(result.pk, user.pk)
+
     def test_raises_for_nonexistent_email(self):
         with self.assertRaises(User.DoesNotExist):
             get_user_by_email("ghost@example.com")
@@ -45,8 +57,9 @@ def setUp(self):
         _drop_email_unique_constraint()
 
     def tearDown(self):
-        # Remove duplicate users before re-adding unique constraint
-        User.objects.filter(username__in=("bob_old", "bob_new")).delete()
+        User.objects.filter(
+            username__in=("bob_old", "bob_new", "eve_lower", "eve_upper")
+        ).delete()
         _add_email_unique_constraint()
 
     def test_returns_oldest_when_duplicates_exist(self):
@@ -58,3 +71,13 @@ def test_returns_oldest_when_duplicates_exist(self):
         )
         result = get_user_by_email("bob@example.com")
         self.assertEqual(result.pk, older.pk)
+
+    def test_returns_oldest_when_case_differing_duplicates_exist(self):
+        older = User.objects.create_user(
+            username="eve_lower", email="eve@example.com", password="password"
+        )
+        User.objects.create_user(
+            username="eve_upper", email="EVE@EXAMPLE.COM", password="password"
+        )
+        result = get_user_by_email("Eve@Example.com")
+        self.assertEqual(result.pk, older.pk)