def clean_email(self) -> str:
        """Returns the email if and only if the user's email address is
        allowed to join the realm they are trying to join."""
        email = self.cleaned_data['email']

        # Otherwise, the user is trying to join a specific realm.
        realm = self.realm
        from_multiuse_invite = self.from_multiuse_invite

        if realm is None:
            raise ValidationError(_("The organization you are trying to "
                                    "join using {email} does not "
                                    "exist.").format(email=email))

        if not from_multiuse_invite and realm.invite_required:
            raise ValidationError(_("Please request an invite for {email} "
                                    "from the organization "
                                    "administrator.").format(email=email))

        try:
            email_allowed_for_realm(email, realm)
        except DomainNotAllowedForRealmError:
            raise ValidationError(
                _("Your email address, {email}, is not in one of the domains "
                  "that are allowed to register for accounts in this organization.").format(
                      string_id=realm.string_id, email=email))
        except DisposableEmailError:
            raise ValidationError(_("Please use your real email address."))

        validate_email_for_realm(realm, email)

        if realm.is_zephyr_mirror_realm:
            email_is_not_mit_mailing_list(email)

        return email

def create_user_backend(request: HttpRequest, user_profile: UserProfile,
                        email: str=REQ(), password: str=REQ(), full_name_raw: str=REQ("full_name"),
                        short_name: str=REQ()) -> HttpResponse:
    full_name = check_full_name(full_name_raw)
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        return json_error(_('Bad name or username'))

    # Check that the new user's email address belongs to the admin's realm
    # (Since this is an admin API, we don't require the user to have been
    # invited first.)
    realm = user_profile.realm
    try:
        email_allowed_for_realm(email, user_profile.realm)
    except DomainNotAllowedForRealmError:
        return json_error(_("Email '%(email)s' not allowed in this organization") %
                          {'email': email})
    except DisposableEmailError:
        return json_error(_("Disposable email addresses are not allowed in this organization"))
    except EmailContainsPlusError:
        return json_error(_("Email addresses containing + are not allowed."))

    try:
        get_user_by_delivery_email(email, user_profile.realm)
        return json_error(_("Email '%s' already in use") % (email,))
    except UserProfile.DoesNotExist:
        pass

    do_create_user(email, password, realm, full_name, short_name)
    return json_success()

    def handle(self, *args: Any, **options: Any) -> None:
        duplicates = False
        realm = self.get_realm(options)
        assert realm is not None  # Should be ensured by parser

        if not options['emails']:
            self.print_help("./manage.py", "generate_invite_links")
            exit(1)

        for email in options['emails']:
            try:
                self.get_user(email, realm)
                print(email + ": There is already a user registered with that address.")
                duplicates = True
                continue
            except CommandError:
                pass

        if duplicates:
            return

        for email in options['emails']:
            try:
                email_allowed_for_realm(email, realm)
            except DomainNotAllowedForRealmError:
                if not options["force"]:
                    print("You've asked to add an external user '%s' to a closed realm '%s'." % (
                        email, realm.string_id))
                    print("Are you sure? To do this, pass --force.")
                    exit(1)

            prereg_user = PreregistrationUser(email=email, realm=realm)
            prereg_user.save()
            print(email + ": " + create_confirmation_link(prereg_user, realm.host,
                                                          Confirmation.INVITATION))

    def clean_email(self):
        # type: () -> str
        """Returns the email if and only if the user's email address is
        allowed to join the realm they are trying to join."""
        email = self.cleaned_data['email']

        if get_unique_open_realm():
            return email

        # Otherwise, the user is trying to join a specific realm.
        realm = None
        if self.string_id:
            realm = get_realm_by_string_id(self.string_id)
        elif not settings.REALMS_HAVE_SUBDOMAINS:
            realm = get_realm_by_email_domain(email)

        if realm is None:
            if settings.REALMS_HAVE_SUBDOMAINS:
                raise ValidationError(_("The organization you are trying to join does not exist."))
            else:
                raise ValidationError(_("Your email address does not correspond to any existing organization."))

        if realm.invite_required:
            raise ValidationError(_("Please request an invite from the organization administrator."))

        if not email_allowed_for_realm(email, realm):
            raise ValidationError(
                _("The organization you are trying to join, %(string_id)s, only allows users with e-mail "
                  "addresses within the organization. Please try a different e-mail address."
                  % {'string_id': realm.string_id}))

        if realm.is_zephyr_mirror_realm:
            email_is_not_mit_mailing_list(email)

        return email

