feat(account): Reauthentication required

ChangeLog.rst

@@ -16,6 +16,10 @@ Note worthy changes
   for example, because the user logs in by username, removal of the email
   address is allowed.
 
+- Added a new setting ``ACCOUNT_REAUTHENTICATION_REQUIRED`` that, when enabled,
+  requires the user to reauthenticate before changes (such as changing the
+  primary email address, adding a new email address, etc.) can be performed.
+
 
 Backwards incompatible changes
 ------------------------------

allauth/account/adapter.py

@@ -481,7 +481,7 @@ def post_login(
         return response
 
     def login(self, request, user):
-        from allauth.account.utils import record_authentication
+        from allauth.account.reauthentication import record_authentication
 
         # HACK: This is not nice. The proper Django way is to use an
         # authentication backend

allauth/account/app_settings.py

@@ -390,6 +390,10 @@ def PASSWORD_RESET_TOKEN_GENERATOR(self):
     def REAUTHENTICATION_TIMEOUT(self):
         return self._setting("REAUTHENTICATION_TIMEOUT", 300)
 
+    @property
+    def REAUTHENTICATION_REQUIRED(self):
+        return self._setting("REAUTHENTICATION_REQUIRED", False)
+
 
 _app_settings = AppSettings("ACCOUNT_")
 

allauth/account/decorators.py

@@ -2,13 +2,15 @@
 
 from django.contrib.auth import REDIRECT_FIELD_NAME
 from django.contrib.auth.decorators import login_required
-from django.http import HttpResponseRedirect
 from django.shortcuts import render
 from django.urls import reverse
-from django.utils.http import urlencode
 
-from .models import EmailAddress
-from .utils import did_recently_authenticate, send_email_confirmation
+from allauth.account.models import EmailAddress
+from allauth.account.reauthentication import (
+    did_recently_authenticate,
+    suspend_request,
+)
+from allauth.account.utils import send_email_confirmation
 
 
 def verified_email_required(
@@ -43,27 +45,27 @@ def _wrapped_view(request, *args, **kwargs):
     return decorator
 
 
-def reauthentication_required(function=None, redirect_field_name=REDIRECT_FIELD_NAME):
+def reauthentication_required(
+    function=None,
+    redirect_field_name=REDIRECT_FIELD_NAME,
+    allow_get=False,
+    enabled=None,
+):
     def decorator(view_func):
         @wraps(view_func)
         def _wrapper_view(request, *args, **kwargs):
-            path = request.get_full_path()
-            if request.user.is_anonymous:
-                redirect_url = (
-                    reverse("account_login")
-                    + "?"
-                    + urlencode({redirect_field_name: path})
-                )
-                return HttpResponseRedirect(redirect_url)
-
-            if not did_recently_authenticate(request):
-                redirect_url = (
-                    reverse("account_reauthenticate")
-                    + "?"
-                    + urlencode({redirect_field_name: path})
-                )
-                return HttpResponseRedirect(redirect_url)
-
+            pass_method = allow_get and request.method == "GET"
+            ena = (enabled is None) or (
+                enabled(request) if callable(enabled) else enabled
+            )
+            if ena and not pass_method:
+                if request.user.is_anonymous or not did_recently_authenticate(request):
+                    redirect_url = reverse(
+                        "account_login"
+                        if request.user.is_anonymous
+                        else "account_reauthenticate"
+                    )
+                    return suspend_request(request, redirect_url)
             return view_func(request, *args, **kwargs)
 
         return _wrapper_view

allauth/account/reauthentication.py

@@ -0,0 +1,74 @@
+import time
+
+from django.contrib.auth import REDIRECT_FIELD_NAME
+from django.http import HttpResponseRedirect
+from django.urls import resolve, reverse
+from django.utils.http import urlencode
+
+from allauth.account import app_settings
+from allauth.account.utils import get_next_redirect_url
+from allauth.core.internal.http import deserialize_request, serialize_request
+from allauth.utils import import_callable
+
+
+STATE_SESSION_KEY = "account_reauthentication_state"
+AUTHENTICATED_AT_SESSION_KEY = "account_authenticated_at"
+
+
+def suspend_request(request, redirect_to):
+    path = request.get_full_path()
+    if request.method == "POST":
+        request.session[STATE_SESSION_KEY] = {"request": serialize_request(request)}
+    return HttpResponseRedirect(
+        redirect_to + "?" + urlencode({REDIRECT_FIELD_NAME: path})
+    )
+
+
+def resume_request(request):
+    state = request.session.pop(STATE_SESSION_KEY, None)
+    if state and "callback" in state:
+        callback = import_callable(state["callback"])
+        return callback(request, state["state"])
+
+    url = get_next_redirect_url(request, REDIRECT_FIELD_NAME)
+    if not url:
+        return None
+    if state and "request" in state:
+        suspended_request = deserialize_request(state["request"], request)
+        if suspended_request.path == url:
+            resolved = resolve(suspended_request.path)
+            return resolved.func(suspended_request, *resolved.args, **resolved.kwargs)
+    return HttpResponseRedirect(url)
+
+
+def record_authentication(request, user):
+    request.session[AUTHENTICATED_AT_SESSION_KEY] = time.time()
+
+
+def reauthenticate_then_callback(request, serialize_state, callback):
+    # TODO: Currently, ACCOUNT_REAUTHENTICATION_REQUIRED does not play well with
+    # XHR.
+    if did_recently_authenticate(request):
+        return None
+    request.session[STATE_SESSION_KEY] = {
+        "state": serialize_state(request),
+        "callback": callback,
+    }
+    return HttpResponseRedirect(reverse("account_reauthenticate"))
+
+
+def did_recently_authenticate(request):
+    if request.user.is_anonymous:
+        return False
+    if not request.user.has_usable_password():
+        # TODO: This user only has social accounts attached. Now, ideally, you
+        # would want to reauthenticate over at the social account provider. For
+        # now, this is not implemented. Although definitely suboptimal, this
+        # method is currently used for reauthentication checks over at MFA, and,
+        # users that delegate the security of their account to an external
+        # provider like Google typically use MFA over there anyway.
+        return True
+    authenticated_at = request.session.get(AUTHENTICATED_AT_SESSION_KEY)
+    if not authenticated_at:
+        return False
+    return time.time() - authenticated_at < app_settings.REAUTHENTICATION_TIMEOUT

allauth/account/tests/test_change_email.py

@@ -263,6 +263,22 @@ def test_add(auth_client, user, settings):
     assertTemplateUsed(resp, "account/messages/email_confirmation_sent.txt")
 
 
+def test_add_with_reauthentication(auth_client, user, user_password, settings):
+    settings.ACCOUNT_REAUTHENTICATION_REQUIRED = True
+    resp = auth_client.post(
+        reverse("account_email"),
+        {"action_add": "", "email": "[email protected]"},
+    )
+    assert not EmailAddress.objects.filter(email="[email protected]").exists()
+    assert resp.status_code == 302
+    assert resp["location"] == reverse("account_reauthenticate") + "?next=%2Femail%2F"
+    resp = auth_client.post(resp["location"], {"password": user_password})
+    assert EmailAddress.objects.filter(email="[email protected]").exists()
+    assertTemplateUsed(resp, "account/messages/email_confirmation_sent.txt")
+    assert resp.status_code == 302
+    assert resp["location"] == reverse("account_email")
+
+
 @pytest.mark.parametrize(
     "prevent_enumeration",
     [

allauth/account/utils.py

@@ -1,4 +1,3 @@
-import time
 import unicodedata
 from collections import OrderedDict
 from typing import Optional
@@ -527,27 +526,6 @@ def url_str_to_user_pk(pk_str):
     return pk
 
 
-def record_authentication(request, user):
-    request.session["account_authenticated_at"] = time.time()
-
-
-def did_recently_authenticate(request):
-    if request.user.is_anonymous:
-        return False
-    if not request.user.has_usable_password():
-        # TODO: This user only has social accounts attached. Now, ideally, you
-        # would want to reauthenticate over at the social account provider. For
-        # now, this is not implemented. Although definitely suboptimal, this
-        # method is currently used for reauthentication checks over at MFA, and,
-        # users that delegate the security of their account to an external
-        # provider like Google typically use MFA over there anyway.
-        return True
-    authenticated_at = request.session.get("account_authenticated_at")
-    if not authenticated_at:
-        return False
-    return time.time() - authenticated_at < app_settings.REAUTHENTICATION_TIMEOUT
-
-
 def assess_unique_email(email) -> Optional[bool]:
     """
     True -- email is unique

allauth/account/views.py

@@ -17,14 +17,10 @@
 from django.views.generic.base import TemplateResponseMixin, TemplateView, View
 from django.views.generic.edit import FormView
 
-from allauth.core import ratelimit
-from allauth.core.exceptions import ImmediateHttpResponse
-from allauth.decorators import rate_limit
-from allauth.utils import get_form_class, get_request_param
-
-from . import app_settings, signals
-from .adapter import get_adapter
-from .forms import (
+from allauth.account import app_settings, signals
+from allauth.account.adapter import get_adapter
+from allauth.account.decorators import reauthentication_required
+from allauth.account.forms import (
     AddEmailForm,
     ChangePasswordForm,
     LoginForm,
@@ -35,19 +31,30 @@
     SignupForm,
     UserTokenForm,
 )
-from .models import EmailAddress, EmailConfirmation, EmailConfirmationHMAC
-from .utils import (
+from allauth.account.models import (
+    EmailAddress,
+    EmailConfirmation,
+    EmailConfirmationHMAC,
+)
+from allauth.account.reauthentication import (
+    record_authentication,
+    resume_request,
+)
+from allauth.account.utils import (
     complete_signup,
     get_login_redirect_url,
     get_next_redirect_url,
     logout_on_password_change,
     passthrough_next_redirect_url,
     perform_login,
-    record_authentication,
     send_email_confirmation,
     sync_user_email_addresses,
     url_str_to_user_pk,
 )
+from allauth.core import ratelimit
+from allauth.core.exceptions import ImmediateHttpResponse
+from allauth.decorators import rate_limit
+from allauth.utils import get_form_class, get_request_param
 
 
 INTERNAL_RESET_SESSION_KEY = "_password_reset_key"
@@ -450,6 +457,12 @@ def get_redirect_url(self):
 
 
 @method_decorator(rate_limit(action="manage_email"), name="dispatch")
+@method_decorator(
+    reauthentication_required(
+        allow_get=True, enabled=lambda request: app_settings.REAUTHENTICATION_REQUIRED
+    ),
+    name="dispatch",
+)
 class EmailView(AjaxCapableProcessFormViewMixin, FormView):
     template_name = (
         "account/email_change." if app_settings.CHANGE_EMAIL else "account/email."
@@ -995,6 +1008,9 @@ def get_form_kwargs(self):
 
     def form_valid(self, form):
         record_authentication(self.request, self.request.user)
+        response = resume_request(self.request)
+        if response:
+            return response
         return super().form_valid(form)
 
     def get_context_data(self, **kwargs):

allauth/core/internal/http.py

@@ -0,0 +1,27 @@
+import json
+
+from django.http import QueryDict
+
+
+def serialize_request(request):
+    return json.dumps(
+        {
+            "path": request.path,
+            "path_info": request.path_info,
+            "META": {k: v for k, v in request.META.items() if isinstance(v, str)},
+            "GET": request.GET.urlencode(),
+            "POST": request.POST.urlencode(),
+            "method": request.method,
+        }
+    )
+
+
+def deserialize_request(s, request):
+    data = json.loads(s)
+    request.GET = QueryDict(data["GET"])
+    request.POST = QueryDict(data["POST"])
+    request.META = data["META"]
+    request.path = data["path"]
+    request.path_info = data["path_info"]
+    request.method = data["method"]
+    return request

allauth/mfa/tests/test_views.py

@@ -82,7 +82,7 @@ def test_index(auth_client, user_with_totp):
 
 
 def test_deactivate_totp_success(auth_client, user_with_totp, user_password):
-    resp = auth_client.post(reverse("mfa_deactivate_totp"))
+    resp = auth_client.get(reverse("mfa_deactivate_totp"))
     assert resp.status_code == 302
     assert resp["location"].startswith(reverse("account_reauthenticate"))
     resp = auth_client.post(resp["location"], {"password": user_password})

allauth/socialaccount/helpers.py

@@ -6,6 +6,7 @@
 
 from allauth.account import app_settings as account_settings
 from allauth.account.adapter import get_adapter as get_account_adapter
+from allauth.account.reauthentication import reauthenticate_then_callback
 from allauth.account.utils import (
     assess_unique_email,
     complete_signup,
@@ -15,11 +16,10 @@
     user_username,
 )
 from allauth.core.exceptions import ImmediateHttpResponse
-
-from . import app_settings, signals
-from .adapter import get_adapter
-from .models import SocialLogin
-from .providers.base import AuthError, AuthProcess
+from allauth.socialaccount import app_settings, signals
+from allauth.socialaccount.adapter import get_adapter
+from allauth.socialaccount.models import SocialLogin
+from allauth.socialaccount.providers.base import AuthError, AuthProcess
 
 
 def _process_auto_signup(request, sociallogin):
@@ -139,6 +139,11 @@ def render_authentication_error(
     )
 
 
+def resume_add_social_account(request, serialized_state):
+    sociallogin = SocialLogin.deserialize(serialized_state)
+    return _add_social_account(request, sociallogin)
+
+
 def _add_social_account(request, sociallogin):
     if request.user.is_anonymous:
         # This should not happen. Simply redirect to the connections
@@ -147,6 +152,14 @@ def _add_social_account(request, sociallogin):
             request, sociallogin.account
         )
         return HttpResponseRedirect(connect_redirect_url)
+    if account_settings.REAUTHENTICATION_REQUIRED:
+        response = reauthenticate_then_callback(
+            request,
+            lambda request: sociallogin.serialize(),
+            "allauth.socialaccount.helpers.resume_add_social_account",
+        )
+        if response:
+            return response
     level = messages.INFO
     message = "socialaccount/messages/account_connected.txt"
     action = None