def clean_username(self):
        # type: () -> str
        email = self.cleaned_data['username']
        try:
            user_profile = get_user_profile_by_email(email)
        except UserProfile.DoesNotExist:
            return email

        if user_profile.realm.deactivated:
            error_msg = u"""Sorry for the trouble, but %s has been deactivated.

Please contact %s to reactivate this group.""" % (
                user_profile.realm.name,
                FromAddress.SUPPORT)
            raise ValidationError(mark_safe(error_msg))

        if not user_profile.is_active and not user_profile.is_mirror_dummy:
            error_msg = (
                u"Sorry for the trouble, but your account has been deactivated. "
                u"Please contact your organization administrator to reactivate it. "
                u"If you're not sure who that is, try contacting %s.") % (FromAddress.SUPPORT,)
            raise ValidationError(mark_safe(error_msg))

        if not user_matches_subdomain(get_subdomain(self.request), user_profile):
            logging.warning("User %s attempted to password login to wrong subdomain %s" %
                            (user_profile.email, get_subdomain(self.request)))
            raise ValidationError(mark_safe(WRONG_SUBDOMAIN_ERROR))
        return email

 def test(expected: str, host: str, *, plusport: bool=True,
          external_host: str='example.org',
          realm_hosts: Dict[str, str]={},
          root_aliases: List[str]=[]) -> None:
     with self.settings(EXTERNAL_HOST=external_host,
                        REALM_HOSTS=realm_hosts,
                        ROOT_SUBDOMAIN_ALIASES=root_aliases):
         self.assertEqual(get_subdomain(request_mock(host)), expected)
         if plusport and ':' not in host:
             self.assertEqual(get_subdomain(request_mock(host + ':443')),
                              expected)

def validate_api_key(request: HttpRequest, role: Optional[str],
                     api_key: str, is_webhook: bool=False,
                     client_name: Optional[str]=None) -> Union[UserProfile, RemoteZulipServer]:
    # Remove whitespace to protect users from trivial errors.
    api_key = api_key.strip()
    if role is not None:
        role = role.strip()

    if settings.ZILENCER_ENABLED and role is not None and is_remote_server(role):
        try:
            remote_server = get_remote_server_by_uuid(role)
        except RemoteZulipServer.DoesNotExist:
            raise InvalidZulipServerError(role)
        if api_key != remote_server.api_key:
            raise InvalidZulipServerKeyError(role)

        if get_subdomain(request) != Realm.SUBDOMAIN_FOR_ROOT_DOMAIN:
            raise JsonableError(_("Invalid subdomain for push notifications bouncer"))
        request.user = remote_server
        request._email = "zulip-server:" + role
        remote_server.rate_limits = ""
        process_client(request, remote_server, remote_server_request=True)
        return remote_server

    user_profile = access_user_by_api_key(request, api_key, email=role)
    if user_profile.is_incoming_webhook and not is_webhook:
        raise JsonableError(_("This API is not available to incoming webhook bots."))

    request.user = user_profile
    request._email = user_profile.email
    process_client(request, user_profile, client_name=client_name)

    return user_profile

def oauth_redirect_to_root(request: HttpRequest, url: str,
                           sso_type: str, is_signup: bool=False) -> HttpResponse:
    main_site_uri = settings.ROOT_DOMAIN_URI + url
    if settings.SOCIAL_AUTH_SUBDOMAIN is not None and sso_type == 'social':
        main_site_uri = (settings.EXTERNAL_URI_SCHEME +
                         settings.SOCIAL_AUTH_SUBDOMAIN +
                         "." +
                         settings.EXTERNAL_HOST) + url

    params = {
        'subdomain': get_subdomain(request),
        'is_signup': '1' if is_signup else '0',
    }

    # mobile_flow_otp is a one-time pad provided by the app that we
    # can use to encrypt the API key when passing back to the app.
    mobile_flow_otp = request.GET.get('mobile_flow_otp')
    if mobile_flow_otp is not None:
        if not is_valid_otp(mobile_flow_otp):
            raise JsonableError(_("Invalid OTP"))
        params['mobile_flow_otp'] = mobile_flow_otp

    next = request.GET.get('next')
    if next:
        params['next'] = next

    return redirect(main_site_uri + '?' + urllib.parse.urlencode(params))

def get_auth_backends_data(request: HttpRequest) -> Dict[str, Any]:
    """Returns which authentication methods are enabled on the server"""
    subdomain = get_subdomain(request)
    try:
        realm = Realm.objects.get(string_id=subdomain)
    except Realm.DoesNotExist:
        # If not the root subdomain, this is an error
        if subdomain != Realm.SUBDOMAIN_FOR_ROOT_DOMAIN:
            raise JsonableError(_("Invalid subdomain"))
        # With the root subdomain, it's an error or not depending
        # whether ROOT_DOMAIN_LANDING_PAGE (which indicates whether
        # there are some realms without subdomains on this server)
        # is set.
        if settings.ROOT_DOMAIN_LANDING_PAGE:
            raise JsonableError(_("Subdomain required"))
        else:
            realm = None
    return {
        "password": password_auth_enabled(realm),
        "dev": dev_auth_enabled(realm),
        "email": email_auth_enabled(realm),
        "github": github_auth_enabled(realm),
        "google": google_auth_enabled(realm),
        "ldap": ldap_auth_enabled(realm),
    }

def log_into_subdomain(request: HttpRequest, token: Text) -> HttpResponse:
    try:
        data = signing.loads(token, salt=_subdomain_token_salt, max_age=15)
    except signing.SignatureExpired as e:
        logging.warning('Subdomain cookie: {}'.format(e))
        return HttpResponse(status=400)
    except signing.BadSignature:
        logging.warning('Subdomain cookie: Bad signature.')
        return HttpResponse(status=400)

    subdomain = get_subdomain(request)
    if data['subdomain'] != subdomain:
        logging.warning('Login attempt on invalid subdomain')
        return HttpResponse(status=400)

    email_address = data['email']
    full_name = data['name']
    is_signup = data['is_signup']
    if is_signup:
        # If we are signing up, user_profile should be None. In case
        # email_address already exists, user will get an error message.
        user_profile = None
        return_data = {}  # type: Dict[str, Any]
    else:
        # We can be reasonably confident that this subdomain actually
        # has a corresponding realm, since it was referenced in a
        # signed cookie.  But we probably should add some error
        # handling for the case where the realm disappeared in the
        # meantime.
        realm = get_realm(subdomain)
        user_profile, return_data = authenticate_remote_user(realm, email_address)
    invalid_subdomain = bool(return_data.get('invalid_subdomain'))
    return login_or_register_remote_user(request, email_address, user_profile,
                                         full_name, invalid_subdomain=invalid_subdomain,
                                         is_signup=is_signup)

def remote_user_sso(request: HttpRequest,
                    mobile_flow_otp: Optional[str]=REQ(default=None)) -> HttpResponse:
    try:
        remote_user = request.META["REMOTE_USER"]
    except KeyError:
        # TODO: Arguably the JsonableError values here should be
        # full-page HTML configuration errors instead.
        raise JsonableError(_("No REMOTE_USER set."))

    # Django invokes authenticate methods by matching arguments, and this
    # authentication flow will not invoke LDAP authentication because of
    # this condition of Django so no need to check if LDAP backend is
    # enabled.
    validate_login_email(remote_user_to_email(remote_user))

    # Here we support the mobile flow for REMOTE_USER_BACKEND; we
    # validate the data format and then pass it through to
    # login_or_register_remote_user if appropriate.
    if mobile_flow_otp is not None:
        if not is_valid_otp(mobile_flow_otp):
            raise JsonableError(_("Invalid OTP"))

    subdomain = get_subdomain(request)
    realm = get_realm(subdomain)
    # Since RemoteUserBackend will return None if Realm is None, we
    # don't need to check whether `get_realm` returned None.
    user_profile = authenticate(remote_user=remote_user, realm=realm)
    return login_or_register_remote_user(request, remote_user, user_profile,
                                         mobile_flow_otp=mobile_flow_otp)

