feat: check old password for admin #811

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

@@ -106,6 +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)  # 只有admin用户重置密码时才需要传递该字段
 
     class Meta:
         model = Profile
@@ -115,6 +116,9 @@ class Meta:
     def validate_password(self, password):
         return get_raw_password(self.instance.category_id, password)
 
+    def validate_old_password(self, old_password):
+        return get_raw_password(self.instance.category_id, old_password)
+
 
 class ProfileCreateInputSLZ(serializers.ModelSerializer):
     category_id = serializers.IntegerField(required=False)

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

@@ -40,7 +40,13 @@
 from bkuser_core.profiles.exceptions import CountryISOCodeNotMatch
 from bkuser_core.profiles.models import DynamicFieldInfo, Profile
 from bkuser_core.profiles.signals import post_profile_create, post_profile_update
-from bkuser_core.profiles.utils import align_country_iso_code, make_password_by_config, parse_username_domain
+from bkuser_core.profiles.utils import (
+    align_country_iso_code,
+    check_old_password,
+    make_password_by_config,
+    parse_username_domain,
+    should_check_old_password,
+)
 from bkuser_core.user_settings.constants import SettingsEnableNamespaces
 from bkuser_core.user_settings.models import Setting, SettingMeta
 
@@ -114,7 +120,6 @@ def _update(self, request, partial):
         slz = ProfileUpdateInputSLZ(instance, data=request.data, partial=partial)
         slz.is_valid(raise_exception=True)
         operate_type = OperationType.UPDATE.value
-
         validated_data = slz.validated_data
 
         # 前端是把extras字段打平提交的
@@ -158,6 +163,10 @@ def _update(self, request, partial):
         update_summary = {"request": request}
         # 密码修改加密
         if validated_data.get("password"):
+            # 如果重置的是admin账号的密码，需要对原始密码进行校验
+            if should_check_old_password(username=instance.username):
+                check_old_password(instance=instance, old_password=validated_data["old_password"], request=request)
+
             operate_type = (
                 OperationType.FORGET_PASSWORD.value
                 if request.headers.get("User-From-Token")

src/api/bkuser_core/audit/constants.py

@@ -35,6 +35,12 @@ class LogInFailReason(AutoLowerEnum):
     )
 
 
+class ResetPasswordFailReason(AutoLowerEnum):
+    BAD_OLD_PASSWORD = auto()
+
+    _choices_labels = ((BAD_OLD_PASSWORD, "原密码校验错误"),)
+
+
 class OperationType(AutoLowerEnum):
     CREATE = auto()
     UPDATE = auto()

src/api/bkuser_core/audit/handlers.py

@@ -42,6 +42,17 @@ def create_reset_password_log(sender, instance: "Profile", operator: str, extra_
         except Exception:  # pylint: disable=broad-except
             logger.exception("failed to create reset password log")
 
+    if "failed_reason" in extra_values:
+        try:
+            create_profile_log(
+                instance,
+                "ResetPassword",
+                {"is_success": False, "reason": extra_values["failed_reason"]},
+                extra_values["request"],
+            )
+        except Exception:  # pylint: disable=broad-except
+            logger.exception("failed to create reset password log")
+
 
 @receiver([post_profile_create, post_department_create, post_category_create, post_field_create, post_setting_create])
 def create_audit_log(sender, instance: "Profile", operator: str, extra_values: dict, **kwargs):

src/api/bkuser_core/audit/managers.py

@@ -16,12 +16,28 @@
 from django.db import models
 from django.utils.timezone import now
 
-from .constants import LogInFailReason
+from .constants import LogInFailReason, ResetPasswordFailReason
 
 
 class ResetPasswordManager(models.Manager):
     """重置密码DB管理器"""
 
+    def latest_check_old_password_failed_count(self):
+        """最近一段时间重置密码失败的次数，其中最近一段时间指从上一次成功重置密码后到现在"""
+        # 查找最近一次成功的时间
+        try:
+            latest_time = self.filter(is_success=True).latest().create_time
+        except ObjectDoesNotExist:
+            # 当没有任何成功记录时，防止存在大量失败记录时进行统计导致可能的慢查询，只计算默认配置的统计时间
+            # 这里取配置里默认设置的统计时间
+            latest_time = now() - datetime.timedelta(seconds=settings.RESET_PASSWORD_RECORD_COUNT_SECONDS)
+
+        return self.filter(
+            is_success=False,
+            reason=ResetPasswordFailReason.BAD_OLD_PASSWORD.value,
+            create_time__gte=latest_time,
+        ).count()
+
 
 class LogInManager(models.Manager):
     def latest_failed_count(self) -> int:

src/api/bkuser_core/audit/migrations/0007_auto_20221212_1131.py

@@ -0,0 +1,22 @@
+# Generated by Django 3.2.13 on 2022-12-12 03:31
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('audit', '0006_alter_login_index_together'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='resetpassword',
+            options={'get_latest_by': 'create_time', 'ordering': ['-create_time']},
+        ),
+        migrations.AddField(
+            model_name='resetpassword',
+            name='reason',
+            field=models.CharField(blank=True, choices=[('bad_old_password', '原密码校验错误')], max_length=32, null=True, verbose_name='重置密码失败原因'),
+        ),
+    ]

src/api/bkuser_core/audit/models.py

@@ -15,7 +15,7 @@
 from django.db import models
 from jsonfield import JSONField
 
