From 39afdadf7fb5745f7cd4e1d8182c6451b8b9a663 Mon Sep 17 00:00:00 2001 From: wxg0103 <727495428@qq.com> Date: Thu, 22 Oct 2026 17:18:50 +0800 Subject: [PATCH] feat: enhance authentication flow with access token support and captcha validation --- apps/chat/serializers/chat_user_serializer.py | 168 +++++++++++---- apps/chat/views/v3/chat.py | 10 +- apps/portal/serializers/portal.py | 192 +----------------- apps/portal/urls.py | 2 - apps/portal/views/portal.py | 45 ---- .../serializers/chat_user_serializer.py | 18 +- apps/users/serializers/login.py | 25 ++- 7 files changed, 162 insertions(+), 298 deletions(-) diff --git a/apps/chat/serializers/chat_user_serializer.py b/apps/chat/serializers/chat_user_serializer.py index ee9c6f76dc6..38b9afbfd15 100644 --- a/apps/chat/serializers/chat_user_serializer.py +++ b/apps/chat/serializers/chat_user_serializer.py @@ -10,6 +10,7 @@ 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 @@ -17,47 +18,131 @@ 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): @@ -65,32 +150,29 @@ def local_login(instance): 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 diff --git a/apps/chat/views/v3/chat.py b/apps/chat/views/v3/chat.py index 83298c54ab7..f53a03c1c76 100644 --- a/apps/chat/views/v3/chat.py +++ b/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 ''}") diff --git a/apps/portal/serializers/portal.py b/apps/portal/serializers/portal.py index 6494ad39814..c6df825d637 100644 --- a/apps/portal/serializers/portal.py +++ b/apps/portal/serializers/portal.py @@ -7,30 +7,15 @@ @desc: 门户配置序列化器 """ -import json import uuid_utils.compat as uuid -from django.core.cache import cache from django.utils.translation import gettext_lazy as _ from rest_framework import serializers -from common.auth.common import ChatToken -from common.auth.constants.operate_constants import Operate -from common.constants.authentication_type import AuthenticationType -from common.constants.cache_version import Cache_Version from common.database_model_manage.database_model_manage import DatabaseModelManage from common.exception.app_exception import AppApiException -from common.log.log import record_log -from common.utils.common import password_verify, needs_password_upgrade, password_encrypt -from common.utils.rsa_util import decrypt, get_key_pair_by_sql +from common.utils.rsa_util import get_key_pair_by_sql from knowledge.models import File, FileSourceType -from maxkb.const import CONFIG from portal.models import Portal -from system_manage.models.chat_user import ( - ChatUser, -) -from users.serializers.login import LoginRequest - -system_version, system_get_key = Cache_Version.SYSTEM.value class LogoField(serializers.CharField): @@ -170,93 +155,6 @@ def edit(self, instance, with_valid=True): class PortalLoginSerializer(serializers.Serializer): - @staticmethod - def login(instance): - username = instance.get("username", "") - encrypted_data = instance.get("encryptedData", "") - - if encrypted_data: - try: - decrypted_raw = decrypt(encrypted_data) - decrypted_data = json.loads(decrypted_raw) if decrypted_raw else {} - if isinstance(decrypted_data, dict): - instance.update(decrypted_data) - except Exception: - raise AppApiException(500, _("Invalid encrypted data")) - - try: - request_serializer = LoginRequest(data=instance) - request_serializer.is_valid(raise_exception=True) - except serializers.ValidationError: - raise - except Exception as e: - raise AppApiException(500, str(e)) - - validated_data = request_serializer.validated_data - username = validated_data.get("username", "") - password = validated_data.get("password", "") - captcha = validated_data.get("captcha", "") - - 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 {} - login_value = auth_config.get("login_value", []) - if "LOCAL" not in login_value: - raise AppApiException(500, _("Portal local login is not enabled")) - - max_attempts = auth_config.get("max_attempts", 1) - - license_validator = DatabaseModelManage.get_model("license_is_valid") or (lambda: False) - is_license_valid = license_validator() if license_validator() is not None else False - - if is_license_valid: - failed_attempts = auth_config.get("failed_attempts", 5) - lock_time = auth_config.get("lock_time", 10) - else: - failed_attempts = 5 - lock_time = 10 - - cache_key = system_get_key(f"portal_{username}") - - if PortalLoginSerializer._is_account_locked(username, failed_attempts): - raise AppApiException( - 1005, _("This account has been locked for %s minutes, please try again later") % lock_time - ) - if PortalLoginSerializer._need_captcha(username, max_attempts): - PortalLoginSerializer._validate_captcha(username, captcha, failed_attempts, lock_time) - - user = ChatUser.objects.filter(username=username).first() - - if not user or not password_verify(password, user.password): - PortalLoginSerializer._handle_failed_login(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(cache_key, version=system_version) - cache.delete(system_get_key(f"portal_{username}_lock"), version=system_version) - - token = ChatToken(str(user.id), AuthenticationType.CHAT_USER, str(Operate.LOCAL)).to_token() - version, get_key = Cache_Version.CHAT_USER_TOKEN.value - timeout = CONFIG.get_session_timeout() - cache.set(get_key(token), user, timeout=timeout, version=version) - record_log( - menu="Portal", - operate="Log in", - request=None, - user={"username": user.username}, - status=200, - operation_object={"name": user.username}, - workspace_id="default", - ) - return {"token": token} - @staticmethod def get_login_profile(): portal = Portal.objects.first() @@ -273,91 +171,3 @@ def get_login_profile(): "max_attempts": auth_config.get("max_attempts", 1) if portal.enable_auth else 1, "rsa_key": get_key_pair_by_sql().get("key", ""), } - - @staticmethod - def _is_account_locked(username: str, failed_attempts: int) -> bool: - if failed_attempts == -1: - return False - lock_cache = cache.get(system_get_key(f"portal_{username}_lock"), version=system_version) - return bool(lock_cache) - - @staticmethod - def _need_captcha(username: str, max_attempts: int) -> bool: - cache_key = system_get_key(f"portal_{username}") - if max_attempts == -1: - return False - if max_attempts > 0: - fail_count = cache.get(cache_key, version=system_version) or 0 - return fail_count >= max_attempts - return True - - @staticmethod - def _validate_captcha(username: str, captcha: str, failed_attempts: int = 5, lock_time: int = 10) -> None: - if not captcha: - raise AppApiException(1005, _("Captcha is required")) - captcha_key = Cache_Version.CAPTCHA.get_key(captcha=f"portal_{username}") - captcha_cache = cache.get(captcha_key, version=Cache_Version.CAPTCHA.get_version()) - if captcha_cache is None or captcha.lower() != captcha_cache: - # 校验失败与口令失败共用同一失败计数与锁定机制，防止"识别-试错"循环绕过验证码 - PortalLoginSerializer._record_login_failure(username, failed_attempts, lock_time) - raise AppApiException(1005, _("Captcha code error or expiration")) - # 校验通过即销毁，保证验证码一次性使用 - cache.delete(captcha_key, version=Cache_Version.CAPTCHA.get_version()) - - @staticmethod - def _record_login_failure(username: str, failed_attempts: int, lock_time: int) -> int: - """记录一次认证失败（口令或验证码），累计计数并在达到阈值时创建锁键；不抛异常。""" - try: - _record_login_fail(username) - except Exception: - pass - lock_fail_count = 0 - try: - lock_fail_count = _record_login_fail_lock(username, lock_time) - except Exception: - pass - if failed_attempts > 0 and lock_fail_count >= failed_attempts: - try: - cache.add(system_get_key(f"portal_{username}_lock"), 1, timeout=lock_time * 60, version=system_version) - except Exception: - pass - return lock_fail_count - - @staticmethod - def _handle_failed_login(username: str, failed_attempts: int, lock_time: int) -> None: - lock_fail_count = PortalLoginSerializer._record_login_failure(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 - ) - - -def _record_login_fail(username: str, expire: int = 600): - if not username: - return - fail_key = system_get_key(f"portal_{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(username: str, expire: int = 10): - if not username: - return 0 - lock_key = system_get_key(f"portal_{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 diff --git a/apps/portal/urls.py b/apps/portal/urls.py index 36fdce98d2c..f9c47084691 100644 --- a/apps/portal/urls.py +++ b/apps/portal/urls.py @@ -7,6 +7,4 @@ urlpatterns = [ path("portal", views.PortalView.as_view()), path("portal/info", views.PortalInfoView.as_view()), - path("portal/login", views.PortalLoginView.as_view()), - path("portal/logout", views.PortalLogoutView.as_view()), ] diff --git a/apps/portal/views/portal.py b/apps/portal/views/portal.py index a1fe30365a7..f5b4f441110 100644 --- a/apps/portal/views/portal.py +++ b/apps/portal/views/portal.py @@ -18,9 +18,7 @@ from common.auth.authentication import has_permissions from common.auth.constants.permission_constants import PermissionConstants from common.auth.constants.role_constants import RoleConstants -from common.constants.cache_version import Cache_Version from common.log.log import log -from django.core.cache import cache from portal.api.portal import PortalAPI from portal.serializers.portal import ( PortalSerializer, @@ -59,33 +57,6 @@ def put(self, request: Request): return result.success(PortalSerializer(data=request.data).edit(request.data)) -class PortalLoginView(APIView): - @extend_schema( - methods=["POST"], - description=_("Portal login"), - summary=_("Portal login"), - operation_id=_("Portal login"), # type: ignore - tags=[_("Portal")], # type: ignore - request=PortalAPI.Login.get_request(), - responses=PortalAPI.Login.get_response(), - ) - def post(self, request: Request): - token_data = PortalLoginSerializer.login(request.data) - response = result.success(token_data) - secure = request.is_secure() - response.set_cookie( - "mk_file_auth", - value=token_data.get("token"), - max_age=7 * 24 * 3600, - path="/portal/", - domain=None, - secure=secure, - httponly=True, - samesite="Lax", - ) - return response - - class PortalInfoView(APIView): @extend_schema( methods=["GET"], @@ -96,19 +67,3 @@ class PortalInfoView(APIView): ) def get(self, request: Request): return result.success(PortalLoginSerializer.get_login_profile()) - - -class PortalLogoutView(APIView): - @extend_schema( - methods=["POST"], - summary=_("Portal logout"), - description=_("Portal logout"), - operation_id=_("Portal logout"), # type: ignore - tags=[_("Portal")], # type: ignore - responses=PortalAPI.Logout.get_response(), - ) - @log(menu="Portal", operate="Log out") - def post(self, request: Request): - version, get_key = Cache_Version.TOKEN.value - cache.delete(get_key(token=request.META.get("HTTP_AUTHORIZATION")[7:]), version=version) - return result.success(True) diff --git a/apps/system_manage/serializers/chat_user_serializer.py b/apps/system_manage/serializers/chat_user_serializer.py index 279a1486c6d..c96b1173851 100644 --- a/apps/system_manage/serializers/chat_user_serializer.py +++ b/apps/system_manage/serializers/chat_user_serializer.py @@ -109,16 +109,16 @@ def local_login(instance, access_token): 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 + fail_count = cache.get(system_get_key(f"chat_{access_token}_{username}"), version=system_version) or 0 need_captcha = fail_count >= max_attempts if need_captcha: - ChatUserAccessTokenSerializer._validate_captcha(username, captcha) + ChatUserAccessTokenSerializer._validate_captcha(username, captcha, access_token) user = ChatUser.objects.filter(username=username).first() if not user or not password_verify(password, user.password): - record_login_fail(username) + record_login_fail(access_token, username) raise AppApiException(500, _("The username or password is incorrect")) if needs_password_upgrade(user.password): @@ -126,32 +126,32 @@ def local_login(instance, access_token): 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"chat_{access_token}_{username}"), version=system_version) return user @staticmethod - def _validate_captcha(username: str, captcha: str) -> None: + def _validate_captcha(username: str, captcha: str, access_token: str) -> None: """验证验证码（一次性消费）""" if not captcha: raise AppApiException(1005, _("Captcha is required")) - captcha_key = Cache_Version.CAPTCHA.get_key(captcha=f"chat_{username}") + captcha_key = Cache_Version.CAPTCHA.get_key(captcha=f"chat_{access_token}_{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) + record_login_fail(access_token, 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(access_token: str, username: str, expire: int = 600): """记录登录失败次数（原子递增）""" if not username: return - fail_key = system_get_key(f"chat_{username}") + fail_key = system_get_key(f"chat_{access_token}_{username}") try: cache.incr(fail_key, 1, version=system_version) except ValueError: diff --git a/apps/users/serializers/login.py b/apps/users/serializers/login.py index 4cdd9d31bec..35bbda84c1e 100644 --- a/apps/users/serializers/login.py +++ b/apps/users/serializers/login.py @@ -11,6 +11,7 @@ import json from application.models import ApplicationAccessToken +from portal.models import Portal from captcha.image import ImageCaptcha from common.auth.common import SystemToken from common.constants.authentication_type import AuthenticationType @@ -300,19 +301,35 @@ def chat_generate(username: str, type: str = "chat", access_token: str = ""): auth_setting = application_access_token.authentication_value max_attempts = auth_setting.get("max_attempts", 1) - need_captcha = LoginSerializer._need_captcha_by_key(system_get_key(f"{type}_{username}"), max_attempts) + # key 拼上 access_token，与应用登录的失败计数/验证码存储保持一致 + key = f"{type}_{username}" + if access_token: + key = f"{type}_{access_token}_{username}" + need_captcha = LoginSerializer._need_captcha_by_key(system_get_key(key), max_attempts) - return CaptchaSerializer._generate_captcha_if_needed(username, type, need_captcha) + return CaptchaSerializer._generate_captcha_if_needed(username, type, need_captcha, key) + + @staticmethod + def portal_generate(username: str, type: str = "portal"): + portal = Portal.objects.first() + if portal is None or not portal.enable_auth: + return {"captcha": ""} + auth_config = portal.auth_config or {} + max_attempts = auth_config.get("max_attempts", 1) + key = f"{type}_{username}" + need_captcha = LoginSerializer._need_captcha_by_key(system_get_key(key), max_attempts) + + return CaptchaSerializer._generate_captcha_if_needed(username, type, need_captcha, key) @staticmethod - def _generate_captcha_if_needed(username: str, type: str, need_captcha: bool): + def _generate_captcha_if_needed(username: str, type: str, need_captcha: bool, key: str = ""): """提取的公共验证码生成方法""" if need_captcha: chars = get_random_chars() data = ImageCaptcha().generate(chars) captcha = base64.b64encode(data.getbuffer()) cache.set( - Cache_Version.CAPTCHA.get_key(captcha=f"{type}_{username}"), + Cache_Version.CAPTCHA.get_key(captcha=key or f"{type}_{username}"), chars.lower(), timeout=300, version=Cache_Version.CAPTCHA.get_version(),