fix(socialaccount): Wipe password on email authentication

ChangeLog.rst

@@ -16,6 +16,12 @@ Security notice
   email addresses and locking other users out. This safety check is now added to
   WebAuthn security keys as well.
 
+- In case a user signs in into an account using social account email
+  authentication (``SOCIALACCOUNT_EMAIL_AUTHENTICATION``) and the email used is
+  not verified, the password of the account is now wiped (made unusable) to
+  prevent the person that created the account (without verifying it) from
+  signing in.
+
 
 64.2.0 (2024-08-30)
 *******************

allauth/socialaccount/internal/flows/__init__.py

@@ -1,4 +1,9 @@
-from allauth.socialaccount.internal.flows import connect, login, signup
+from allauth.socialaccount.internal.flows import (
+    connect,
+    email_authentication,
+    login,
+    signup,
+)
 
 
-__all__ = ["connect", "login", "signup"]
+__all__ = ["connect", "login", "signup", "email_authentication"]

allauth/socialaccount/internal/flows/email_authentication.py

@@ -0,0 +1,34 @@
+from allauth import app_settings as allauth_settings
+from allauth.account.models import EmailAddress
+
+
+def wipe_password(request, user, email: str):
+    """
+    Consider a scenario where an attacker signs up for an account using the
+    email address of a victim. Obviously, the email address cannot be
+    verified, yet the attacker -- knowing the password -- can wait until the
+    victim appears.  When the victim signs in using email authentication, it
+    is not obvious that the victim is signing into an account that was not
+    created by the victim. As a result, both the attacker and the victim now
+    have access to the account. To prevent this, we wipe the password of the
+    account in case the email address was not verified, effectively locking
+    out the attacker.
+    """
+    try:
+        address = EmailAddress.objects.get_for_user(user, email)
+    except EmailAddress.DoesNotExist:
+        address = None
+    if address and address.verified:
+        # Verified email address, no reason to worry.
+        return
+    if user.has_usable_password():
+        user.set_unusable_password()
+        user.save(update_fields=["password"])
+    # Also wipe any other sessions (upstream integrators may hook up to the
+    # ending of the sessions to trigger e.g. backchannel logout.
+    if allauth_settings.USERSESSIONS_ENABLED:
+        from allauth.usersessions.internal.flows.sessions import (
+            end_other_sessions,
+        )
+
+        end_other_sessions(request, user)

allauth/socialaccount/internal/flows/login.py

@@ -20,7 +20,7 @@
 
 
 def _login(request, sociallogin):
-    sociallogin._accept_login()
+    sociallogin._accept_login(request)
     record_authentication(request, sociallogin)
     return perform_login(
         request,

allauth/socialaccount/models.py

@@ -208,6 +208,7 @@ class SocialLogin:
     token: Optional[SocialToken]
     email_addresses: List[EmailAddress]
     state: Dict
+    _did_authenticate_by_email: Optional[str]
 
     def __init__(
         self,
@@ -309,7 +310,7 @@ def lookup(self) -> None:
         """Look up the existing local user account to which this social login
         points, if any.
         """
-        self._did_authenticate_by_email = False
+        self._did_authenticate_by_email = None
         if not self._lookup_by_socialaccount():
             self._lookup_by_email()
 
@@ -364,11 +365,16 @@ def _lookup_by_email(self) -> None:
             users = filter_users_by_email(email, prefer_verified=True)
             if users:
                 self.user = users[0]
-                self._did_authenticate_by_email = True
+                self._did_authenticate_by_email = email
                 return
 
-    def _accept_login(self) -> None:
+    def _accept_login(self, request) -> None:
+        from allauth.socialaccount.internal.flows.email_authentication import (
+            wipe_password,
+        )
+
         if self._did_authenticate_by_email:
+            wipe_password(request, self.user, self._did_authenticate_by_email)
             if app_settings.EMAIL_AUTHENTICATION_AUTO_CONNECT:
                 self.connect(context.request, self.user)
 

allauth/socialaccount/tests/test_login.py

@@ -54,6 +54,7 @@ def test_email_authentication(
     settings.SOCIALACCOUNT_EMAIL_AUTHENTICATION_AUTO_CONNECT = auto_connect
 
     user = user_factory(with_emailaddress=with_emailaddress)
+    assert user.has_usable_password()
 
     sociallogin = sociallogin_factory(
         email=user.email, provider="unittest-server", with_token=True
@@ -69,14 +70,21 @@ def test_email_authentication(
                 "allauth.socialaccount.signals.social_account_added.send"
             ) as added_signal:
                 resp = complete_social_login(request, sociallogin)
+    user.refresh_from_db()
     if setting == "off":
         assert resp["location"] == reverse("account_email_verification_sent")
         assert not added_signal.called
         assert not updated_signal.called
     else:
         if with_emailaddress:
             assert resp["location"] == "/accounts/profile/"
+            assert user.has_usable_password()
         else:
+            assert not user.has_usable_password()
+            # This should be improved. The provider vouches for the fact that
+            # the user verified the email, so we can mark it as such locally as
+            # well.
+
             # user.email is set, but not verified.
             assert resp["location"] == reverse("account_email_verification_sent")
         assert get_user_model().objects.count() == 1

allauth/usersessions/forms.py

@@ -1,7 +1,6 @@
 from django import forms
 
 from allauth.usersessions.internal import flows
-from allauth.usersessions.models import UserSession
 
 
 class ManageUserSessionsForm(forms.Form):
@@ -10,9 +9,4 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def save(self, request):
-        sessions_to_end = []
-        for session in UserSession.objects.filter(user=request.user):
-            if session.is_current():
-                continue
-            sessions_to_end.append(session)
-        flows.sessions.end_sessions(request, sessions_to_end)
+        flows.sessions.end_other_sessions(request, request.user)

allauth/usersessions/internal/flows/sessions.py

@@ -1,5 +1,15 @@
 from allauth.account.internal import flows
 from allauth.usersessions.adapter import get_adapter
+from allauth.usersessions.models import UserSession
+
+
+def end_other_sessions(request, user):
+    sessions_to_end = []
+    for session in UserSession.objects.filter(user=user):
+        if session.is_current():
+            continue
+        sessions_to_end.append(session)
+    end_sessions(request, sessions_to_end)
 
 
 def end_sessions(request, sessions):