-from bkuser_core.audit.constants import LogInFailReason, OperationStatus
+from bkuser_core.audit.constants import LogInFailReason, OperationStatus, ResetPasswordFailReason
 from bkuser_core.audit.managers import LogInManager, ResetPasswordManager
 from bkuser_core.common.fields import EncryptField
 from bkuser_core.common.models import TimestampedModel
@@ -102,6 +102,13 @@ class ResetPassword(ProfileRelatedLog):
     token = models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, null=True)
     is_success = models.BooleanField("是否重置成功", default=False)
     password = EncryptField(default="")
+    reason = models.CharField(
+        "重置密码失败原因",
+        max_length=32,
+        choices=ResetPasswordFailReason.get_choices(),
+        null=True,
+        blank=True,
+    )
 
     objects = ResetPasswordManager()
 
@@ -110,3 +117,4 @@ def __str__(self):
 
     class Meta:
         ordering = ["-create_time"]
+        get_latest_by = "create_time"

src/api/bkuser_core/common/error_codes.py

@@ -124,6 +124,7 @@ def __getattr__(self, code_name):
         ErrorCode("USER_ALREADY_EXISTED", _("该目录下此用户名已存在"), status_code=HTTP_409_CONFLICT),
         ErrorCode("SAVE_USER_INFO_FAILED", _("保存用户信息失败")),
         ErrorCode("PASSWORD_DUPLICATED", _("新密码不能与最近{max_password_history}次密码相同")),
+        ErrorCode("OLD_PASSWORD_ERROR", _("原密码校验失败")),
         # 上传文件相关
         ErrorCode("FILE_IMPORT_TOO_LARGE", _("上传文件过大")),
         ErrorCode("FILE_IMPORT_FORMAT_ERROR", _("上传文件格式错误")),

src/api/bkuser_core/config/common/system.py

@@ -17,6 +17,11 @@
 # 密码配置
 # ==============================================================================
 
+# 允许原始密码校验错误次数
+RESET_PASSWORD_OLD_PASSWORD_ERROR_MAX_COUNT = 3
+# 重置密码时对原始密码校验超限是否锁定
+ENABLE_RESET_PASSWORD_ERROR_PROFILE_LOCK = env.bool("ENABLE_RESET_PASSWORD_ERROR_PROFILE_LOCK", default=False)
+
 # 最大密码长度（明文）
 PASSWORD_MAX_LENGTH = 32
 # 重复密码最大历史数量
@@ -87,6 +92,9 @@
 # 登录次数统计时间周期, 默认为一个月
 LOGIN_RECORD_COUNT_SECONDS = env.int("LOGIN_RECORD_COUNT_SECONDS", default=60 * 60 * 24 * 30)
 
+# 重置密码次数统计时间周期, 默认为十分钟
+RESET_PASSWORD_RECORD_COUNT_SECONDS = env.int("RESET_PASSWORD_RECORD_COUNT_SECONDS", default=60 * 10)
+
 # sync, 用户管理本身做业务 HTTP API 数据源, 可以被另一个用户管理同步过去
 # 复用 API, 接口参数中存在 SYNC_API_PARAM 时, 以sync的接口协议返回
 SYNC_API_PARAM = "for_sync"

src/api/bkuser_core/profiles/models.py

@@ -180,6 +180,10 @@ def to_audit_info(self):
     def bad_check_cnt(self) -> int:
         return self.login_set.latest_failed_count()
 
+    @property
+    def bad_old_password_check_cnt(self):
+        return self.resetpassword_set.latest_check_old_password_failed_count()
+
     @property
     def latest_check_time(self):
         return self.login_set.filter(is_success=False, reason=LogInFailReason.BAD_PASSWORD.value).latest().create_time

src/api/bkuser_core/profiles/utils.py

@@ -16,22 +16,27 @@
 from typing import TYPE_CHECKING, Dict, Tuple
 
 from django.conf import settings
-from django.contrib.auth.hashers import make_password
+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
 from bkuser_core.user_settings.constants import InitPasswordMethod
 from bkuser_core.user_settings.loader import ConfigProvider
 from bkuser_global.local import local
 
-if TYPE_CHECKING:
-    from bkuser_core.profiles.models import Profile
-
 logger = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    from rest_framework.request import Request
+
 
 def gen_password(length):
     # 必须包含至少一个数字
@@ -245,3 +250,53 @@ def remove_sensitive_fields_for_profile(request, data: Dict) -> Dict:
                 extras.pop(key)
 
     return data
+
+
+def check_old_password(instance: "Profile", old_password: str, request: "Request"):
+    """原密码校验"""
+    raw_profile = Profile.objects.get(id=instance.id)
+
+    if not check_password(old_password, raw_profile.password):
+        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": ResetPasswordFailReason.get_choice_label(failed_reason.value)},
+        )
+
+        if (
+            instance.bad_old_password_check_cnt >= settings.RESET_PASSWORD_OLD_PASSWORD_ERROR_MAX_COUNT
+            and settings.ENABLE_RESET_PASSWORD_ERROR_PROFILE_LOCK
+        ):
+            # 校验失败次数超过配置次数并且配置锁定则对用户进行锁定
+            raw_profile.status = ProfileStatus.LOCKED.value
+            raw_profile.save()
+            create_general_log(
+                operator=request.operator,
+                operate_type=OperationType.UPDATE.value,
+                operator_obj=instance,
+                request=request,
+            )
+
+        raise error_codes.OLD_PASSWORD_ERROR
+
+
+def should_check_old_password(username: str) -> bool:
+    """重置密码时，校验是否为需要检查旧密码的用户"""
+    formatted_username = username.replace(" ", "").lower()
+    if not formatted_username == "admin":
+        return False
+    return True