| 80 | |
| 81 | |
| 82 | class LoginSerializer(serializers.Serializer): |
| 83 | @staticmethod |
| 84 | def get_auth_setting(): |
| 85 | """获取认证设置""" |
| 86 | auth_setting_model = DatabaseModelManage.get_model("auth_setting") |
| 87 | auth_setting = {} |
| 88 | if auth_setting_model: |
| 89 | setting_obj = auth_setting_model.objects.filter(param_key="auth_setting").first() |
| 90 | if setting_obj: |
| 91 | try: |
| 92 | auth_setting = json.loads(setting_obj.param_value) or {} |
| 93 | except Exception: |
| 94 | auth_setting = {} |
| 95 | return auth_setting |
| 96 | |
| 97 | @staticmethod |
| 98 | def login(instance): |
| 99 | # 解密数据 |
| 100 | username = instance.get("username", "") |
| 101 | encrypted_data = instance.get("encryptedData", "") |
| 102 | |
| 103 | if encrypted_data: |
| 104 | try: |
| 105 | decrypted_raw = decrypt(encrypted_data) |
| 106 | # decrypt 可能返回非 JSON 字符串，防护解析异常 |
| 107 | decrypted_data = json.loads(decrypted_raw) if decrypted_raw else {} |
| 108 | if isinstance(decrypted_data, dict): |
| 109 | instance.update(decrypted_data) |
| 110 | except Exception as e: |
| 111 | maxkb_logger.exception("Failed to decrypt/parse encryptedData for user %s: %s", username, e) |
| 112 | raise AppApiException(500, _("Invalid encrypted data")) |
| 113 | |
| 114 | try: |
| 115 | LoginRequest(data=instance).is_valid(raise_exception=True) |
| 116 | except serializers.ValidationError: |
| 117 | raise |
| 118 | except Exception as e: |
| 119 | raise AppApiException(500, str(e)) |
| 120 | |
| 121 | password = instance.get("password") |
| 122 | captcha = instance.get("captcha", "") |
| 123 | |
| 124 | # 获取认证配置 |
| 125 | auth_setting = LoginSerializer.get_auth_setting() |
| 126 | max_attempts = auth_setting.get("max_attempts", 1) |
| 127 | failed_attempts = auth_setting.get("failed_attempts", 5) |
| 128 | lock_time = auth_setting.get("lock_time", 10) |
| 129 | |
| 130 | # 检查许可证有效性 |
| 131 | license_validator = DatabaseModelManage.get_model("license_is_valid") or (lambda: False) |
| 132 | is_license_valid = license_validator() if license_validator() is not None else False |
| 133 | |
| 134 | if is_license_valid: |
| 135 | # 检查账户是否被锁定 |
| 136 | if LoginSerializer._is_account_locked(username, failed_attempts): |
| 137 | raise AppApiException( |
| 138 | 1005, _("This account has been locked for %s minutes, please try again later") % lock_time |
| 139 | ) |