fix: replace unsalted MD5 password hashing with Django PBKDF2 (CWE-327)

apps/common/utils/common.py

@@ -17,6 +17,7 @@
 from functools import reduce
 from typing import List, Dict
 
+from django.contrib.auth.hashers import check_password, make_password
 from django.core.files.uploadedfile import InMemoryUploadedFile
 from django.db.models import QuerySet
 from django.utils.translation import gettext as _
@@ -26,16 +27,62 @@
 from ..exception.app_exception import AppApiException
 
 
+def _legacy_md5_hash(row_password):
+    """
+    Legacy MD5 hashing — used only to detect old hashes during migration.
+    Do NOT use for new passwords.
+    """
+    md5 = hashlib.md5()
+    md5.update(row_password.encode())
+    return md5.hexdigest()
+
+
 def password_encrypt(row_password):
     """
-    密码 md5加密
+    密码加密（使用 Django PBKDF2）
     :param row_password: 密码
     :return:  加密后密码
     """
-    md5 = hashlib.md5()  # 2，实例化md5() 方法
-    md5.update(row_password.encode())  # 3，对字符串的字节类型加密
-    result = md5.hexdigest()  # 4，加密
-    return result
+    return make_password(row_password)
+
+
+def password_verify(row_password, hashed_password):
+    """
+    验证密码是否匹配已存储的哈希值。
+    支持透明升级：如果存储的是旧版 MD5 哈希，也能正确验证。
+    :param row_password: 明文密码
+    :param hashed_password: 数据库中存储的密码哈希
+    :return: 是否匹配
+    """
+    # First try Django's built-in check (PBKDF2, bcrypt, argon2, etc.)
+    if check_password(row_password, hashed_password):
+        return True
+    # Fall back to legacy MD5 comparison for not-yet-migrated hashes
+    if _is_legacy_md5_hash(hashed_password):
+        return _legacy_md5_hash(row_password) == hashed_password
+    return False
+
+
+def _is_legacy_md5_hash(hashed_password):
+    """
+    Detect legacy unsalted MD5 hex-digest hashes (exactly 32 hex chars).
+    Django password hashes always contain '$' separators.
+    """
+    if hashed_password and len(hashed_password) == 32:
+        try:
+            int(hashed_password, 16)
+            return True
+        except ValueError:
+            pass
+    return False
+
+
+def needs_password_upgrade(hashed_password):
+    """
+    Check if a stored password hash should be upgraded to PBKDF2.
+    Returns True for legacy MD5 hashes.
+    """
+    return _is_legacy_md5_hash(hashed_password)
 
 
 def group_by(list_source: List, key):

apps/users/serializers/login.py

@@ -22,7 +22,7 @@
 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.utils.common import password_encrypt, get_random_chars
+from common.utils.common import password_encrypt, password_verify, needs_password_upgrade, get_random_chars
 from common.utils.rsa_util import decrypt
 from maxkb.const import CONFIG
 from users.models import User
@@ -65,7 +65,7 @@ def record_login_fail(username: str, expire: int = 600):
 def record_login_fail_lock(username: str, expire: int = 10):
     """
     使用 cache.incr 保证原子递增，并在不存在时初始化计数器并返回当前值。
-    这里的计数器用于判断是否应当进入“锁定”分支，避免依赖非原子 get -> set 的组合。
+    这里的计数器用于判断是否应当进入"锁定"分支，避免依赖非原子 get -> set 的组合。
     """
     if not username:
         return 0
@@ -144,16 +144,18 @@ def login(instance):
             if LoginSerializer._need_captcha(username, max_attempts):
                 LoginSerializer._validate_captcha(username, captcha)
 
-        # 验证用户凭据
-        user = User.objects.filter(
-            username=username,
-            password=password_encrypt(password)
-        ).first()
+        # 验证用户凭据：先按用户名查找，再用 password_verify 验证密码
+        user = User.objects.filter(username=username).first()
 
-        if not user:
+        if not user or not password_verify(password, user.password):
             LoginSerializer._handle_failed_login(username, is_license_valid, failed_attempts, lock_time)
             raise AppApiException(500, _('The username or password is incorrect'))
 
+        # Transparently upgrade legacy MD5 hash to PBKDF2
+        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!"))
 
@@ -213,7 +215,7 @@ def _handle_failed_login(username: str, is_license_valid: bool, failed_attempts:
         - 使用 record_login_fail / record_login_fail_lock 两个原子 incr 来记录失败；
         - 不再依赖精确等于 0 的比较来触发锁，而是基于原子计数 >= 阈值来决定进入锁定分支；
         - 使用 cache.add 原子创建锁键，cache.add 保证只有第一个成功创建者可写入该键；
-          其他并发到达的请求若发现计数已到达阈值也应当返回“已锁定”响应，避免出现绕过。
+          其他并发到达的请求若发现计数已到达阈值也应当返回"已锁定"响应，避免出现绕过。
         """
         # 记录普通失败计数（供验证码触发使用）
         try:

apps/users/serializers/user.py

@@ -26,9 +26,10 @@
 from common.database_model_manage.database_model_manage import DatabaseModelManage
 from common.db.search import page_search
 from common.exception.app_exception import AppApiException
-from common.utils.common import valid_license, password_encrypt, get_random_chars
+from common.utils.common import valid_license, password_encrypt, password_verify, get_random_chars
 from common.utils.rsa_util import decrypt
 from maxkb import settings
+from maxkb.const import CONFIG
 from maxkb.conf import PROJECT_DIR
 from system_manage.models import SystemSetting, SettingType, AuthTargetType, WorkspaceUserResourcePermission
 from users.models import User
@@ -116,7 +117,7 @@ def profile(user: User, auth: Auth):
             'source': user.source,
             'role': auth.role_list,
             'permissions': auth.permission_list,
-            'is_edit_password': user.password == 'd880e722c47a34d8e9fce789fc62389d' if user.source == 'LOCAL' else False,
+            'is_edit_password': password_verify(CONFIG.get('DEFAULT_PASSWORD', 'MaxKB@123..'), user.password) if user.source == 'LOCAL' else False,
             'language': user.language,
             'workspace_list': workspace_list,
             'role_name': role_name