feat: Captcha (#2913)

Author: shaohuzhang1

apps/common/constants/cache_version.py

@@ -21,6 +21,8 @@ class Cache_Version(Enum):
     # 当前用户所有权限
     PERMISSION_LIST = "PERMISSION:LIST", lambda user_id: user_id
 
+    CAPTCHA = "CAPTCHA", lambda captcha: captcha
+
     def get_version(self):
         return self.value[0]
 

apps/common/utils/common.py

@@ -7,6 +7,8 @@
     @desc:
 """
 import hashlib
+
+import random
 import io
 import mimetypes
 import re
@@ -48,6 +50,13 @@ def group_by(list_source: List, key):
     return result
 
 
+
+CHAR_SET = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
+
+
+def get_random_chars(number=6):
+    return "".join([CHAR_SET[random.randint(0, len(CHAR_SET) - 1)] for index in range(number)])
+
 def encryption(message: str):
     """
         加密敏感字段数据  加密方式是 如果密码是 1234567890  那么给前端则是 123******890

apps/locales/en_US/LC_MESSAGES/django.po

@@ -102,3 +102,12 @@ msgstr ""
 #: .\apps\users\views\user.py:24 .\apps\users\views\user.py:25
 msgid "Get current user information"
 msgstr ""
+
+msgid "Get captcha"
+msgstr ""
+
+msgid "captcha"
+msgstr ""
+
+msgid "Captcha code error or expiration"
+msgstr ""

apps/locales/zh_CN/LC_MESSAGES/django.po

@@ -103,4 +103,11 @@ msgstr "用户管理"
 msgid "Get current user information"
 msgstr "获取当前用户信息"
 
+msgid "Get captcha"
+msgstr "获取验证码"
 
+msgid "captcha"
+msgstr "验证码"
+
+msgid "Captcha code error or expiration"
+msgstr "验证码错误或过期"

apps/locales/zh_Hant/LC_MESSAGES/django.po

@@ -102,3 +102,12 @@ msgstr "用戶管理"
 #: .\apps\users\views\user.py:24 .\apps\users\views\user.py:25
 msgid "Get current user information"
 msgstr "獲取當前用戶資訊"
+
+msgid "Get captcha"
+msgstr "獲取驗證碼"
+
+msgid "captcha"
+msgstr "驗證碼"
+
+msgid "Captcha code error or expiration"
+msgstr "驗證碼錯誤或過期"

apps/users/api/login.py

@@ -9,7 +9,7 @@
 
 from common.mixins.api_mixin import APIMixin
 from common.result import ResultSerializer
-from users.serializers.login import LoginResponse, LoginRequest
+from users.serializers.login import LoginResponse, LoginRequest, CaptchaResponse
 
 
 class ApiLoginResponse(ResultSerializer):
@@ -40,3 +40,14 @@ def get_request():
     @staticmethod
     def get_response():
         return ApiLoginResponse
+
+
+class ApiCaptchaResponse(ResultSerializer):
+    def get_data(self):
+        return CaptchaResponse()
+
+
+class CaptchaAPI(APIMixin):
+    @staticmethod
+    def get_response():
+        return ApiCaptchaResponse

apps/users/serializers/login.py

@@ -6,8 +6,10 @@
     @date：2025/4/14 11:08
     @desc:
 """
+import base64
 import datetime
 
+from captcha.image import ImageCaptcha
 from django.core import signing
 from django.core.cache import cache
 from django.db.models import QuerySet
@@ -17,13 +19,14 @@
 from common.constants.authentication_type import AuthenticationType
 from common.constants.cache_version import Cache_Version
 from common.exception.app_exception import AppApiException
-from common.utils.common import password_encrypt
+from common.utils.common import password_encrypt, get_random_chars
 from users.models import User
 
 
 class LoginRequest(serializers.Serializer):
     username = serializers.CharField(required=True, max_length=64, help_text=_("Username"), label=_("Username"))
     password = serializers.CharField(required=True, max_length=128, label=_("Password"))
+    captcha = serializers.CharField(required=True, max_length=64, label=_('captcha'))
 
 
 class LoginResponse(serializers.Serializer):
@@ -40,6 +43,11 @@ def login(instance):
         LoginRequest(data=instance).is_valid(raise_exception=True)
         username = instance.get('username')
         password = instance.get('password')
+        captcha = instance.get('captcha')
+        captcha_cache = cache.get(Cache_Version.CAPTCHA.get_key(captcha=captcha),
+                                  version=Cache_Version.CAPTCHA.get_version())
+        if captcha_cache is None:
+            raise AppApiException(1005, _("Captcha code error or expiration"))
         user = QuerySet(User).filter(username=username, password=password_encrypt(password)).first()
         if user is None:
             raise AppApiException(500, _('The username or password is incorrect'))
@@ -52,3 +60,22 @@ def login(instance):
         version, get_key = Cache_Version.TOKEN.value
         cache.set(get_key(token), user, timeout=datetime.timedelta(seconds=60 * 60 * 2).seconds, version=version)
         return {'token': token}
+
+
+class CaptchaResponse(serializers.Serializer):
+    """
+       登录响应对象
+       """
+    captcha = serializers.CharField(required=True, label=_("captcha"))
+
+
+class CaptchaSerializer(serializers.Serializer):
+    @staticmethod
+    def generate():
+        chars = get_random_chars()
+        image = ImageCaptcha()
+        data = image.generate(chars)
+        captcha = base64.b64encode(data.getbuffer())
+        cache.set(Cache_Version.CAPTCHA.get_key(captcha=chars), chars,
+                  timeout=60, version=Cache_Version.CAPTCHA.get_version())
+        return {'captcha': 'data:image/png;base64,' + captcha.decode()}

apps/users/urls.py

@@ -6,6 +6,7 @@
 urlpatterns = [
     path('user/login', views.LoginView.as_view(), name='login'),
     path('user/profile', views.UserProfileView.as_view(), name="user_profile"),
+    path('user/captcha', views.CaptchaView.as_view(), name='captcha'),
     path('user/test', views.TestPermissionsUserView.as_view(), name="test"),
     path('workspace/<str:workspace_id>/user/profile', views.TestWorkspacePermissionUserView.as_view(),
          name="test_workspace_id_permission")

apps/users/views/login.py

@@ -12,8 +12,8 @@
 from rest_framework.views import APIView
 
 from common import result
-from users.api.login import LoginAPI
-from users.serializers.login import LoginSerializer
+from users.api.login import LoginAPI, CaptchaAPI
+from users.serializers.login import LoginSerializer, CaptchaSerializer
 
 
 class LoginView(APIView):
@@ -25,3 +25,13 @@ class LoginView(APIView):
                    responses=LoginAPI.get_response())
     def post(self, request: Request):
         return result.success(LoginSerializer().login(request.data))
+
+
+class CaptchaView(APIView):
+    @extend_schema(methods=['GET'],
+                   description=_("Get captcha"),
+                   operation_id=_("Get captcha"),
+                   tags=[_("User management")],
+                   responses=CaptchaAPI.get_response())
+    def get(self, request: Request):
+        return result.success(CaptchaSerializer().generate())

pyproject.toml

@@ -15,6 +15,7 @@ psycopg = { extras = ["binary"], version = "3.2.6" }
 python-dotenv = "1.1.0"
 uuid-utils = "0.10.0"
 diskcache2 = "0.1.2"
+captcha = "0.7.1"
 langchain-openai = "^0.3.0"
 langchain-anthropic = "^0.3.0"
 langchain-community = "^0.3.0"