def create_user_backend(request, user_profile, email=REQ(), password=REQ(),
                        full_name=REQ(), short_name=REQ()):
    # type: (HttpRequest, UserProfile, text_type, text_type, text_type, text_type) -> HttpResponse
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        return json_error(_('Bad name or username'))

    # Check that the new user's email address belongs to the admin's realm
    # (Since this is an admin API, we don't require the user to have been
    # invited first.)
    realm = user_profile.realm
    if not email_allowed_for_realm(email, user_profile.realm):
        return json_error(_("Email '%(email)s' does not belong to domain '%(domain)s'") %
                          {'email': email, 'domain': realm.domain})

    try:
        user_profile = get_user_profile_by_email(email)
        return json_error(
            _("Email '%s' already in use") % (email,),
            dict(api_key=user_profile.api_key)
        )
    except UserProfile.DoesNotExist:
        pass

    user_profile = do_create_user(email, password, realm, full_name, short_name)
    return json_success(dict(api_key=user_profile.api_key))

    def test_email_allowed_for_realm(self) -> None:
        realm1 = do_create_realm('testrealm1', 'Test Realm 1', restricted_to_domain=True)
        realm2 = do_create_realm('testrealm2', 'Test Realm 2', restricted_to_domain=True)

        realm_domain = RealmDomain.objects.create(realm=realm1, domain='test1.com', allow_subdomains=False)
        RealmDomain.objects.create(realm=realm2, domain='test2.test1.com', allow_subdomains=True)

        self.assertEqual(email_allowed_for_realm('[email protected]', realm1), True)
        self.assertEqual(email_allowed_for_realm('[email protected]', realm1), False)
        self.assertEqual(email_allowed_for_realm('[email protected]', realm2), True)
        self.assertEqual(email_allowed_for_realm('[email protected]', realm2), True)
        self.assertEqual(email_allowed_for_realm('[email protected]', realm2), False)

        do_change_realm_domain(realm_domain, True)
        self.assertEqual(email_allowed_for_realm('[email protected]', realm1), True)
        self.assertEqual(email_allowed_for_realm('[email protected]', realm1), True)
        self.assertEqual(email_allowed_for_realm('[email protected]', realm1), False)

def create_user_backend(request, user_profile, email=REQ, password=REQ,
                        full_name=REQ, short_name=REQ):
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        return json_error('Bad name or username')

    # Check that the new user's email address belongs to the admin's realm
    # (Since this is an admin API, we don't require the user to have been
    # invited first.)
    realm = user_profile.realm
    if not email_allowed_for_realm(email, user_profile.realm):
        return json_error("Email '%s' does not belong to domain '%s'" % (email, realm.domain))

    try:
        get_user_profile_by_email(email)
        return json_error("Email '%s' already in use" % (email,))
    except UserProfile.DoesNotExist:
        pass

    do_create_user(email, password, realm, full_name, short_name)
    return json_success()

    def clean_email(self):
        # type: () -> str
        """Returns the email if and only if the user's email address is
        allowed to join the realm they are trying to join."""
        email = self.cleaned_data['email']

        if get_unique_open_realm():
            return email

        # Otherwise, the user is trying to join a specific realm.
        realm = self.realm
        if realm is None and not settings.REALMS_HAVE_SUBDOMAINS:
            realm = get_realm_by_email_domain(email)

        if realm is None:
            if settings.REALMS_HAVE_SUBDOMAINS:
                raise ValidationError(_("The organization you are trying to "
                                        "join using {email} does not "
                                        "exist.").format(email=email))
            else:
                raise ValidationError(_("Your email address, {email}, does not "
                                        "correspond to any existing "
                                        "organization.").format(email=email))

        if realm.invite_required:
            raise ValidationError(_("Please request an invite for {email} "
                                    "from the organization "
                                    "administrator.").format(email=email))

        if not email_allowed_for_realm(email, realm):
            raise ValidationError(
                _("Your email address, {email}, is not in one of the domains "
                  "that are allowed to register for accounts in this organization.").format(
                      string_id=realm.string_id, email=email))

        if realm.is_zephyr_mirror_realm:
            email_is_not_mit_mailing_list(email)

        return email

    def handle(self, *args, **options):
        # type: (*Any, **Any) -> None
        duplicates = False
        for email in options['emails']:
            try:
                get_user_profile_by_email(email)
                print(email + ": There is already a user registered with that address.")
                duplicates = True
                continue
            except UserProfile.DoesNotExist:
                pass

        if duplicates:
            return

        realm = None
        string_id = options["string_id"]
        if string_id:
            realm = get_realm(string_id)
        if not realm:
            print("The realm %s doesn't exist yet, please create it first." % (string_id,))
            print("Don't forget default streams!")
            exit(1)

        for email in options['emails']:
            if realm:
                if not email_allowed_for_realm(email, realm) and not options["force"]:
                    print("You've asked to add an external user (%s) to a closed realm (%s)." % (
                        email, string_id))
                    print("Are you sure? To do this, pass --force.")
                    exit(1)
                else:
                    prereg_user = PreregistrationUser(email=email, realm=realm)
            else:
                prereg_user = PreregistrationUser(email=email)
            prereg_user.save()
            print(email + ": " + Confirmation.objects.get_link_for_object(prereg_user, host=realm.host))

def create_user_backend(request, user_profile, email=REQ(), password=REQ(),
                        full_name_raw=REQ("full_name"), short_name=REQ()):
    # type: (HttpRequest, UserProfile, Text, Text, Text, Text) -> HttpResponse
    full_name = check_full_name(full_name_raw)
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        return json_error(_('Bad name or username'))

    # Check that the new user's email address belongs to the admin's realm
    # (Since this is an admin API, we don't require the user to have been
    # invited first.)
    realm = user_profile.realm
    if not email_allowed_for_realm(email, user_profile.realm):
        return json_error(_("Email '%(email)s' not allowed for realm '%(realm)s'") %
                          {'email': email, 'realm': realm.string_id})

    try:
        get_user_profile_by_email(email)
        return json_error(_("Email '%s' already in use") % (email,))
    except UserProfile.DoesNotExist:
        pass

    do_create_user(email, password, realm, full_name, short_name)
    return json_success()

def accounts_register(request: HttpRequest) -> HttpResponse:
    key = request.POST['key']
    confirmation = Confirmation.objects.get(confirmation_key=key)
    prereg_user = confirmation.content_object
    email = prereg_user.email
    realm_creation = prereg_user.realm_creation
    password_required = prereg_user.password_required
    is_realm_admin = prereg_user.invited_as_admin or realm_creation

    try:
        validators.validate_email(email)
    except ValidationError:
        return render(request, "zerver/invalid_email.html", context={"invalid_email": True})

    if realm_creation:
        # For creating a new realm, there is no existing realm or domain
        realm = None
    else:
        realm = get_realm(get_subdomain(request))
        if realm is None or realm != prereg_user.realm:
            return render_confirmation_key_error(
                request, ConfirmationKeyException(ConfirmationKeyException.DOES_NOT_EXIST))

        try:
            email_allowed_for_realm(email, realm)
        except DomainNotAllowedForRealmError:
            return render(request, "zerver/invalid_email.html",
                          context={"realm_name": realm.name, "closed_domain": True})
        except DisposableEmailError:
            return render(request, "zerver/invalid_email.html",
                          context={"realm_name": realm.name, "disposable_emails_not_allowed": True})
        except EmailContainsPlusError:
            return render(request, "zerver/invalid_email.html",
                          context={"realm_name": realm.name, "email_contains_plus": True})

        if realm.deactivated:
            # The user is trying to register for a deactivated realm. Advise them to
            # contact support.
            return redirect_to_deactivation_notice()

        try:
            validate_email_for_realm(realm, email)
        except ValidationError:  # nocoverage # We need to add a test for this.
            return HttpResponseRedirect(reverse('django.contrib.auth.views.login') + '?email=' +
                                        urllib.parse.quote_plus(email))

    name_validated = False
    full_name = None

    if request.POST.get('from_confirmation'):
        try:
            del request.session['authenticated_full_name']
        except KeyError:
            pass
        if realm is not None and realm.is_zephyr_mirror_realm:
            # For MIT users, we can get an authoritative name from Hesiod.
            # Technically we should check that this is actually an MIT
            # realm, but we can cross that bridge if we ever get a non-MIT
            # zephyr mirroring realm.
            hesiod_name = compute_mit_user_fullname(email)
            form = RegistrationForm(
                initial={'full_name': hesiod_name if "@" not in hesiod_name else ""},
                realm_creation=realm_creation)
            name_validated = True
        elif settings.POPULATE_PROFILE_VIA_LDAP:
            for backend in get_backends():
                if isinstance(backend, LDAPBackend):
                    try:
                        ldap_username = backend.django_to_ldap_username(email)
                    except ZulipLDAPException:
                        logging.warning("New account email %s could not be found in LDAP" % (email,))
                        form = RegistrationForm(realm_creation=realm_creation)
                        break

                    ldap_attrs = _LDAPUser(backend, ldap_username).attrs

                    try:
                        ldap_full_name = ldap_attrs[settings.AUTH_LDAP_USER_ATTR_MAP['full_name']][0]
                        request.session['authenticated_full_name'] = ldap_full_name
                        name_validated = True
                        # We don't use initial= here, because if the form is
                        # complete (that is, no additional fields need to be
                        # filled out by the user) we want the form to validate,
                        # so they can be directly registered without having to
                        # go through this interstitial.
                        form = RegistrationForm({'full_name': ldap_full_name},
                                                realm_creation=realm_creation)
                        # FIXME: This will result in the user getting
                        # validation errors if they have to enter a password.
                        # Not relevant for ONLY_SSO, though.
                        break
                    except TypeError:
                        # Let the user fill out a name and/or try another backend
                        form = RegistrationForm(realm_creation=realm_creation)
        elif 'full_name' in request.POST:
            form = RegistrationForm(
                initial={'full_name': request.POST.get('full_name')},
                realm_creation=realm_creation
            )
        else:
#.........这里部分代码省略.........

