feat: password reset by code

Reviewed-on: https://codeberg.org/allauth/django-allauth/pulls/4270
Co-authored-by: Raymond Penners <raymond.penners@intenct.nl>
Co-committed-by: Raymond Penners <raymond.penners@intenct.nl>

ChangeLog.rst

@@ -4,7 +4,7 @@
 Note worthy changes
 -------------------
 
-- ...
+- Added support for resetting passwords by code, instead of a link.
 
 
 Fixes

allauth/account/adapter.py

@@ -797,18 +797,24 @@ def generate_login_code(self) -> str:
         """
         return self._generate_code()
 
+    def generate_password_reset_code(self) -> str:
+        """
+        Generates a new password reset code.
+        """
+        return self._generate_code(length=8)
+
     def generate_email_verification_code(self) -> str:
         """
         Generates a new email verification code.
         """
         return self._generate_code()
 
-    def _generate_code(self):
+    def _generate_code(self, length=6):
         forbidden_chars = "0OI18B2ZAEU"
         allowed_chars = string.ascii_uppercase + string.digits
         for ch in forbidden_chars:
             allowed_chars = allowed_chars.replace(ch, "")
-        return get_random_string(length=6, allowed_chars=allowed_chars)
+        return get_random_string(length=length, allowed_chars=allowed_chars)
 
     def is_login_by_code_required(self, login) -> bool:
         """

allauth/account/app_settings.py

@@ -395,6 +395,18 @@ def USERNAME_VALIDATORS(self):
                 ret = []
         return ret
 
+    @property
+    def PASSWORD_RESET_BY_CODE_ENABLED(self):
+        return self._setting("PASSWORD_RESET_BY_CODE_ENABLED", False)
+
+    @property
+    def PASSWORD_RESET_BY_CODE_MAX_ATTEMPTS(self):
+        return self._setting("PASSWORD_RESET_BY_CODE_MAX_ATTEMPTS", 3)
+
+    @property
+    def PASSWORD_RESET_BY_CODE_TIMEOUT(self):
+        return self._setting("PASSWORD_RESET_BY_CODE_TIMEOUT", 3 * 60)
+
     @property
     def PASSWORD_RESET_TOKEN_GENERATOR(self):
         from allauth.account.forms import EmailAwarePasswordResetTokenGenerator

allauth/account/forms.py

@@ -11,29 +11,27 @@
 from allauth.account.app_settings import LoginMethod
 from allauth.account.internal import flows
 from allauth.account.internal.stagekit import LOGIN_SESSION_KEY
+from allauth.account.internal.textkit import compare_code
 from allauth.account.stages import EmailVerificationStage
 from allauth.core import context, ratelimit
 from allauth.utils import get_username_max_length, set_form_field_order
 
 from . import app_settings
-from .adapter import DefaultAccountAdapter, get_adapter
+from .adapter import get_adapter
 from .models import EmailAddress, Login
 from .utils import (
     filter_users_by_email,
     setup_user_email,
     sync_user_email_addresses,
     url_str_to_user_pk,
     user_email,
-    user_pk_to_url_str,
     user_username,
 )
 
 
 class EmailAwarePasswordResetTokenGenerator(PasswordResetTokenGenerator):
     def _make_hash_value(self, user, timestamp):
-        ret = super(EmailAwarePasswordResetTokenGenerator, self)._make_hash_value(
-            user, timestamp
-        )
+        ret = super()._make_hash_value(user, timestamp)
         sync_user_email_addresses(user)
         email = user_email(user)
         emails = set([email] if email else [])
@@ -49,7 +47,7 @@ def _make_hash_value(self, user, timestamp):
 
 class PasswordVerificationMixin:
     def clean(self):
-        cleaned_data = super(PasswordVerificationMixin, self).clean()
+        cleaned_data = super().clean()
         password1 = cleaned_data.get("password1")
         password2 = cleaned_data.get("password2")
         if (password1 and password2) and password1 != password2:
@@ -95,7 +93,7 @@ class LoginForm(forms.Form):
 
     def __init__(self, *args, **kwargs):
         self.request = kwargs.pop("request", None)