def json_fetch_api_key(request, user_profile, password=REQ(default='')):
    # type: (HttpRequest, UserProfile, str) -> HttpResponse
    if password_auth_enabled(user_profile.realm):
        if not authenticate(username=user_profile.email, password=password,
                            realm_subdomain=get_subdomain(request)):
            return json_error(_("Your username or password is incorrect."))
    return json_success({"api_key": user_profile.api_key})

def add_api_uri_context(context: Dict[str, Any], request: HttpRequest) -> None:
    context.update(zulip_default_context(request))

    subdomain = get_subdomain(request)
    if (subdomain != Realm.SUBDOMAIN_FOR_ROOT_DOMAIN
            or not settings.ROOT_DOMAIN_LANDING_PAGE):
        display_subdomain = subdomain
        html_settings_links = True
    else:
        display_subdomain = 'yourZulipDomain'
        html_settings_links = False

    display_host = Realm.host_for_subdomain(display_subdomain)
    api_url_scheme_relative = display_host + "/api"
    api_url = settings.EXTERNAL_URI_SCHEME + api_url_scheme_relative

    context['external_uri_scheme'] = settings.EXTERNAL_URI_SCHEME
    context['api_url'] = api_url
    context['api_url_scheme_relative'] = api_url_scheme_relative

    context["html_settings_links"] = html_settings_links
    if html_settings_links:
        settings_html = '<a href="/#settings">Zulip settings page</a>'
        subscriptions_html = '<a target="_blank" href="/#streams">streams page</a>'
    else:
        settings_html = 'Zulip settings page'
        subscriptions_html = 'streams page'
    context['settings_html'] = settings_html
    context['subscriptions_html'] = subscriptions_html

def api_dev_fetch_api_key(request: HttpRequest, username: str=REQ()) -> HttpResponse:
    """This function allows logging in without a password on the Zulip
    mobile apps when connecting to a Zulip development environment.  It
    requires DevAuthBackend to be included in settings.AUTHENTICATION_BACKENDS.
    """
    if not dev_auth_enabled() or settings.PRODUCTION:
        return json_error(_("Dev environment not enabled."))

    # Django invokes authenticate methods by matching arguments, and this
    # authentication flow will not invoke LDAP authentication because of
    # this condition of Django so no need to check if LDAP backend is
    # enabled.
    validate_login_email(username)

    subdomain = get_subdomain(request)
    realm = get_realm(subdomain)

    return_data = {}  # type: Dict[str, bool]
    user_profile = authenticate(dev_auth_username=username,
                                realm=realm,
                                return_data=return_data)
    if return_data.get("inactive_realm"):
        return json_error(_("Your realm has been deactivated."),
                          data={"reason": "realm deactivated"}, status=403)
    if return_data.get("inactive_user"):
        return json_error(_("Your account has been disabled."),
                          data={"reason": "user disable"}, status=403)
    if user_profile is None:
        return json_error(_("This user is not registered."),
                          data={"reason": "unregistered"}, status=403)
    do_login(request, user_profile)
    return json_success({"api_key": user_profile.api_key, "email": user_profile.email})

def logged_in_and_active(request: HttpRequest) -> bool:
    if not request.user.is_authenticated:
        return False
    if not request.user.is_active:
        return False
    if request.user.realm.deactivated:
        return False
    return user_matches_subdomain(get_subdomain(request), request.user)

