Function _login

web/pgadmin/authenticate/__init__.py:103–203

Internal authentication process locked by a mutex.

()

def _login():
    """
    Internal authentication process locked by a mutex.
    """
    form = _security.forms.get('login_form').cls(request.form)
    if OAUTH2 in config.AUTHENTICATION_SOURCES \
            and 'oauth2_button' in request.form:
        # Sending empty form as oauth2 does not require form attribute
        auth_obj = AuthSourceManager({}, copy.deepcopy(
            config.AUTHENTICATION_SOURCES))
        # Persist only the OAuth2 provider selection across the redirect.
        # The auth-source instance lives on current_app's registry so we
        # re-look-up rather than persist a live class instance in session.
        session['oauth2_current_client'] = request.form.get('oauth2_button')
    else:
        auth_obj = AuthSourceManager(form, copy.deepcopy(
            config.AUTHENTICATION_SOURCES))

    session['auth_source_manager'] = None

    username = form.data['email']
    user = User.query.filter_by(username=username,
                                auth_source=INTERNAL).first()

    if user:
        if user.login_attempts >= config.MAX_LOGIN_ATTEMPTS > 0:
            user.locked = True
        else:
            user.locked = False
        db.session.commit()

        if user.login_attempts >= config.MAX_LOGIN_ATTEMPTS > 0:
            flash(gettext('Your account is locked. Please contact the '
                          'Administrator.'),
                  MessageType.WARNING)
            logout_user()
            return redirect(pga_utils.get_safe_post_logout_redirect())

    # Validate the user
    if not auth_obj.validate():
        for field in form.errors:
            flash_login_attempt_error = None
            if user and field in config.LOGIN_ATTEMPT_FIELDS:
                if config.MAX_LOGIN_ATTEMPTS > 0:
                    user.login_attempts += 1
                    left_attempts = \
                        config.MAX_LOGIN_ATTEMPTS - user.login_attempts
                    if left_attempts > 1:
                        flash_login_attempt_error = \
                            gettext('{0} more attempts remaining.'.
                                    format(left_attempts))
                    else:
                        flash_login_attempt_error = \
                            gettext('{0} more attempt remaining.'.
                                    format(left_attempts))
                db.session.commit()
            for error in form.errors[field]:
                if flash_login_attempt_error: