perf: 优化confirm接口 (#8451)

* perf: 优化confirm接口

* perf: 修改 校验

* perf: 优化 confirm API 逻辑

* Delete django.po

Co-authored-by: feng626 <1304903146@qq.com>
Co-authored-by: ibuler <ibuler@qq.com>
Co-authored-by: Jiangjie.Bai <bugatti_it@163.com>
Co-authored-by: feng626 <57284900+feng626@users.noreply.github.com>

apps/authentication/api/confirm.py

@@ -1,85 +1,57 @@
 # -*- coding: utf-8 -*-
 #
 import time
-from datetime import datetime
 
-from django.utils import timezone
-from django.conf import settings
 from django.utils.translation import ugettext_lazy as _
-from rest_framework.generics import ListCreateAPIView
+from rest_framework.generics import RetrieveAPIView, CreateAPIView
 from rest_framework.response import Response
+from rest_framework import status
 
 from common.permissions import IsValidUser
-from ..mfa import MFAOtp
 from ..const import ConfirmType
-from ..mixins import authenticate
 from ..serializers import ConfirmSerializer
 
 
-class ConfirmViewSet(ListCreateAPIView):
+class ConfirmApi(RetrieveAPIView, CreateAPIView):
     permission_classes = (IsValidUser,)
     serializer_class = ConfirmSerializer
 
-    def check(self, confirm_type: str):
-        if confirm_type == ConfirmType.MFA:
-            return self.user.mfa_enabled
+    def get_confirm_backend(self, confirm_type):
+        backend_classes = ConfirmType.get_can_confirm_backend_classes(confirm_type)
+        if not backend_classes:
+            return
+        for backend_cls in backend_classes:
+            backend = backend_cls(self.request.user, self.request)
+            if not backend.check():
+                continue
+            return backend
 
-        if confirm_type == ConfirmType.PASSWORD:
-            return self.user.is_password_authenticate()
+    def retrieve(self, request, *args, **kwargs):
+        confirm_type = request.query_params.get('confirm_type')
+        backend = self.get_confirm_backend(confirm_type)
+        if backend is None:
+            msg = _('This action require verify your MFA')
+            return Response(data={'error': msg}, status=status.HTTP_404_NOT_FOUND)
 
-        if confirm_type == ConfirmType.RELOGIN:
-            return not self.user.is_password_authenticate()
-
-    def authenticate(self, confirm_type, secret_key):
-        if confirm_type == ConfirmType.MFA:
-            ok, msg = MFAOtp(self.user).check_code(secret_key)
-            return ok, msg
-
-        if confirm_type == ConfirmType.PASSWORD:
-            ok = authenticate(self.request, username=self.user.username, password=secret_key)
-            msg = '' if ok else _('Authentication failed password incorrect')
-            return ok, msg
-
-        if confirm_type == ConfirmType.RELOGIN:
-            now = timezone.now().strftime("%Y-%m-%d %H:%M:%S")
-            now = datetime.strptime(now, '%Y-%m-%d %H:%M:%S')
-            login_time = self.request.session.get('login_time')
-            SPECIFIED_TIME = 5
-            msg = _('Login time has exceeded {} minutes, please login again').format(SPECIFIED_TIME)
-            if not login_time:
-                return False, msg
-            login_time = datetime.strptime(login_time, '%Y-%m-%d %H:%M:%S')
-            if (now - login_time).seconds >= SPECIFIED_TIME * 60:
-                return False, msg
-            return True, ''
-
-    @property
-    def user(self):
-        return self.request.user
-
-    def list(self, request, *args, **kwargs):
-        if not settings.SECURITY_VIEW_AUTH_NEED_MFA:
-            return Response('ok')
-
-        mfa_verify_time = request.session.get('MFA_VERIFY_TIME', 0)
-        if time.time() - mfa_verify_time < settings.SECURITY_MFA_VERIFY_TTL:
-            return Response('ok')
-
-        data = []
-        for i, confirm_type in enumerate(ConfirmType.values, 1):
-            if self.check(confirm_type):
-                data.append({'name': confirm_type, 'level': i})
-        msg = _('This action require verify your MFA')
-        return Response({'error': msg, 'backends': data}, status=400)
+        data = {
+            'confirm_type': backend.name,
+            'content': backend.content,
+        }
+        return Response(data=data)
 
     def create(self, request, *args, **kwargs):
         serializer = self.get_serializer(data=request.data)
         serializer.is_valid(raise_exception=True)
         validated_data = serializer.validated_data
+
         confirm_type = validated_data.get('confirm_type')
+        mfa_type = validated_data.get('mfa_type')
         secret_key = validated_data.get('secret_key')
-        ok, msg = self.authenticate(confirm_type, secret_key)
+
+        backend = self.get_confirm_backend(confirm_type)
+        ok, msg = backend.authenticate(secret_key, mfa_type)
         if ok:
-            request.session["MFA_VERIFY_TIME"] = int(time.time())
+            request.session['CONFIRM_LEVEL'] = ConfirmType.values.index(confirm_type) + 1
+            request.session['CONFIRM_TIME'] = int(time.time())
             return Response('ok')
         return Response({'error': msg}, status=400)