def api_fetch_api_key(request, username=REQ(), password=REQ()):
    # type: (HttpRequest, str, str) -> HttpResponse
    return_data = {}  # type: Dict[str, bool]
    if username == "google-oauth2-token":
        user_profile = authenticate(google_oauth2_token=password,
                                    realm_subdomain=get_subdomain(request),
                                    return_data=return_data)
    else:
        if not ldap_auth_enabled(realm=get_realm_from_request(request)):
            # In case we don't authenticate against LDAP, check for a valid
            # email. LDAP backend can authenticate against a non-email.
            validate_login_email(username)

        user_profile = authenticate(username=username,
                                    password=password,
                                    realm_subdomain=get_subdomain(request),
                                    return_data=return_data)
    if return_data.get("inactive_user"):
        return json_error(_("Your account has been disabled."),
                          data={"reason": "user disable"}, status=403)
    if return_data.get("inactive_realm"):
        return json_error(_("Your realm has been deactivated."),
                          data={"reason": "realm deactivated"}, status=403)
    if return_data.get("password_auth_disabled"):
        return json_error(_("Password auth is disabled in your team."),
                          data={"reason": "password auth disabled"}, status=403)
    if user_profile is None:
        if return_data.get("valid_attestation"):
            # We can leak that the user is unregistered iff they present a valid authentication string for the user.
            return json_error(_("This user is not registered; do so from a browser."),
                              data={"reason": "unregistered"}, status=403)
        return json_error(_("Your username or password is incorrect."),
                          data={"reason": "incorrect_creds"}, status=403)

    # Maybe sending 'user_logged_in' signal is the better approach:
    #   user_logged_in.send(sender=user_profile.__class__, request=request, user=user_profile)
    # Not doing this only because over here we don't add the user information
    # in the session. If the signal receiver assumes that we do then that
    # would cause problems.
    email_on_new_login(sender=user_profile.__class__, request=request, user=user_profile)

    # Mark this request as having a logged-in user for our server logs.
    process_client(request, user_profile)
    request._email = user_profile.email

    return json_success({"api_key": user_profile.api_key, "email": user_profile.email})

    def process_response(self, request: HttpRequest, response: HttpResponse) -> HttpResponse:
        try:
            request.get_host()
        except DisallowedHost:
            # If we get a DisallowedHost exception trying to access
            # the host, (1) the request is failed anyway and so the
            # below code will do nothing, and (2) the below will
            # trigger a recursive exception, breaking things, so we
            # just return here.
            return response

        if (not request.path.startswith("/static/") and not request.path.startswith("/api/") and
                not request.path.startswith("/json/")):
            subdomain = get_subdomain(request)
            if subdomain != Realm.SUBDOMAIN_FOR_ROOT_DOMAIN:
                realm = get_realm(subdomain)
                if (realm is None):
                    return render(request, "zerver/invalid_realm.html")
        """
        If request.session was modified, or if the configuration is to save the
        session every time, save the changes and set a session cookie.
        """
        try:
            accessed = request.session.accessed
            modified = request.session.modified
        except AttributeError:
            pass
        else:
            if accessed:
                patch_vary_headers(response, ('Cookie',))
            if modified or settings.SESSION_SAVE_EVERY_REQUEST:
                if request.session.get_expire_at_browser_close():
                    max_age = None
                    expires = None
                else:
                    max_age = request.session.get_expiry_age()
                    expires_time = time.time() + max_age
                    expires = cookie_date(expires_time)
                # Save the session data and refresh the client cookie.
                # Skip session save for 500 responses, refs #3881.
                if response.status_code != 500:
                    request.session.save()
                    host = request.get_host().split(':')[0]

                    # The subdomains feature overrides the
                    # SESSION_COOKIE_DOMAIN setting, since the setting
                    # is a fixed value and with subdomains enabled,
                    # the session cookie domain has to vary with the
                    # subdomain.
                    session_cookie_domain = host
                    response.set_cookie(settings.SESSION_COOKIE_NAME,
                                        request.session.session_key, max_age=max_age,
                                        expires=expires, domain=session_cookie_domain,
                                        path=settings.SESSION_COOKIE_PATH,
                                        secure=settings.SESSION_COOKIE_SECURE or None,
                                        httponly=settings.SESSION_COOKIE_HTTPONLY or None)
        return response

