perf: Allow admins to configure available MFA services for user auth

apps/authentication/forms.py

@@ -50,7 +50,7 @@ class UserLoginForm(forms.Form):
 
 class UserCheckOtpCodeForm(forms.Form):
     code = forms.CharField(label=_('MFA Code'), max_length=128, required=False)
-    mfa_type = forms.CharField(label=_('MFA type'), max_length=128)
+    mfa_type = forms.CharField(label=_('MFA type'), max_length=128, required=False)
 
 
 class CustomCaptchaTextInput(CaptchaTextInput):

apps/authentication/mfa/base.py

@@ -72,10 +72,9 @@ class BaseMFA(abc.ABC):
     def is_active(self):
         return False
 
-    @staticmethod
-    @abc.abstractmethod
-    def global_enabled():
-        return False
+    @classmethod
+    def global_enabled(cls):
+        return cls.name in settings.SECURITY_MFA_ENABLED_BACKENDS
 
     @abc.abstractmethod
     def get_enable_url(self) -> str:

apps/authentication/mfa/custom.py

@@ -39,9 +39,9 @@ class MFACustom(BaseMFA):
     def is_active(self):
         return True
 
-    @staticmethod
-    def global_enabled():
-        return settings.MFA_CUSTOM and callable(mfa_custom_method)
+    @classmethod
+    def global_enabled(cls):
+        return super().global_enabled() and settings.MFA_CUSTOM and callable(mfa_custom_method)
 
     def get_enable_url(self) -> str:
         return ''

apps/authentication/mfa/email.py

@@ -50,9 +50,9 @@ class MFAEmail(BaseMFA):
         )
         sender_util.gen_and_send_async()
 
-    @staticmethod
-    def global_enabled():
-        return settings.SECURITY_MFA_BY_EMAIL
+    @classmethod
+    def global_enabled(cls):
+        return super().global_enabled and settings.SECURITY_MFA_BY_EMAIL
 
     def disable(self):
         return '/ui/#/profile/index'

apps/authentication/mfa/face.py

@@ -29,9 +29,10 @@ class MFAFace(BaseMFA, AuthFaceMixin):
             return True
         return bool(self.user.face_vector)
 
