perf: During MFA authentication, if the current code has been used and successfully authenticated, it cannot be used again for authentication

apps/authentication/api/mfa.py

@@ -3,7 +3,7 @@
 from django.shortcuts import get_object_or_404
 from django.utils.translation import gettext as _
 from rest_framework import exceptions
-from rest_framework.generics import CreateAPIView, RetrieveAPIView
+from rest_framework.generics import CreateAPIView
 from rest_framework.permissions import AllowAny
 from rest_framework.response import Response
 from rest_framework.serializers import ValidationError
@@ -23,8 +23,6 @@ __all__ = [
 ]
 
 
-
-
 # MFASelectAPi 原来的名字
 class MFASendCodeApi(AuthMixin, CreateAPIView):
     """

apps/authentication/mfa/__init__.py

@@ -1,5 +1,5 @@
-from .otp import MFAOtp, otp_failed_msg
-from .sms import MFASms
-from .radius import MFARadius
 from .custom import MFACustom
-from .face import MFAFace
+from .face import MFAFace
+from .otp import MFAOtp, otp_failed_msg
+from .radius import MFARadius
+from .sms import MFASms

apps/authentication/mfa/base.py

@@ -1,10 +1,12 @@
 import abc
 
+from django.core.cache import cache
 from django.utils.translation import gettext_lazy as _
 
 
 class BaseMFA(abc.ABC):
     placeholder = _('Please input security code')
+    skip_cache_check = False
 
     def __init__(self, user):
         """
@@ -14,6 +16,25 @@ class BaseMFA(abc.ABC):
         self.user = user
         self.request = None
 
+    def check_code(self, code):
+        if self.skip_cache_check:
+            return self._check_code(code)
+
+        cache_key = f'{self.name}_{self.user.username}'
+        cache_code = cache.get(cache_key)
+        if cache_code == code:
+            return False, _(
+                "The two-factor code you entered has either already been used or has expired. "
+                "Please request a new one."
+            )
+
+        ok, msg = self._check_code(code)
+        if not ok:
+            return False, msg
+
+        cache.set(cache_key, code, 60 * 5)
+        return True, msg
+
     def is_authenticated(self):
         return self.user and self.user.is_authenticated
 
@@ -38,7 +59,7 @@ class BaseMFA(abc.ABC):
         pass
 
     @abc.abstractmethod
-    def check_code(self, code) -> tuple:
+    def _check_code(self, code) -> tuple:
         return False, 'Error msg'
 
     @abc.abstractmethod

apps/authentication/mfa/custom.py

@@ -26,7 +26,7 @@ class MFACustom(BaseMFA):
     display_name = MFAType.Custom.name
     placeholder = _("MFA custom verification code")
 
-    def check_code(self, code):
+    def _check_code(self, code):
         assert self.is_authenticated()
         ok = False
         try:

apps/authentication/mfa/face.py

@@ -10,9 +10,9 @@ class MFAFace(BaseMFA, AuthFaceMixin):
     name = MFAType.Face.value
     display_name = MFAType.Face.name
     placeholder = 'Face Recognition'
+    skip_cache_check = True
 
-    def check_code(self, code):
-
+    def _check_code(self, code):
         assert self.is_authenticated()
 
         try:

apps/authentication/mfa/otp.py

@@ -1,10 +1,9 @@
-from django.utils.translation import gettext_lazy as _
 from django.shortcuts import reverse
+from django.utils.translation import gettext_lazy as _
 
 from .base import BaseMFA
 from ..const import MFAType
 
-
 otp_failed_msg = _("OTP code invalid, or server time error")
 
 
@@ -13,7 +12,7 @@ class MFAOtp(BaseMFA):
     display_name = MFAType.OTP.name
     placeholder = _('OTP verification code')
 
-    def check_code(self, code):
+    def _check_code(self, code):
         from users.utils import check_otp_code
         assert self.is_authenticated()
 

apps/authentication/mfa/radius.py

@@ -13,7 +13,7 @@ class MFARadius(BaseMFA):
     display_name = MFAType.Radius.name
     placeholder = _("Radius verification code")
 
-    def check_code(self, code=None):
+    def _check_code(self, code=None):
         assert self.is_authenticated()
         backend = RadiusBackend()
         username = self.user.username

apps/authentication/mfa/sms.py

@@ -24,7 +24,7 @@ class MFASms(BaseMFA):
             phone, backend=self.name, user_info=user_info
         )
 
-    def check_code(self, code):
+    def _check_code(self, code):
         assert self.is_authenticated()
         ok = False
         msg = ''

apps/common/decorators.py

@@ -381,4 +381,3 @@ def bulk_update_decorator(instance_model, batch_size=50, update_fields=None, tim
         instance_model.objects.bulk_update(cache, update_fields)
 
     return bulk_handle(handle, batch_size, timeout)
-

apps/users/views/profile/otp.py

@@ -7,8 +7,8 @@ from django.http.response import HttpResponseRedirect
 from django.shortcuts import redirect
 from django.templatetags.static import static
 from django.urls import reverse
-from django.utils.translation import gettext as _
 from django.utils._os import safe_join
+from django.utils.translation import gettext as _
 from django.views.generic.base import TemplateView
 from django.views.generic.edit import FormView