def maybe_send_to_registration(request: HttpRequest, email: str, full_name: str='',
                               is_signup: bool=False, password_required: bool=True,
                               multiuse_object_key: str='') -> HttpResponse:
    realm = get_realm(get_subdomain(request))
    from_multiuse_invite = False
    multiuse_obj = None
    streams_to_subscribe = None
    invited_as = PreregistrationUser.INVITE_AS['MEMBER']
    if multiuse_object_key:
        from_multiuse_invite = True
        multiuse_obj = Confirmation.objects.get(confirmation_key=multiuse_object_key).content_object
        realm = multiuse_obj.realm
        streams_to_subscribe = multiuse_obj.streams.all()
        invited_as = multiuse_obj.invited_as

    form = HomepageForm({'email': email}, realm=realm, from_multiuse_invite=from_multiuse_invite)
    if form.is_valid():
        # Construct a PreregistrationUser object and send the user over to
        # the confirmation view.
        prereg_user = None
        if settings.ONLY_SSO:
            try:
                prereg_user = PreregistrationUser.objects.filter(
                    email__iexact=email, realm=realm).latest("invited_at")
            except PreregistrationUser.DoesNotExist:
                prereg_user = create_preregistration_user(email, request,
                                                          password_required=password_required)
        else:
            prereg_user = create_preregistration_user(email, request,
                                                      password_required=password_required)

        if multiuse_object_key:
            request.session.modified = True
            if streams_to_subscribe is not None:
                prereg_user.streams.set(streams_to_subscribe)
            prereg_user.invited_as = invited_as
            prereg_user.save()

        confirmation_link = create_confirmation_link(prereg_user, request.get_host(),
                                                     Confirmation.USER_REGISTRATION)
        if is_signup:
            return redirect(confirmation_link)

        context = {'email': email,
                   'continue_link': confirmation_link,
                   'full_name': full_name}
        return render(request,
                      'zerver/confirm_continue_registration.html',
                      context=context)
    else:
        url = reverse('register')
        return render(request,
                      'zerver/accounts_home.html',
                      context={'form': form, 'current_url': lambda: url,
                               'from_multiuse_invite': from_multiuse_invite,
                               'multiuse_object_key': multiuse_object_key},
                      )

def create_preregistration_user(email: Text, request: HttpRequest, realm_creation: bool=False,
                                password_required: bool=True) -> HttpResponse:
    realm = None
    if not realm_creation:
        realm = get_realm(get_subdomain(request))
    return PreregistrationUser.objects.create(email=email,
                                              realm_creation=realm_creation,
                                              password_required=password_required,
                                              realm=realm)

def maybe_send_to_registration(request: HttpRequest, email: str, full_name: str='',
                               is_signup: bool=False, password_required: bool=True) -> HttpResponse:
    realm = get_realm(get_subdomain(request))
    from_multiuse_invite = False
    multiuse_obj = None
    streams_to_subscribe = None
    multiuse_object_key = request.session.get("multiuse_object_key", None)
    if multiuse_object_key is not None:
        from_multiuse_invite = True
        multiuse_obj = Confirmation.objects.get(confirmation_key=multiuse_object_key).content_object
        realm = multiuse_obj.realm
        streams_to_subscribe = multiuse_obj.streams.all()

    form = HomepageForm({'email': email}, realm=realm, from_multiuse_invite=from_multiuse_invite)
    if form.is_valid():
        # Construct a PreregistrationUser object and send the user over to
        # the confirmation view.
        prereg_user = None
        if settings.ONLY_SSO:
            try:
                prereg_user = PreregistrationUser.objects.filter(
                    email__iexact=email, realm=realm).latest("invited_at")
            except PreregistrationUser.DoesNotExist:
                prereg_user = create_preregistration_user(email, request,
                                                          password_required=password_required)
        else:
            prereg_user = create_preregistration_user(email, request,
                                                      password_required=password_required)

        if multiuse_object_key is not None:
            del request.session["multiuse_object_key"]
            request.session.modified = True
            if streams_to_subscribe is not None:
                prereg_user.streams.set(streams_to_subscribe)

        confirmation_link = "".join((
            create_confirmation_link(prereg_user, request.get_host(), Confirmation.USER_REGISTRATION),
            '?full_name=',
            # urllib does not handle Unicode, so coerece to encoded byte string
            # Explanation: http://stackoverflow.com/a/5605354/90777
            urllib.parse.quote_plus(full_name.encode('utf8'))))
        if is_signup:
            return redirect(confirmation_link)

        context = {'email': email,
                   'continue_link': confirmation_link}
        return render(request,
                      'zerver/confirm_continue_registration.html',
                      context=context)
    else:
        url = reverse('register')
        return render(request,
                      'zerver/accounts_home.html',
                      context={'form': form, 'current_url': lambda: url,
                               'from_multiuse_invite': from_multiuse_invite},
                      )

def validate_account_and_subdomain(request: HttpRequest, user_profile: UserProfile) -> None:
    if user_profile.realm.deactivated:
        raise JsonableError(_("This organization has been deactivated"))
    if not user_profile.is_active:
        raise JsonableError(_("Account is deactivated"))

    # Either the subdomain matches, or processing a websockets message
    # in the message_sender worker (which will have already had the
    # subdomain validated), or we're accessing Tornado from and to
    # localhost (aka spoofing a request as the user).
    if (not user_matches_subdomain(get_subdomain(request), user_profile) and
        not (request.method == "SOCKET" and
             request.META['SERVER_NAME'] == "127.0.0.1") and
        not (settings.RUNNING_INSIDE_TORNADO and
             request.META["SERVER_NAME"] == "127.0.0.1" and
             request.META["REMOTE_ADDR"] == "127.0.0.1")):
        logging.warning("User %s (%s) attempted to access API on wrong subdomain (%s)" % (
            user_profile.email, user_profile.realm.subdomain, get_subdomain(request)))
        raise JsonableError(_("Account is not associated with this subdomain"))

def json_fetch_api_key(request: HttpRequest, user_profile: UserProfile,
                       password: str=REQ(default='')) -> HttpResponse:
    subdomain = get_subdomain(request)
    realm = get_realm(subdomain)
    if password_auth_enabled(user_profile.realm):
        if not authenticate(username=user_profile.email, password=password,
                            realm=realm):
            return json_error(_("Your username or password is incorrect."))

    api_key = get_api_key(user_profile)
    return json_success({"api_key": api_key})

def dev_direct_login(request, **kwargs):
    # type: (HttpRequest, **Any) -> HttpResponse
    # This function allows logging in without a password and should only be called in development environments.
    # It may be called if the DevAuthBackend is included in settings.AUTHENTICATION_BACKENDS
    if (not dev_auth_enabled()) or settings.PRODUCTION:
        # This check is probably not required, since authenticate would fail without an enabled DevAuthBackend.
        raise Exception('Direct login not supported.')
    email = request.POST['direct_email']
    user_profile = authenticate(username=email, realm_subdomain=get_subdomain(request))
    if user_profile is None:
        raise Exception("User cannot login")
    do_login(request, user_profile)
    return HttpResponseRedirect(user_profile.realm.uri)

def password_reset(request: HttpRequest, **kwargs: Any) -> HttpResponse:
    realm = get_realm(get_subdomain(request))

    if realm is None:
        # If trying to get to password reset on a subdomain that
        # doesn't exist, just go to find_account.
        redirect_url = reverse('zerver.views.registration.find_account')
        return HttpResponseRedirect(redirect_url)

    return django_password_reset(request,
                                 template_name='zerver/reset.html',
                                 password_reset_form=ZulipPasswordResetForm,
                                 post_reset_redirect='/accounts/password/reset/done/')