-    @staticmethod
-    def global_enabled():
+    @classmethod
+    def global_enabled(cls):
         return (
+                super().global_enabled() and
                 settings.XPACK_LICENSE_IS_VALID and
                 settings.XPACK_LICENSE_EDITION_ULTIMATE and
                 settings.FACE_RECOGNITION_ENABLED

apps/authentication/mfa/otp.py

@@ -25,10 +25,6 @@ class MFAOtp(BaseMFA):
             return True
         return self.user.otp_secret_key
 
-    @staticmethod
-    def global_enabled():
-        return True
-
     def get_enable_url(self) -> str:
         return reverse('authentication:user-otp-enable-start')
 

apps/authentication/mfa/passkey.py

@@ -23,9 +23,9 @@ class MFAPasskey(BaseMFA):
             return False
         return self.user.passkey_set.count()
 
-    @staticmethod
-    def global_enabled():
-        return settings.AUTH_PASSKEY
+    @classmethod
+    def global_enabled(cls):
+        return super().global_enabled() and settings.AUTH_PASSKEY
 
     def get_enable_url(self) -> str:
         return '/ui/#/profile/passkeys'

apps/authentication/mfa/radius.py

@@ -27,9 +27,9 @@ class MFARadius(BaseMFA):
     def is_active(self):
         return True
 
-    @staticmethod
-    def global_enabled():
-        return settings.OTP_IN_RADIUS
+    @classmethod
+    def global_enabled(cls):
+        return super().global_enabled() and settings.OTP_IN_RADIUS
 
     def get_enable_url(self) -> str:
         return ''

apps/authentication/mfa/sms.py

@@ -46,9 +46,9 @@ class MFASms(BaseMFA):
     def send_challenge(self):
         self.sms.gen_and_send_async()
 
-    @staticmethod
-    def global_enabled():
-        return settings.SMS_ENABLED
+    @classmethod
+    def global_enabled(cls):
+        return super().global_enabled() and settings.SMS_ENABLED
 
     def get_enable_url(self) -> str:
         return '/ui/#/profile/index'

apps/authentication/templates/authentication/login.html

@@ -376,7 +376,7 @@
                     </div>
                     {% if form.challenge %}
                         {% bootstrap_field form.challenge show_label=False %}
-                    {% elif form.mfa_type %}
+                    {% elif form.mfa_type and mfa_backends %}
                         <div class="form-group" style="display: flex">
                             {% include '_mfa_login_field.html' %}
                         </div>

apps/jumpserver/conf.py

@@ -569,7 +569,7 @@ class Config(dict):
         'SAFE_MODE': False,
         'SECURITY_MFA_AUTH': 0,  # 0 不开启 1 全局开启 2 管理员开启
         'SECURITY_MFA_AUTH_ENABLED_FOR_THIRD_PARTY': True,
-        'SECURITY_MFA_BY_EMAIL': False,
+        'SECURITY_MFA_ENABLED_BACKENDS': [],
         'SECURITY_COMMAND_EXECUTION': False,
         'SECURITY_COMMAND_BLACKLIST': [
             'reboot', 'shutdown', 'poweroff', 'halt', 'dd', 'half', 'top'

apps/settings/serializers/security.py

@@ -1,3 +1,7 @@
+import importlib
+import os
+
+from django.conf import settings
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
@@ -117,6 +121,35 @@ class SecurityLoginLimitSerializer(serializers.Serializer):
     )
 
 
+class DynamicMFAChoiceField(serializers.MultipleChoiceField):
+    def __init__(self, **kwargs):
+        _choices = self._get_dynamic_choices()
+        super().__init__(choices=_choices, **kwargs)
+
+    @staticmethod
+    def _get_dynamic_choices():
+        choices = []
+        mfa_dir = os.path.join(settings.APPS_DIR, 'authentication', 'mfa')
+        for filename in os.listdir(mfa_dir):
+            if not filename.endswith('.py') or filename.startswith('__init__'):
+                continue
+
+            module_name = f'authentication.mfa.{filename[:-3]}'
+            try:
+                module = importlib.import_module(module_name)
+            except ImportError:
+                continue
+
+            for attr_name in dir(module):
+                item = getattr(module, attr_name)
+                if not isinstance(item, type) or not attr_name.startswith('MFA'):
+                    continue
+                if 'BaseMFA' != item.__base__.__name__:
+                    continue
+                choices.append((item.name, item.display_name))
+        return choices
+
+
 class SecurityAuthSerializer(serializers.Serializer):
     SECURITY_MFA_AUTH = serializers.ChoiceField(
         choices=(
@@ -130,10 +163,10 @@ class SecurityAuthSerializer(serializers.Serializer):
         required=False, default=True,
         label=_('Third-party login MFA'),
     )
-    SECURITY_MFA_BY_EMAIL = serializers.BooleanField(
-        required=False, default=False,
-        label=_('MFA via Email'),
-        help_text=_('Email as a method for multi-factor authentication')
+    SECURITY_MFA_ENABLED_BACKENDS = DynamicMFAChoiceField(
+        default=[], allow_empty=True,
+        label=_('MFA Backends'),
+        help_text=_('MFA methods supported for user login')
     )
     OTP_ISSUER_NAME = serializers.CharField(
         required=False, max_length=16, label=_('OTP issuer name'),

apps/users/views/profile/otp.py

@@ -37,12 +37,25 @@ logger = get_logger(__name__)
 class UserOtpEnableStartView(AuthMixin, TemplateView):
     template_name = 'users/user_otp_check_password.html'
 
+    @staticmethod
+    def get_redirect_url():
+        message_data = {
+            'title': _('Redirecting'),
+            'message': _('No MFA services are available. Please contact the administrator'),
+            'redirect_url': reverse('authentication:login'),
+            'auto_redirect': True,
+        }
+        return FlashMessageUtil.gen_message_url(message_data)
+
     def get(self, request, *args, **kwargs):
         try:
             self.get_user_from_session()
         except SessionEmptyError:
             url = reverse('authentication:login') + '?_=otp_enable_start'
             return redirect(url)
+
+        if not MFAOtp.global_enabled():
+            return redirect(self.get_redirect_url())
         return super().get(request, *args, **kwargs)