-        super(LoginForm, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
         if app_settings.LOGIN_METHODS == {LoginMethod.EMAIL}:
             login_widget = forms.EmailInput(
                 attrs={
@@ -591,33 +589,17 @@ def clean_email(self):
 
     def save(self, request, **kwargs) -> str:
         email = self.cleaned_data["email"]
-        if not self.users:
-            flows.signup.send_unknown_account_mail(request, email)
-            return email
-
-        adapter: DefaultAccountAdapter = get_adapter()
-        token_generator = kwargs.get("token_generator", default_token_generator)
-        for user in self.users:
-            temp_key = token_generator.make_token(user)
-
-            # send the password reset email
-            uid = user_pk_to_url_str(user)
-            # We intentionally pass an opaque `key` on the interface here, and
-            # not implementation details such as a separate `uidb36` and
-            # `key. Ideally, this should have done on `urls` level as well.
-            key = f"{uid}-{temp_key}"
-            url = adapter.get_reset_password_from_key_url(key)
-            context = {
-                "user": user,
-                "password_reset_url": url,
-                "uid": uid,
-                "key": temp_key,
-                "request": request,
-            }
-
-            if LoginMethod.USERNAME in app_settings.LOGIN_METHODS:
-                context["username"] = user_username(user)
-            adapter.send_password_reset_mail(user, email, context)
+        if app_settings.PASSWORD_RESET_BY_CODE_ENABLED:
+            flows.password_reset_by_code.PasswordResetVerificationProcess.initiate(
+                request=request,
+                user=(self.users[0] if self.users else None),
+                email=email,
+            )
+        else:
+            token_generator = kwargs.get("token_generator", default_token_generator)
+            flows.password_reset.request_password_reset(
+                request, email, self.users, token_generator
+            )
         return email
 
 
@@ -628,7 +610,7 @@ class ResetPasswordKeyForm(PasswordVerificationMixin, forms.Form):
     def __init__(self, *args, **kwargs):
         self.user = kwargs.pop("user", None)
         self.temp_key = kwargs.pop("temp_key", None)
-        super(ResetPasswordKeyForm, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
         self.fields["password1"].user = self.user
 
     def save(self):
@@ -721,7 +703,7 @@ def __init__(self, *args, **kwargs):
 
     def clean_code(self):
         code = self.cleaned_data.get("code")
-        if not flows.login_by_code.compare_code(actual=code, expected=self.code):
+        if not compare_code(actual=code, expected=self.code):
             raise get_adapter().validation_error("incorrect_code")
         return code
 
@@ -732,3 +714,7 @@ class ConfirmLoginCodeForm(BaseConfirmCodeForm):
 
 class ConfirmEmailVerificationCodeForm(BaseConfirmCodeForm):
     pass
+
+
+class ConfirmPasswordResetCodeForm(BaseConfirmCodeForm):
+    pass

allauth/account/internal/flows/__init__.py

@@ -7,20 +7,22 @@
     manage_email,
     password_change,
     password_reset,
+    password_reset_by_code,
     reauthentication,
     signup,
 )
 
 
 __all__ = [
-    "password_reset",
-    "password_change",
     "email_verification",
     "email_verification_by_code",
     "login",
     "login_by_code",
     "logout",
-    "signup",
     "manage_email",
+    "password_change",
+    "password_reset",
+    "password_reset_by_code",
     "reauthentication",
+    "signup",
 ]

allauth/account/internal/flows/code_verification.py

@@ -0,0 +1,69 @@
+import abc
+import time
+
+from django.contrib.auth import get_user_model
+
+from allauth.account.internal.userkit import str_to_user_id, user_id_to_str
+
+
+class AbstractCodeVerificationProcess(abc.ABC):
+    def __init__(
+        self,
+        max_attempts: int,
+        timeout: int,
+        state: dict,
+        user=None,
+    ):
+        self._user = user
+        self.max_attempts = max_attempts
+        self.timeout = timeout
+        self.state = state
+
+    @property
+    def user(self):
+        if self._user:
+            return self._user
+        user_id = self.state.get("user_id")
+        if not user_id:
+            return None
+        user_id = str_to_user_id(user_id)
+        return get_user_model().objects.filter(pk=user_id).first()
+
+    @property
+    def code(self):
+        return self.state.get("code", "")
+
+    @classmethod
+    def initial_state(cls, user):
+        state = {
+            "at": time.time(),
+            "failed_attempts": 0,
+        }
+        if user:
+            state["user_id"] = user_id_to_str(user)
+        return state
+
+    def record_invalid_attempt(self) -> bool:
+        n = self.state["failed_attempts"]
+        n += 1
+        self.state["failed_attempts"] = n
+        if n >= self.max_attempts:
+            self.abort()
+            return False
+        self.persist()
+        return True
+
+    def is_valid(self) -> bool:
+        return time.time() - self.state["at"] <= self.timeout
+
+    @abc.abstractmethod
+    def persist(self): ...  # noqa: E704
+
+    @abc.abstractmethod
+    def send(self): ...  # noqa: E704
+
+    @abc.abstractmethod
+    def finish(self): ...  # noqa: E704
+
+    @abc.abstractmethod
+    def abort(self): ...  # noqa: E704

allauth/account/internal/flows/email_verification_by_code.py

@@ -6,8 +6,8 @@
 
 from allauth.account import app_settings
 from allauth.account.adapter import get_adapter
-from allauth.account.internal.flows.login_by_code import compare_code
 from allauth.account.internal.stagekit import clear_login
+from allauth.account.internal.textkit import compare_code
 from allauth.account.models import EmailAddress, EmailConfirmationMixin
 from allauth.core import context
 

allauth/account/internal/flows/login_by_code.py

@@ -125,9 +125,3 @@ def perform_login_by_code(
         return perform_login(request, login)
     else:
         return stage.exit()
-
-
-def compare_code(*, actual, expected) -> bool:
-    actual = actual.replace(" ", "").lower()
-    expected = expected.replace(" ", "").lower()
-    return expected and actual == expected

allauth/account/internal/flows/logout.py

@@ -1,11 +1,15 @@
 from django.contrib import messages
 from django.http import HttpRequest
 
+from allauth.account.internal.flows.password_reset_by_code import (
+    PASSWORD_RESET_VERIFICATION_SESSION_KEY,
+)
 from allauth.account.internal.stagekit import clear_login
 
 
 def logout(request: HttpRequest) -> None:
     from allauth.account.adapter import get_adapter
+    from allauth.account.views import INTERNAL_RESET_SESSION_KEY
 
     if request.user.is_authenticated:
         adapter = get_adapter()
@@ -14,3 +18,5 @@ def logout(request: HttpRequest) -> None:
         )
         adapter.logout(request)
     clear_login(request)
+    request.session.pop(PASSWORD_RESET_VERIFICATION_SESSION_KEY, None)
+    request.session.pop(INTERNAL_RESET_SESSION_KEY, None)

allauth/account/internal/flows/password_reset.py

@@ -5,8 +5,10 @@
 from django.http import HttpRequest
 from django.urls import reverse
 
-from allauth.account import signals
+from allauth.account import app_settings, signals
 from allauth.account.adapter import get_adapter
+from allauth.account.app_settings import LoginMethod
+from allauth.account.internal.flows.signup import send_unknown_account_mail
 from allauth.account.models import EmailAddress
 from allauth.core.internal.httpkit import get_frontend_url
 from allauth.utils import build_absolute_uri
@@ -60,3 +62,35 @@ def get_reset_password_from_key_url(request: HttpRequest, key: str) -> str:
         path = path.replace("UID-KEY", quote(key))
         url = build_absolute_uri(request, path)
     return url
+
+
+def request_password_reset(request, email, users, token_generator):
+    from allauth.account.utils import user_pk_to_url_str, user_username
+
+    if not users:
+        send_unknown_account_mail(request, email)
+        return
+    adapter = get_adapter()
+    for user in users:
+        temp_key = (
+            token_generator or app_settings.PASSWORD_RESET_TOKEN_GENERATOR()
+        ).make_token(user)
+
+        # send the password reset email
+        uid = user_pk_to_url_str(user)
+        # We intentionally pass an opaque `key` on the interface here, and
+        # not implementation details such as a separate `uidb36` and
+        # `key. Ideally, this should have done on `urls` level as well.
+        key = f"{uid}-{temp_key}"
+        url = adapter.get_reset_password_from_key_url(key)
+        context = {
+            "user": user,
+            "password_reset_url": url,
+            "uid": uid,
+            "key": temp_key,
+            "request": request,
+        }
+
+        if LoginMethod.USERNAME in app_settings.LOGIN_METHODS:
+            context["username"] = user_username(user)
+        adapter.send_password_reset_mail(user, email, context)