feat: enhance authentication flow with access token support and captcha validation

apps/chat/serializers/chat_user_serializer.py

@@ -10,87 +10,169 @@
 from common.utils.common import password_encrypt
 from common.utils.common import password_verify, needs_password_upgrade
 from common.utils.rsa_util import decrypt
+from portal.models import Portal
 from system_manage.models import ChatUser
 from users.serializers.login import LoginRequest
 
 system_version, system_get_key = Cache_Version.SYSTEM.value
 
 
 class ChatUserAccessTokenV3Serializer(serializers.Serializer):
-    @staticmethod
-    def get_auth_setting():
-        application_access_token = ApplicationAccessToken.objects.filter(is_active=True).first()
+    @classmethod
+    def get_auth_setting(cls, access_token):
+        application_access_token = ApplicationAccessToken.objects.filter(
+            access_token=access_token, is_active=True
+        ).first()
 
         if not application_access_token:
             raise AppApiException(1005, _("Invalid access token"))
 
         return application_access_token.authentication_value
 
-    @staticmethod
-    def local_login(instance):
+    @classmethod
+    def _get_login_policy(cls, access_token):
+        """返回 (scope, max_attempts, failed_attempts, lock_time)；scope 作为缓存 key 前缀。"""
+        if access_token:
+            auth_setting = cls.get_auth_setting(access_token)
+            return (
+                f"chat_{access_token}",
+                auth_setting.get("max_attempts", 1),
+                # 应用未配置锁定阈值时默认不锁定，保持原有行为
+                auth_setting.get("failed_attempts", 5),
+                auth_setting.get("lock_time", 10),
+            )
+        portal = Portal.objects.first()
+        if portal is None or not portal.enable_auth:
+            raise AppApiException(500, _("Portal authentication is not enabled"))
+        auth_config = portal.auth_config or {}
+        if "LOCAL" not in auth_config.get("login_value", []):
+            raise AppApiException(500, _("Portal local login is not enabled"))
+        return (
+            "portal",
+            auth_config.get("max_attempts", 1),
+            auth_config.get("failed_attempts", 5),
+            auth_config.get("lock_time", 10),
+        )
+
+    @classmethod
+    def _is_account_locked(cls, scope, username, failed_attempts):
+        if failed_attempts == -1:
+            return False
+        return bool(cache.get(system_get_key(f"{scope}_{username}_lock"), version=system_version))
+
+    @classmethod
+    def _need_captcha(cls, scope, username, max_attempts):
+        if max_attempts == -1:
+            return False
+        if max_attempts > 0:
+            fail_count = cache.get(system_get_key(f"{scope}_{username}"), version=system_version) or 0
+            return fail_count >= max_attempts
+        return True
+
+    @classmethod
+    def _validate_captcha(cls, scope, username, captcha, failed_attempts=5, lock_time=10):
+        if not captcha:
+            raise AppApiException(1005, _("Captcha is required"))
+        captcha_key = Cache_Version.CAPTCHA.get_key(captcha=f"{scope}_{username}")
+        captcha_cache = cache.get(captcha_key, version=Cache_Version.CAPTCHA.get_version())
+        if captcha_cache is None or captcha.lower() != captcha_cache:
+            # 校验失败与口令失败共用同一失败计数与锁定机制，防止"识别-试错"循环绕过验证码
+            cls._record_login_failure(scope, username, failed_attempts, lock_time)
+            raise AppApiException(1005, _("Captcha code error or expiration"))
+        # 校验通过即销毁，保证验证码一次性使用
+        cache.delete(captcha_key, version=Cache_Version.CAPTCHA.get_version())
+
+    @classmethod
+    def _record_login_failure(cls, scope, username, failed_attempts, lock_time):
+        """记录一次认证失败（口令或验证码），累计计数并在达到阈值时创建锁键；不抛异常。"""
+        try:
+            _record_login_fail(scope, username)
+        except Exception:
+            pass
+        lock_fail_count = 0
+        try:
+            lock_fail_count = _record_login_fail_lock(scope, username, lock_time)
+        except Exception:
+            pass
+        if failed_attempts > 0 and lock_fail_count >= failed_attempts:
+            try:
+                cache.add(system_get_key(f"{scope}_{username}_lock"), 1, timeout=lock_time * 60, version=system_version)
+            except Exception:
+                pass
+        return lock_fail_count
+
+    @classmethod
+    def _handle_failed_login(cls, scope, username, failed_attempts, lock_time):
+        lock_fail_count = cls._record_login_failure(scope, username, failed_attempts, lock_time)
+        # 仅由失败次数配置控制（CE/PE 同样生效）；计数在此之前已记录
+        if failed_attempts <= 0:
+            return
+        if lock_fail_count < failed_attempts:
+            remain_attempts = failed_attempts - lock_fail_count
+            raise AppApiException(
+                1005,
+                _("Login failed %s times, account will be locked, you have %s more chances !")
+                % (failed_attempts, remain_attempts),
+            )
+        raise AppApiException(
+            1005, _("This account has been locked for %s minutes, please try again later") % lock_time
+        )
+
+    @classmethod
+    def local_login(cls, instance, access_token=None):
         username = instance.get("username", "")
         encryptedData = instance.get("encryptedData", "")
         if encryptedData:
             json_data = json.loads(decrypt(encryptedData))
             instance.update(json_data)
