fix(headless): Connect third-party account vs HEADLESS_ONLY

ChangeLog.rst

@@ -4,6 +4,9 @@
 - ``allauth.headless`` now supports the ``is_open_for_signup()`` adapter method.
   In case signup is closed, a 403 is returned during signup.
 
+- Connecting a third-party account in ``HEADLESS_ONLY`` mode failed if the
+  connections view could not be reversed, fixed.
+
 
 0.63.1 (2024-05-17)
 *******************

allauth/account/internal/flows/reauthentication.py

@@ -1,5 +1,19 @@
+from django.http import HttpResponseRedirect
+from django.urls import reverse
+
 from allauth.account.authentication import record_authentication
 
 
+STATE_SESSION_KEY = "account_reauthentication_state"
+
+
 def reauthenticate_by_password(request):
     record_authentication(request, method="password", reauthenticated=True)
+
+
+def stash_and_reauthenticate(request, state, callback):
+    request.session[STATE_SESSION_KEY] = {
+        "state": state,
+        "callback": callback,
+    }
+    return HttpResponseRedirect(reverse("account_reauthenticate"))

allauth/account/reauthentication.py

@@ -2,12 +2,16 @@
 
 from django.contrib.auth import REDIRECT_FIELD_NAME
 from django.http import HttpResponseRedirect
-from django.urls import resolve, reverse
+from django.urls import resolve
 from django.utils.http import urlencode
 
 from allauth.account import app_settings
 from allauth.account.adapter import get_adapter
 from allauth.account.authentication import get_authentication_records
