Merge branch 'release/2.025.47'

Author: kobo-bot[bot]

kobo/apps/accounts/mfa/forms.py

@@ -1,43 +1,22 @@
 from allauth.mfa.adapter import get_adapter
 from allauth.mfa.base.forms import AuthenticateForm, ReauthenticateForm
-from allauth.mfa.base.internal.flows import check_rate_limit
-from allauth.mfa.models import Authenticator
-from django.contrib.auth.hashers import check_password
+
+from .serializers import MfaCodeSerializer
 
 
 class MfaAuthenticateMixin:
 
     def clean_code(self):
-        clear_rl = check_rate_limit(self.user)
         code = self.cleaned_data['code']
-        for auth in Authenticator.objects.filter(user=self.user).exclude(
-            # WebAuthn cannot validate manual codes.
-            type=Authenticator.Type.WEBAUTHN
-        ):
-            if auth.wrap().validate_code(code):
-                self.authenticator = auth
-                clear_rl()
-                return code
-            if auth.type == Authenticator.Type.RECOVERY_CODES:
-                hashed_code = self.validate_migrated_codes(code, auth.wrap())
-                if hashed_code is not None:
-                    self.authenticator = auth
-                    clear_rl()
-                    return code
+        serializer = MfaCodeSerializer(
+            data={'code': code}, context={'user': self.user, 'method': 'app'}
+        )
 
-        raise get_adapter().validation_error('incorrect_code')
+        if serializer.is_valid():
+            self.authenticator = serializer.authenticator
+            return code
 
-    def validate_migrated_codes(self, input_code, recovery_codes):
-        codes = recovery_codes._get_migrated_codes()
-        if codes is None:
-            return
-        if not codes[0].startswith('pbkdf2_sha256$'):
-            return
-        # if codes are sha256 hashes do the recovery codes logic
-        for idx, hashed_code in enumerate(codes):
-            if check_password(input_code, hashed_code):
-                recovery_codes.validate_code(hashed_code)
-                return hashed_code
+        raise get_adapter().validation_error('incorrect_code')
 
 
 class MfaAuthenticateForm(MfaAuthenticateMixin, AuthenticateForm):

kobo/apps/accounts/mfa/serializers.py

@@ -1,4 +1,6 @@
-from allauth.mfa.totp.internal.auth import TOTP
+from allauth.mfa.base.internal.flows import check_rate_limit
+from allauth.mfa.models import Authenticator
+from django.contrib.auth.hashers import check_password
 from rest_framework import serializers
 from rest_framework.exceptions import NotFound, ValidationError
 
@@ -21,9 +23,9 @@ class Meta:
         )
 
 
-class TOTPCodeSerializer(serializers.Serializer):
+class MfaCodeSerializer(serializers.Serializer):
     """
-    Intended to be used for validating TOTP codes
+    Intended to be used for validating TOTP and recovery codes
     """
 
     code = serializers.CharField()
@@ -38,10 +40,31 @@ def __init__(self, *args, **kwargs):
             )
         except MfaMethodsWrapper.DoesNotExist:
             raise NotFound
-        self.totp = TOTP(self.mfamethods.totp)
 
-    def validate_code(self, value):
-        if not self.totp.validate_code(value):
-            raise ValidationError('Invalid TOTP code')
+    def validate_code(self, code):
+        clear_rl = check_rate_limit(self.user)
+        for auth in [self.mfamethods.totp, self.mfamethods.recovery_codes]:
+            if auth.wrap().validate_code(code):
+                self.authenticator = auth
+                clear_rl()
+                return code
+            if auth.type == Authenticator.Type.RECOVERY_CODES:
+                hashed_code = self.validate_migrated_codes(code, auth.wrap())
+                if hashed_code is not None:
+                    self.authenticator = auth
+                    clear_rl()
+                    return code
 
-        return value
+        raise ValidationError('Invalid code')
+
+    def validate_migrated_codes(self, input_code, recovery_codes):
+        codes = recovery_codes._get_migrated_codes()
+        if codes is None:
+            return
+        if not codes[0].startswith('pbkdf2_sha256$'):
+            return
+        # if codes are sha256 hashes do the recovery codes logic
+        for idx, hashed_code in enumerate(codes):
+            if check_password(input_code, hashed_code):
+                recovery_codes.validate_code(hashed_code)
+                return hashed_code

kobo/apps/accounts/mfa/signals.py

@@ -17,11 +17,11 @@ def deactivate_mfa_method_for_user(**kwargs):
     mfa_available_to_user = kwargs['instance']
     # Deactivate any MFA methods user could have already created
     try:
-        # Use `.get()` + `.save()` (from model `MfaMethod`) instead of
+        # Use `.get()` + `.save()` (from model `MfaMethodsWrapper`) instead of
         # `.update()` to run some logic inside `.save()`. It makes an extra
         # query to DB but avoid duplicated code.
         mfa_method = MfaMethodsWrapper.objects.get(user=mfa_available_to_user.user)
-    except MfaMethod.DoesNotExist:
+    except MfaMethodsWrapper.DoesNotExist:
         pass
     else:
         mfa_method.is_active = False

kobo/apps/accounts/mfa/tests/test_api.py

@@ -1,4 +1,5 @@
 from constance.test import override_config
+from django.test import override_settings
 from django.urls import reverse
 from rest_framework import status
 
@@ -10,6 +11,7 @@
 METHOD = 'app'
 
 
+@override_settings(ACCOUNT_RATE_LIMITS=False)
 class MfaApiTestCase(BaseTestCase):
 
     fixtures = ['test_data']

kobo/apps/accounts/mfa/tests/test_migration.py

@@ -1,6 +1,7 @@
 from allauth.account.models import EmailAddress
 from allauth.mfa.adapter import get_adapter
 from django.conf import settings
+from django.test import override_settings
 from django.urls import reverse
 from rest_framework import status
 from trench.command.replace_mfa_method_backup_codes import (
@@ -13,6 +14,7 @@
 from .utils import get_mfa_code_for_user
 
 
+@override_settings(ACCOUNT_RATE_LIMITS=False)
 class MfaMigrationTestCase(BaseTestCase):
     """
     Test the migration scripts

kobo/apps/accounts/mfa/views.py

@@ -16,7 +16,7 @@
 from .flows import activate_totp, deactivate_totp, regenerate_codes
 from .models import MfaMethodsWrapper
 from .permissions import IsMfaEnabled
-from .serializers import TOTPCodeSerializer, UserMfaMethodSerializer
+from .serializers import MfaCodeSerializer, UserMfaMethodSerializer
 
 
 class MfaLoginView(LoginView):
@@ -111,7 +111,7 @@ class MfaMethodDeactivateView(APIView):
 
     @staticmethod
     def post(request: Request, method: str) -> Response:
-        serializer = TOTPCodeSerializer(
+        serializer = MfaCodeSerializer(
             data=request.data, context={'user': request.user, 'method': method}
         )
         serializer.is_valid(raise_exception=True)
@@ -124,7 +124,7 @@ class MfaMethodRegenerateCodesView(APIView):
 
     @staticmethod
     def post(request: Request, method: str) -> Response:
-        serializer = TOTPCodeSerializer(
+        serializer = MfaCodeSerializer(
             data=request.data, context={'user': request.user, 'method': method}
         )
         serializer.is_valid(raise_exception=True)