-        try:
-            LoginRequest(data=instance).is_valid(raise_exception=True)
-        except Exception as e:
-            raise e
-        auth_setting = ChatUserAccessTokenV3Serializer.get_auth_setting()
+        LoginRequest(data=instance).is_valid(raise_exception=True)
 
-        max_attempts = auth_setting.get("max_attempts", 1)
-        password = instance.get("password")
-        captcha = instance.get("captcha", "")
+        scope, max_attempts, failed_attempts, lock_time = cls._get_login_policy(access_token)
 
-        # 判断是否需要验证码
-        need_captcha = True
-        if max_attempts == -1:
-            need_captcha = False
-        elif max_attempts > 0:
-            fail_count = cache.get(system_get_key(f"chat_{username}"), version=system_version) or 0
-            need_captcha = fail_count >= max_attempts
+        if cls._is_account_locked(scope, username, failed_attempts):
+            raise AppApiException(
+                1005, _("This account has been locked for %s minutes, please try again later") % lock_time
+            )
 
-        if need_captcha:
-            ChatUserAccessTokenV3Serializer._validate_captcha(username, captcha)
+        password = instance.get("password")
+        captcha = instance.get("captcha", "")
+        if cls._need_captcha(scope, username, max_attempts):
+            cls._validate_captcha(scope, username, captcha, failed_attempts, lock_time)
 
         user = ChatUser.objects.filter(username=username).first()
 
         if not user or not password_verify(password, user.password):
-            record_login_fail(username)
+            cls._handle_failed_login(scope, username, failed_attempts, lock_time)
             raise AppApiException(500, _("The username or password is incorrect"))
 
         if needs_password_upgrade(user.password):
             user.password = password_encrypt(password)
             user.save(update_fields=["password"])
         if not user.is_active:
             raise AppApiException(1005, _("The user has been disabled, please contact the administrator!"))
-        cache.delete(system_get_key(f"chat_{username}"), version=system_version)
+        cache.delete(system_get_key(f"{scope}_{username}"), version=system_version)
+        cache.delete(system_get_key(f"{scope}_{username}_lock_count"), version=system_version)
+        cache.delete(system_get_key(f"{scope}_{username}_lock"), version=system_version)
         return user
 
-    @staticmethod
-    def _validate_captcha(username: str, captcha: str) -> None:
-        """验证验证码（一次性消费）"""
-        if not captcha:
-            raise AppApiException(1005, _("Captcha is required"))
-
-        captcha_key = Cache_Version.CAPTCHA.get_key(captcha=f"chat_{username}")
-        captcha_cache = cache.get(captcha_key, version=Cache_Version.CAPTCHA.get_version())
-
-        if captcha_cache is None or captcha.lower() != captcha_cache:
-            record_login_fail(username)
-            raise AppApiException(1005, _("Captcha code error or expiration"))
-
-        # 校验通过即销毁，保证验证码一次性使用
-        cache.delete(captcha_key, version=Cache_Version.CAPTCHA.get_version())
 
-
-def record_login_fail(username: str, expire: int = 600):
-    """记录登录失败次数（原子递增）"""
+def _record_login_fail(scope, username, expire=600):
     if not username:
         return
-    fail_key = system_get_key(f"chat_{username}")
+    fail_key = system_get_key(f"{scope}_{username}")
     try:
         cache.incr(fail_key, 1, version=system_version)
     except ValueError:
         cache.set(fail_key, 1, timeout=expire, version=system_version)
+
+
+def _record_login_fail_lock(scope, username, expire=10):
+    if not username:
+        return 0
+    lock_key = system_get_key(f"{scope}_{username}_lock_count")
+    try:
+        fail_count = cache.incr(lock_key, 1, version=system_version)
+    except ValueError:
+        cache.set(lock_key, 1, timeout=expire * 60, version=system_version)
+        fail_count = 1
+    return fail_count

apps/chat/views/v3/chat.py

@@ -341,8 +341,10 @@ class CaptchaView(APIView):
     )
     def get(self, request: Request):
         username = request.query_params.get("username", None)
-        application_id = request.query_params.get("application_id", None)
-        return result.success(CaptchaSerializer().chat_generate(username, "chat", application_id))
+        access_token = request.query_params.get("accessToken")
+        if access_token:
+            return result.success(CaptchaSerializer().chat_generate(username, "chat", access_token))
+        return result.success(CaptchaSerializer().portal_generate(username))
 
 
 class SpeechToText(APIView):
@@ -516,9 +518,9 @@ class LocalLoginView(BaseAuthView):
         responses=LoginAPI.get_response(),
     )
     def post(self, request: Request):
-        user = ChatUserAccessTokenV3Serializer.local_login(request.data)
-        user.source = "LOCAL"
         access_token = request.query_params.get("accessToken")
+        user = ChatUserAccessTokenV3Serializer.local_login(request.data, access_token)
+        user.source = "LOCAL"
         token = self.create_token_and_cache(user, access_token, Operate.LOCAL)
         response = result.success({"token": token})
         return self.generate(request, token, response, path=f"/chat/{access_token + '/' if access_token else ''}")