# coding: utf-8 import logging from datetime import timedelta from django import forms from django.contrib.auth import authenticate from django.utils.translation import gettext_lazy as _ from django.utils import timezone from django.core.exceptions import ValidationError from pingcap_sso.models import User, LoginLog, SecurityCode from pingcap_sso.utils.requests import get_client_ip, get_safe_redirect_uri from pingcap_sso.utils.phone import is_virtual_phone from pingcap_sso.lib.captcha import verify_captcha, enable_captcha from pingcap_sso.exceptions import CaptchaRequiredError from .phone import PhoneField, PhoneForm logger = logging.getLogger(__name__) class AuthenticateFormMixin: error_messages = { 'password_required': _('password is required'), 'sms_code_required': _('sms code is required'), 'invalid_login': _('login failed'), 'invalid_sms_code': _('invalid sms code'), 'inactive': _('this user has been disabled.'), 'invalid_captcha': _('invalid captcha'), 'unknown_error': _('unknown error'), 'user_not_exist': _('user does not exist'), 'too_many_failed_login': _('too many failed login attempts, please try again later'), } def __init__(self, request=None, *args, **kwargs): self.request = request self.auth_user = None # authenticated self.temp_user = None # unauthenticated super(AuthenticateFormMixin, self).__init__(*args, **kwargs) def is_valid(self): is_valid = super(AuthenticateFormMixin, self).is_valid() if is_valid: self.auth_user.create_login_log(self.request, LoginLog.LOGIN_SUCCESS) elif self.temp_user: self.temp_user.create_login_log(self.request, LoginLog.LOGIN_FAILED) return is_valid def get_user_id(self): if self.auth_user: return self.auth_user.id return None def get_user(self): return self.auth_user def _authenticate_user(self, temp_user, request, identifier, identifier_field_name, password, sms_code, re_token_v3): """ Controls whether the given User may log in. """ # if not pingcap_captcha.verify( # temp_user, # get_client_ip(request), # re_token_v3 # ): # self.add_error('re_token_v3', self.error_messages['invalid_captcha']) # return if LoginLog.objects.filter( user=temp_user, login_time__gte=timezone.now() - timedelta(hours=1), login_status=LoginLog.LOGIN_FAILED ).count() > 5: self.add_error('', self.error_messages['too_many_failed_login']) return # Check password or sms_code # func:authenticate will return `None` if auth failed. # This step is very necessary, `auth_user` is different from `_user`. auth_user = authenticate( request, identifier=identifier, user=temp_user, password=password, sms_code=sms_code) if not auth_user: if password: self.add_error('password', self.error_messages['invalid_login']) if sms_code: self.add_error('code', self.error_messages['invalid_sms_code']) return if not auth_user.is_active: self.add_error(identifier_field_name, self.error_messages['inactive']) return auth_user class PhoneLoginForm(AuthenticateFormMixin, forms.Form): phone = PhoneField(required=True) code = forms.CharField(required=True, strip=True) re_token_v3 = forms.CharField(required=False, strip=True) redirect_to = forms.URLField(required=False) def clean_redirect_to(self): redirect_to = self.cleaned_data['redirect_to'] return get_safe_redirect_uri(redirect_to, self.request) def clean(self): phone = self.cleaned_data.get('phone') sms_code = self.cleaned_data.get('code') re_token_v3 = self.cleaned_data.get('re_token_v3') _temp_password = None if phone and sms_code: try: self.temp_user = User.objects.get(phone=phone) except User.DoesNotExist: self.add_error('phone', self.error_messages['user_not_exist']) return self.auth_user = self._authenticate_user( temp_user=self.temp_user, request=self.request, identifier=phone, identifier_field_name='phone', password=_temp_password, sms_code=sms_code, re_token_v3=re_token_v3 ) return self.cleaned_data class PasswordLoginForm(AuthenticateFormMixin, forms.Form): identifier = forms.CharField( widget=forms.TextInput(attrs={ 'type': 'email', 'id': 'inputUserEmail', 'class': 'form-control', 'placeholder': 'email or phone', 'required': True, 'autofocus': True, 'autocomplete': 'username' }), error_messages={ 'required': _('identifier is required') } ) password = forms.CharField( required=True, strip=False, widget=forms.PasswordInput(attrs={ 'type': 'password', 'id': 'inputPassword', 'class': 'form-control', 'placeholder': 'password', 'required': True, 'autocomplete': 'current-password' }), error_messages={ 'required': _('password is required') } ) re_token_v3 = forms.CharField(required=False, strip=True) redirect_to = forms.URLField(required=False) def clean_redirect_to(self): redirect_to = self.cleaned_data['redirect_to'] return get_safe_redirect_uri(redirect_to, self.request) def clean(self): identifier = self.cleaned_data.get('identifier') password = self.cleaned_data.get('password') re_token_v3 = self.cleaned_data.get('re_token_v3') if identifier and password: try: self.temp_user = User.objects.find_user(identifier) except User.DoesNotExist: self.add_error('identifier', self.error_messages['invalid_login']) return self.auth_user = self._authenticate_user( temp_user=self.temp_user, request=self.request, identifier=identifier, identifier_field_name='identifier', password=password, sms_code=None, re_token_v3=re_token_v3 ) return self.cleaned_data class PhoneCheckForm(PhoneForm): def clean_phone(self): phone = self.cleaned_data.get('phone') if phone: try: self.user = User.objects.get(phone=phone) except User.DoesNotExist: raise ValidationError(_('this phone number has not been registered')) if not self.user.is_active: raise ValidationError('该用户已被禁止登录') return phone class PhoneSendCodeForm(PhoneCheckForm): re_token_v3 = forms.CharField(required=False, strip=True) def clean(self): if self._errors: return return self.cleaned_data class UpdatePhoneSendCodeForm(forms.Form): phone = PhoneField(required=True) captcha_key = forms.CharField(required=False) captcha_response = forms.CharField(required=False) def __init__(self, request, *args, **kwargs): self.request = request super(UpdatePhoneSendCodeForm, self).__init__(*args, **kwargs) def clean_phone(self): phone = self.cleaned_data.get('phone') if phone and is_virtual_phone(phone): raise ValidationError(_('currently does not support virtual carrier numbers')) return phone def clean(self): if self._errors: return phone = self.cleaned_data.get('phone') captcha_key = self.cleaned_data.get('captcha_key') captcha_response = self.cleaned_data.get('captcha_response') if enable_captcha(phone): if not (captcha_key and captcha_response): raise CaptchaRequiredError(_('captcha key and response are required')) if not verify_captcha(captcha_key, captcha_response): self.add_error('captcha_response', _('invalid captcha response')) return return self.cleaned_data def save(self): phone = self.cleaned_data.get('phone') SecurityCode.send_code( phone, channel=SecurityCode.DELIVER_BY_PHONE, ip=get_client_ip(self.request), scene=SecurityCode.SCENE_UPDATE_PHONE, )