feat(socialaccount): Email authentication based on user.email

allauth/account/utils.py

@@ -555,9 +555,7 @@ def assess_unique_email(email) -> Optional[bool]:
     False -- email is already in use
     None -- email is in use, but we should hide that using email verification.
     """
-    from .models import EmailAddress
-
-    if not EmailAddress.objects.lookup([email]).exists():
+    if not filter_users_by_email(email):
         # All good.
         return True
     elif not app_settings.PREVENT_ENUMERATION:

allauth/conftest.py

@@ -35,6 +35,7 @@ def factory(
         with_email=True,
         email_verified=True,
         password=None,
+        with_emailaddress=True,
     ):
         if not username:
             username = uuid.uuid4().hex
@@ -49,7 +50,7 @@ def factory(
         user_email(user, email or "")
         if commit:
             user.save()
-            if email:
+            if email and with_emailaddress:
                 EmailAddress.objects.create(
                     user=user, email=email, verified=email_verified, primary=True
                 )

allauth/socialaccount/models.py

@@ -10,7 +10,11 @@
 
 import allauth.app_settings
 from allauth.account.models import EmailAddress
-from allauth.account.utils import get_next_redirect_url, setup_user_email
+from allauth.account.utils import (
+    filter_users_by_email,
+    get_next_redirect_url,
+    setup_user_email,
+)
 from allauth.core import context
 from allauth.socialaccount import signals
 
@@ -328,16 +332,13 @@ def _lookup_by_socialaccount(self):
 
     def _lookup_by_email(self):
         emails = [e.email for e in self.email_addresses if e.verified]
-        if not emails:
-            return
-        address = (
-            EmailAddress.objects.lookup(emails).order_by("-verified", "user_id").first()
-        )
-        if address:
-            if app_settings.EMAIL_AUTHENTICATION_AUTO_CONNECT:
-                self.connect(context.request, address.user)
-            else:
-                self.user = address.user
+        for email in emails:
+            users = filter_users_by_email(email, is_active=True, prefer_verified=True)
+            if users:
+                self.user = users[0]
+                if app_settings.EMAIL_AUTHENTICATION_AUTO_CONNECT:
+                    self.connect(context.request, self.user)
+                return
 
     def get_redirect_url(self, request):
         url = self.state.get("next")

allauth/socialaccount/tests/test_login.py

@@ -1,6 +1,7 @@
 import copy
 from unittest.mock import patch
 
+from django.contrib.auth import get_user_model
 from django.contrib.auth.models import AnonymousUser
 from django.contrib.messages.middleware import MessageMiddleware
 from django.contrib.sessions.middleware import SessionMiddleware
@@ -13,6 +14,7 @@
 from allauth.socialaccount.models import SocialAccount
 
 
+@pytest.mark.parametrize("with_emailaddress", [False, True])
 @pytest.mark.parametrize("auto_connect", [False, True])
 @pytest.mark.parametrize("setting", ["off", "on-global", "on-provider"])
 def test_email_authentication(
@@ -25,6 +27,7 @@ def test_email_authentication(
     rf,
     mailoutbox,
     auto_connect,
+    with_emailaddress,
 ):
     """Tests that when an already existing email is given at the social signup
     form, enumeration preventation kicks in.
@@ -48,7 +51,7 @@ def test_email_authentication(
         settings.SOCIALACCOUNT_EMAIL_AUTHENTICATION = False
     settings.SOCIALACCOUNT_EMAIL_AUTHENTICATION_AUTO_CONNECT = auto_connect
 
-    user = user_factory()
+    user = user_factory(with_emailaddress=with_emailaddress)
 
     sociallogin = sociallogin_factory(email=user.email, provider="unittest-server")
 
@@ -69,7 +72,12 @@ def test_email_authentication(
         assert not added_signal.called
         assert not updated_signal.called
     else:
-        assert resp["location"] == "/accounts/profile/"
+        if with_emailaddress:
+            assert resp["location"] == "/accounts/profile/"
+        else:
+            # user.email is set, but not verified.
+            assert resp["location"] == reverse("account_email_verification_sent")
+        assert get_user_model().objects.count() == 1
         assert SocialAccount.objects.filter(user=user.pk).exists() == auto_connect
         assert added_signal.called == auto_connect
         assert not updated_signal.called