def accounts_register(request):
    # type: (HttpRequest) -> HttpResponse
    key = request.POST['key']
    confirmation = Confirmation.objects.get(confirmation_key=key)
    prereg_user = confirmation.content_object
    email = prereg_user.email
    realm_creation = prereg_user.realm_creation
    try:
        existing_user_profile = get_user_profile_by_email(email)
    except UserProfile.DoesNotExist:
        existing_user_profile = None

    validators.validate_email(email)
    # If OPEN_REALM_CREATION is enabled all user sign ups should go through the
    # special URL with domain name so that REALM can be identified if multiple realms exist
    unique_open_realm = get_unique_open_realm()
    if unique_open_realm is not None:
        realm = unique_open_realm
        domain = realm.domain
    elif prereg_user.referred_by:
        # If someone invited you, you are joining their realm regardless
        # of your e-mail address.
        realm = prereg_user.referred_by.realm
        domain = realm.domain
        if not email_allowed_for_realm(email, realm):
            return render_to_response("zerver/closed_realm.html", {"closed_domain_name": realm.name})
    elif prereg_user.realm:
        # You have a realm set, even though nobody referred you. This
        # happens if you sign up through a special URL for an open
        # realm.
        domain = prereg_user.realm.domain
        realm = get_realm(domain)
    else:
        domain = resolve_email_to_domain(email)
        realm = get_realm(domain)

    if realm and realm.deactivated:
        # The user is trying to register for a deactivated realm. Advise them to
        # contact support.
        return render_to_response("zerver/deactivated.html",
                                  {"deactivated_domain_name": realm.name,
                                   "zulip_administrator": settings.ZULIP_ADMINISTRATOR})

    try:
        if existing_user_profile is not None and existing_user_profile.is_mirror_dummy:
            # Mirror dummy users to be activated must be inactive
            is_inactive(email)
        else:
            # Other users should not already exist at all.
            user_email_is_unique(email)
    except ValidationError:
        return HttpResponseRedirect(reverse('django.contrib.auth.views.login') + '?email=' +
                                    urllib.parse.quote_plus(email))

    name_validated = False
    full_name = None

    if request.POST.get('from_confirmation'):
        try:
            del request.session['authenticated_full_name']
        except KeyError:
            pass
        if realm is not None and realm.is_zephyr_mirror_realm and domain == "mit.edu":
            # for MIT users, we can get an authoritative name from Hesiod
            hesiod_name = compute_mit_user_fullname(email)
            form = RegistrationForm(
                    initial={'full_name': hesiod_name if "@" not in hesiod_name else ""})
            name_validated = True
        elif settings.POPULATE_PROFILE_VIA_LDAP:
            for backend in get_backends():
                if isinstance(backend, LDAPBackend):
                    ldap_attrs = _LDAPUser(backend, backend.django_to_ldap_username(email)).attrs
                    try:
                        ldap_full_name = ldap_attrs[settings.AUTH_LDAP_USER_ATTR_MAP['full_name']][0]
                        request.session['authenticated_full_name'] = ldap_full_name
                        name_validated = True
                        # We don't use initial= here, because if the form is
                        # complete (that is, no additional fields need to be
                        # filled out by the user) we want the form to validate,
                        # so they can be directly registered without having to
                        # go through this interstitial.
                        form = RegistrationForm({'full_name': ldap_full_name})
                        # FIXME: This will result in the user getting
                        # validation errors if they have to enter a password.
                        # Not relevant for ONLY_SSO, though.
                        break
                    except TypeError:
                        # Let the user fill out a name and/or try another backend
                        form = RegistrationForm()
        elif 'full_name' in request.POST:
            form = RegistrationForm(
                initial={'full_name': request.POST.get('full_name')}
            )
        else:
            form = RegistrationForm()
    else:
        postdata = request.POST.copy()
        if name_changes_disabled(realm):
            # If we populate profile information via LDAP and we have a
            # verified name from you on file, use that. Otherwise, fall
