feat: check old password for admin-review修改 #811

src/api/bkuser_core/api/web/profile/serializers.py

@@ -106,7 +106,7 @@ class ProfileUpdateInputSLZ(serializers.ModelSerializer):
     leader = serializers.ListField(child=serializers.IntegerField(), required=False)
     departments = serializers.ListField(child=serializers.IntegerField(), required=False)
     password = serializers.CharField(required=False, write_only=True)
-    old_password = serializers.CharField(required=False, write_only=True)
+    old_password = serializers.CharField(required=False, write_only=True)  # 只有admin用户重置密码时才需要传递该字段
 
     class Meta:
         model = Profile
@@ -116,7 +116,7 @@ class Meta:
     def validate_password(self, password):
         return get_raw_password(self.instance.category_id, password)
 
-    def validated_old_password(self, old_password):
+    def validate_old_password(self, old_password):
         return get_raw_password(self.instance.category_id, old_password)
 
 

src/api/bkuser_core/api/web/profile/views.py

@@ -31,7 +31,7 @@
 )
 from bkuser_core.api.web.utils import get_category, get_operator, validate_password
 from bkuser_core.api.web.viewset import CustomPagination
-from bkuser_core.audit.constants import OperationStatus, OperationType, ResetPasswordFailReason
+from bkuser_core.audit.constants import OperationType
 from bkuser_core.audit.utils import audit_general_log, create_general_log
 from bkuser_core.bkiam.permissions import IAMAction, ManageDepartmentProfilePermission, Permission
 from bkuser_core.categories.models import ProfileCategory
@@ -164,28 +164,7 @@ def _update(self, request, partial):
         if validated_data.get("password"):
             # 如果重置的是admin账号的密码，需要对原始密码进行校验
             if instance.username == "admin":
-                old_password_check_result = check_old_password(
-                    instance=instance, old_password=validated_data["old_password"]
-                )
-
-                if not old_password_check_result:
-                    failed_reason = ResetPasswordFailReason.BAD_OLD_PASSWORD
-                    update_summary.update({"failed_reason": failed_reason.value})
-                    post_profile_update.send(
-                        sender=self,
-                        instance=instance,
-                        operator=request.operator,
-                        extra_values=update_summary,
-                    )
-                    create_general_log(
-                        operator=request.operator,
-                        operate_type=OperationType.ADMIN_RESET_PASSWORD.value,
-                        operator_obj=instance,
-                        request=request,
-                        status=OperationStatus.FAILED.value,
-                        extra_info={"failed_info": failed_reason.get_choices()},
-                    )
-                    raise error_codes.OLD_PASSWORD_ERROR
+                check_old_password(instance=instance, old_password=validated_data["old_password"], request=request)
 
             operate_type = (
                 OperationType.FORGET_PASSWORD.value

src/api/bkuser_core/audit/managers.py

@@ -26,11 +26,10 @@ def latest_check_old_password_failed_count(self):
         """获取上一次成功重置密码前最近重置密码失败的次数"""
         farthest_count_time = now() - datetime.timedelta(seconds=settings.RESET_PASSWORD_RECORD_COUNT_SECONDS)
         try:
-            latest_success_time = (
-                self.filter(is_success=True, create_time__gte=farthest_count_time).latest().create_time
-            )
+            latest_success_time = self.filter(is_success=True).latest().create_time
         except ObjectDoesNotExist:
-            # 当没有任何成功记录时，直接统计时间区域内的错误次数
+            # 当没有任何成功记录时，直接统计时间区域内的错误次数，防止存在大量失败记录时进行统计导致可能的慢查询，
+            # 这里配置统计时间farthest_count_time
             return self.filter(
                 is_success=False,
                 reason=ResetPasswordFailReason.BAD_OLD_PASSWORD.value,

src/api/bkuser_core/profiles/models.py

@@ -181,7 +181,7 @@ def bad_check_cnt(self) -> int:
         return self.login_set.latest_failed_count()
 
     @property
-    def bad_old_pwd_check_cnt(self):
+    def bad_old_password_check_cnt(self):
         return self.resetpassword_set.latest_check_old_password_failed_count()
 
     @property

src/api/bkuser_core/profiles/utils.py

@@ -13,15 +13,18 @@
 import re
 import string
 import urllib.parse
-from typing import Dict, Tuple
+from typing import TYPE_CHECKING, Dict, Tuple
 
 from django.conf import settings
 from django.contrib.auth.hashers import check_password, make_password
 from phonenumbers.phonenumberutil import UNKNOWN_REGION, country_code_for_region, region_code_for_country_code
 
+from ..audit.constants import OperationStatus, OperationType, ResetPasswordFailReason
 from ..audit.models import ResetPassword
 from .exceptions import CountryISOCodeNotMatch, UsernameWithDomainFormatError
+from bkuser_core.audit.utils import create_general_log, create_profile_log
 from bkuser_core.categories.cache import get_default_category_id_from_local_cache
+from bkuser_core.common.error_codes import error_codes
 from bkuser_core.profiles.constants import ProfileStatus
 from bkuser_core.profiles.models import Profile
 from bkuser_core.profiles.validators import DOMAIN_PART_REGEX, USERNAME_REGEX
@@ -31,6 +34,9 @@
 
 logger = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    from rest_framework.request import Request
+
 
 def gen_password(length):
     # 必须包含至少一个数字
@@ -246,13 +252,42 @@ def remove_sensitive_fields_for_profile(request, data: Dict) -> Dict:
     return data
 
 
-def check_old_password(instance: "Profile", old_password: str) -> bool:
-    """原密码校验, 校验失败次数超过配置次数会对用户进行锁定"""
+def check_old_password(instance: "Profile", old_password: str, request: "Request") -> bool:
+    """原密码校验"""
     raw_profile = Profile.objects.get(id=instance.id)
 
     if not check_password(old_password, raw_profile.password):
-        if instance.bad_old_pwd_check_cnt >= settings.ALLOW_OLD_PASSWORD_ERROR_TIME:
+        failed_reason = ResetPasswordFailReason.BAD_OLD_PASSWORD
+        try:
+            create_profile_log(
+                instance,
+                "ResetPassword",
+                {"is_success": False, "reason": failed_reason.value},
+                request,
+            )
+        except Exception:  # pylint: disable=broad-except
+            logger.exception("failed to create reset password log")
+
+        create_general_log(
+            operator=request.operator,
+            operate_type=OperationType.ADMIN_RESET_PASSWORD.value,
+            operator_obj=instance,
+            request=request,
+            status=OperationStatus.FAILED.value,
+            extra_info={"failed_info": failed_reason.get_choices()},
+        )
+
+        if instance.bad_old_password_check_cnt >= settings.ALLOW_OLD_PASSWORD_ERROR_TIME:
+            # 校验失败次数超过配置次数会对用户进行锁定
             raw_profile.status = ProfileStatus.LOCKED.value
             raw_profile.save()
-        return False
+            create_general_log(
+                operator=request.operator,
+                operate_type=OperationType.UPDATE.value,
+                operator_obj=instance,
+                request=request,
+            )
+
+        raise error_codes.OLD_PASSWORD_ERROR
+
     return True