+from allauth.account.internal.flows.reauthentication import (
+    STATE_SESSION_KEY,
+    stash_and_reauthenticate,
+)
 from allauth.core.exceptions import ReauthenticationRequired
 from allauth.core.internal.httpkit import (
     deserialize_request,
@@ -16,9 +20,6 @@
 from allauth.utils import import_callable
 
 
-STATE_SESSION_KEY = "account_reauthentication_state"
-
-
 def suspend_request(request, redirect_to):
     path = request.get_full_path()
     if request.method == "POST":
@@ -52,11 +53,7 @@ def reauthenticate_then_callback(request, serialize_state, callback):
     # XHR.
     if did_recently_authenticate(request):
         return None
-    request.session[STATE_SESSION_KEY] = {
-        "state": serialize_state(request),
-        "callback": callback,
-    }
-    return HttpResponseRedirect(reverse("account_reauthenticate"))
+    return stash_and_reauthenticate(request, serialize_state(request), callback)
 
 
 def raise_if_reauthentication_required(request):

allauth/headless/socialaccount/internal.py

@@ -59,7 +59,7 @@ def complete_login(request, sociallogin):
     """
     error = None
     try:
-        flows.login.complete_login(request, sociallogin)
+        flows.login.complete_login(request, sociallogin, raises=True)
     except SignupClosedException:
         error = "signup_closed"
     else:
@@ -80,7 +80,7 @@ def complete_login(request, sociallogin):
             ]
         ):
             error = AuthError.UNKNOWN
-    next_url = sociallogin.state.get("next")
+    next_url = sociallogin.state["next"]
     if error:
         next_url = httpkit.add_query_params(
             next_url,

allauth/headless/socialaccount/tests/test_views.py

@@ -278,3 +278,18 @@ def test_signup_closed(client, headless_reverse, db, settings):
             content_type="application/json",
         )
     assert resp.status_code == 403
+
+
+def test_connect(user, auth_client, sociallogin_setup_state, headless_reverse, db):
+    state = sociallogin_setup_state(
+        auth_client, process="connect", next="/foo", headless=True
+    )
+    resp = auth_client.post(
+        reverse("dummy_authenticate") + f"?state={state}",
+        data={
+            "id": 123,
+        },
+    )
+    assert resp.status_code == 302
+    assert resp["location"] == "/foo"
+    assert SocialAccount.objects.filter(user=user, provider="dummy", uid="123").exists()

allauth/socialaccount/internal/flows/connect.py

@@ -1,14 +1,14 @@
 from django.contrib import messages
+from django.core.exceptions import PermissionDenied
 from django.http import HttpResponseRedirect
 
 from allauth import app_settings as allauth_settings
 from allauth.account import app_settings as account_settings
 from allauth.account.adapter import get_adapter as get_account_adapter
+from allauth.account.internal import flows
 from allauth.account.models import EmailAddress
-from allauth.account.reauthentication import (
-    raise_if_reauthentication_required,
-    reauthenticate_then_callback,
-)
+from allauth.account.reauthentication import raise_if_reauthentication_required
+from allauth.core.exceptions import ReauthenticationRequired
 from allauth.socialaccount import signals
 from allauth.socialaccount.adapter import get_adapter
 from allauth.socialaccount.models import SocialAccount, SocialLogin
@@ -70,35 +70,50 @@ def resume_connect(request, serialized_state):
 
 
 def connect(request, sociallogin):
-    if request.user.is_anonymous:
+    try:
+        ok, action, message = do_connect(request, sociallogin)
+    except PermissionDenied:
         # This should not happen. Simply redirect to the connections
         # view (which has a login required)
         connect_redirect_url = get_adapter().get_connect_redirect_url(
             request, sociallogin.account
         )
         return HttpResponseRedirect(connect_redirect_url)
-    if account_settings.REAUTHENTICATION_REQUIRED:
-        response = reauthenticate_then_callback(
+    except ReauthenticationRequired:
+        return flows.reauthentication.stash_and_reauthenticate(
             request,
-            lambda request: sociallogin.serialize(),
+            sociallogin.serialize(request),
             "allauth.socialaccount.internal.flows.connect.resume_connect",
         )
-        if response:
-            return response
-    level = messages.INFO
+    level = messages.INFO if ok else messages.ERROR
+    default_next = get_adapter().get_connect_redirect_url(request, sociallogin.account)
+    next_url = sociallogin.get_redirect_url(request) or default_next
+    get_account_adapter(request).add_message(
+        request,
+        level,
+        message,
+        message_context={"sociallogin": sociallogin, "action": action},
+    )
+    return HttpResponseRedirect(next_url)
+
+
+def do_connect(request, sociallogin):
+    if request.user.is_anonymous:
+        raise PermissionDenied()
+    if account_settings.REAUTHENTICATION_REQUIRED:
+        raise_if_reauthentication_required(request)
     message = "socialaccount/messages/account_connected.txt"
     action = None
+    ok = True
     if sociallogin.is_existing:
         if sociallogin.user != request.user:
             # Social account of other user. For now, this scenario
             # is not supported. Issue is that one cannot simply
             # remove the social account from the other user, as
             # that may render the account unusable.
-            level = messages.ERROR
             message = "socialaccount/messages/account_connected_other.txt"
+            ok = False
         elif not sociallogin.account._state.adding:
-            # This account is already connected -- we give the opportunity
-            # for customized behaviour through use of a signal.
             action = "updated"
             message = "socialaccount/messages/account_connected_updated.txt"
         else:
@@ -108,13 +123,4 @@ def connect(request, sociallogin):
         # New account, let's connect
         action = "added"
         sociallogin.connect(request, request.user)
-    assert request.user.is_authenticated
-    default_next = get_adapter().get_connect_redirect_url(request, sociallogin.account)
-    next_url = sociallogin.get_redirect_url(request) or default_next
-    get_account_adapter(request).add_message(
-        request,
-        level,
-        message,
-        message_context={"sociallogin": sociallogin, "action": action},
-    )
-    return HttpResponseRedirect(next_url)
+    return ok, action, message

allauth/socialaccount/internal/flows/login.py

@@ -10,7 +10,7 @@
 )
 from allauth.socialaccount import app_settings, signals
 from allauth.socialaccount.adapter import get_adapter
-from allauth.socialaccount.internal.flows.connect import connect
+from allauth.socialaccount.internal.flows.connect import connect, do_connect
 from allauth.socialaccount.internal.flows.signup import (
     clear_pending_signup,
     process_signup,
@@ -31,20 +31,27 @@ def _login(request, sociallogin):
     )
 
 
-def complete_login(request, sociallogin, raises=False):
+def pre_social_login(request, sociallogin):
     clear_pending_signup(request)
     assert not sociallogin.is_existing
     sociallogin.lookup()
+    get_adapter().pre_social_login(request, sociallogin)
+    signals.pre_social_login.send(
+        sender=SocialLogin, request=request, sociallogin=sociallogin
+    )
+
+
+def complete_login(request, sociallogin, raises=False):
     try:
-        get_adapter().pre_social_login(request, sociallogin)
-        signals.pre_social_login.send(
-            sender=SocialLogin, request=request, sociallogin=sociallogin
-        )
+        pre_social_login(request, sociallogin)
         process = sociallogin.state.get("process")
         if process == AuthProcess.REDIRECT:
             return _redirect(request, sociallogin)
         elif process == AuthProcess.CONNECT:
-            return connect(request, sociallogin)
+            if raises:
+                do_connect(request, sociallogin)
+            else:
+                return connect(request, sociallogin)
         else:
             return _authenticate(request, sociallogin)
     except SignupClosedException: