From df64777a76fb9b1be3a649bc7d9996e2d86e2db2 Mon Sep 17 00:00:00 2001 From: wxg0103 <727495428@qq.com> Date: Mon, 26 Oct 2026 15:36:37 +0800 Subject: [PATCH] feat: implement caching for default password check in user profile --- apps/users/serializers/user.py | 15 +++++++- apps/users/tests.py | 3 -- apps/users/tests/__init__.py | 1 + apps/users/tests/test_profile.py | 59 ++++++++++++++++++++++++++++++++ 4 files changed, 74 insertions(+), 4 deletions(-) delete mode 100644 apps/users/tests.py create mode 100644 apps/users/tests/__init__.py create mode 100644 apps/users/tests/test_profile.py diff --git a/apps/users/serializers/user.py b/apps/users/serializers/user.py index 6895daef90a..58b47f58506 100644 --- a/apps/users/serializers/user.py +++ b/apps/users/serializers/user.py @@ -8,6 +8,7 @@ """ import datetime +import hashlib import json import os import re @@ -61,6 +62,7 @@ VERIFY_CODE_EXPIRE_SECONDS = 10 * 60 # 达到错误上限后的锁定冷却时长 VERIFY_CODE_LOCKOUT_SECONDS = 10 * 60 +DEFAULT_PASSWORD_CHECK_CACHE_TIMEOUT = 7 * 24 * 60 * 60 def _raise_verify_code_limit(): @@ -228,6 +230,17 @@ def get_workspace_list_by_user(user_id): return [{"id": "default", "name": "default"}] +def _is_default_password(default_password: str, hashed_password: str): + cache_version = Cache_Version.USER.get_version() + digest = hashlib.sha256(f"{default_password}\0{hashed_password}".encode()).hexdigest() + cache_key = Cache_Version.USER.get_key(user_id=f"default-password:{digest}") + is_default = cache.get(cache_key, version=cache_version) + if is_default is None: + is_default = password_verify(default_password, hashed_password) + cache.set(cache_key, is_default, timeout=DEFAULT_PASSWORD_CHECK_CACHE_TIMEOUT, version=cache_version) + return is_default + + class UserProfileSerializer(serializers.Serializer): @staticmethod def profile(user: User, auth: Auth): @@ -261,7 +274,7 @@ def profile(user: User, auth: Auth): "source": user.source, "role": list(auth.roles), "permissions": auth.permissions, - "is_edit_password": password_verify(CONFIG.get("DEFAULT_PASSWORD", "MaxKB@123.."), user.password) + "is_edit_password": _is_default_password(CONFIG.get("DEFAULT_PASSWORD", "MaxKB@123.."), user.password) if user.source == "LOCAL" else False, "language": user.language, diff --git a/apps/users/tests.py b/apps/users/tests.py deleted file mode 100644 index 7ce503c2dd9..00000000000 --- a/apps/users/tests.py +++ /dev/null @@ -1,3 +0,0 @@ -from django.test import TestCase - -# Create your tests here. diff --git a/apps/users/tests/__init__.py b/apps/users/tests/__init__.py new file mode 100644 index 00000000000..8b137891791 --- /dev/null +++ b/apps/users/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/apps/users/tests/test_profile.py b/apps/users/tests/test_profile.py new file mode 100644 index 00000000000..011220db30d --- /dev/null +++ b/apps/users/tests/test_profile.py @@ -0,0 +1,59 @@ +from unittest.mock import patch + +import uuid_utils.compat as uuid +from django.core.cache import cache +from django.test import SimpleTestCase, override_settings + +from common.auth.struct.auth import Auth +from users.models import User +from users.serializers.user import UserProfileSerializer + + +@override_settings( + CACHES={ + "default": { + "BACKEND": "django.core.cache.backends.locmem.LocMemCache", + "LOCATION": "user-profile-tests", + } + } +) +class UserProfilePasswordCacheTests(SimpleTestCase): + def setUp(self): + cache.clear() + self.user = User( + id=uuid.uuid4(), + username="profile-user", + nick_name="Profile User", + email="profile@example.com", + password="hash-v1", + role="USER", + source="LOCAL", + language="en-US", + ) + self.auth = Auth(roles=set(), permissions={}) + + def _profile(self): + with ( + patch("users.serializers.user.DatabaseModelManage.get_model", return_value=None), + patch("users.serializers.user.get_workspace_list_by_user", return_value=[]), + ): + return UserProfileSerializer().profile(self.user, self.auth) + + def test_profile_caches_default_password_check_for_same_hash(self): + with patch("users.serializers.user.password_verify", return_value=True) as verify_password: + first = self._profile() + second = self._profile() + + self.assertTrue(first["is_edit_password"]) + self.assertTrue(second["is_edit_password"]) + verify_password.assert_called_once() + + def test_profile_rechecks_when_password_hash_changes(self): + with patch("users.serializers.user.password_verify", side_effect=[True, False]) as verify_password: + first = self._profile() + self.user.password = "hash-v2" + second = self._profile() + + self.assertTrue(first["is_edit_password"]) + self.assertFalse(second["is_edit_password"]) + self.assertEqual(verify_password.call_count, 2)