#.........这里部分代码省略.........

    def test_email_allowed_for_realm(self) -> None:
        realm1 = do_create_realm('testrealm1', 'Test Realm 1', restricted_to_domain=True)
        realm2 = do_create_realm('testrealm2', 'Test Realm 2', restricted_to_domain=True)

        realm_domain = RealmDomain.objects.create(realm=realm1, domain='test1.com', allow_subdomains=False)
        RealmDomain.objects.create(realm=realm2, domain='test2.test1.com', allow_subdomains=True)

        email_allowed_for_realm('[email protected]', realm1)
        with self.assertRaises(DomainNotAllowedForRealmError):
            email_allowed_for_realm('[email protected]', realm1)
        email_allowed_for_realm('[email protected]', realm2)
        email_allowed_for_realm('[email protected]', realm2)
        with self.assertRaises(DomainNotAllowedForRealmError):
            email_allowed_for_realm('[email protected]', realm2)

        do_change_realm_domain(realm_domain, True)
        email_allowed_for_realm('[email protected]', realm1)
        email_allowed_for_realm('[email protected]', realm1)
        with self.assertRaises(DomainNotAllowedForRealmError):
            email_allowed_for_realm('[email protected]', realm1)

    def get_or_build_user(self, username: str, ldap_user: _LDAPUser) -> Tuple[UserProfile, bool]:
        """The main function of our authentication backend extension of
        django-auth-ldap.  When this is called (from `authenticate`),
        django-auth-ldap will already have verified that the provided
        username and password match those in the LDAP database.

        This function's responsibility is to check (1) whether the
        email address for this user obtained from LDAP has an active
        account in this Zulip realm.  If so, it will log them in.

        Otherwise, to provide a seamless Single Sign-On experience
        with LDAP, this function can automatically create a new Zulip
        user account in the realm (assuming the realm is configured to
        allow that email address to sign up).
        """
        return_data = {}  # type: Dict[str, Any]

        if settings.LDAP_EMAIL_ATTR is not None:
            # Get email from ldap attributes.
            if settings.LDAP_EMAIL_ATTR not in ldap_user.attrs:
                return_data["ldap_missing_attribute"] = settings.LDAP_EMAIL_ATTR
                raise ZulipLDAPException("LDAP user doesn't have the needed %s attribute" % (
                    settings.LDAP_EMAIL_ATTR,))

            username = ldap_user.attrs[settings.LDAP_EMAIL_ATTR][0]

        if 'userAccountControl' in settings.AUTH_LDAP_USER_ATTR_MAP:  # nocoverage
            ldap_disabled = self.is_account_control_disabled_user(ldap_user)
            if ldap_disabled:
                # Treat disabled users as deactivated in Zulip.
                return_data["inactive_user"] = True
                raise ZulipLDAPException("User has been deactivated")

        user_profile = common_get_active_user(username, self._realm, return_data)
        if user_profile is not None:
            # An existing user, successfully authed; return it.
            return user_profile, False

        if return_data.get("inactive_realm"):
            # This happens if there is a user account in a deactivated realm
            raise ZulipLDAPException("Realm has been deactivated")
        if return_data.get("inactive_user"):
            raise ZulipLDAPException("User has been deactivated")
        # An invalid_subdomain `return_data` value here is ignored,
        # since that just means we're trying to create an account in a
        # second realm on the server (`ldap_auth_enabled(realm)` would
        # have been false if this user wasn't meant to have an account
        # in this second realm).
        if self._realm.deactivated:
            # This happens if no account exists, but the realm is
            # deactivated, so we shouldn't create a new user account
            raise ZulipLDAPException("Realm has been deactivated")

        # Makes sure that email domain hasn't be restricted for this
        # realm.  The main thing here is email_allowed_for_realm; but
        # we also call validate_email_for_realm just for consistency,
        # even though its checks were already done above.
        try:
            email_allowed_for_realm(username, self._realm)
            validate_email_for_realm(self._realm, username)
        except DomainNotAllowedForRealmError:
            raise ZulipLDAPException("This email domain isn't allowed in this organization.")
        except (DisposableEmailError, EmailContainsPlusError):
            raise ZulipLDAPException("Email validation failed.")

        # We have valid LDAP credentials; time to create an account.
        full_name, short_name = self.get_mapped_name(ldap_user)
        try:
            full_name = check_full_name(full_name)
        except JsonableError as e:
            raise ZulipLDAPException(e.msg)

        opts = {}   # type: Dict[str, Any]
        if self._prereg_user:
            invited_as = self._prereg_user.invited_as
            opts['prereg_user'] = self._prereg_user
            opts['is_realm_admin'] = invited_as == PreregistrationUser.INVITE_AS['REALM_ADMIN']
            opts['is_guest'] = invited_as == PreregistrationUser.INVITE_AS['GUEST_USER']
            opts['default_stream_groups'] = get_default_stream_groups(self._realm)

        user_profile = do_create_user(username, None, self._realm, full_name, short_name, **opts)
        self.sync_avatar_from_ldap(user_profile, ldap_user)
        self.sync_custom_profile_fields_from_ldap(user_profile, ldap_user)

        return user_profile, True