mirror of
https://github.com/zulip/zulip.git
synced 2025-10-23 16:14:02 +00:00
6c069f4365
Limiting lookups by delivery_email to users with "everyone" email visibility is overly simplistic. We can successfully do these lookups whenever the requester has the permission to view the real email address of the user they're looking up.
264 lines
9.7 KiB
Python
264 lines
9.7 KiB
Python
import re
|
|
from datetime import datetime
|
|
from email.headerregistry import Address
|
|
|
|
from django.contrib.auth.models import UserManager
|
|
from django.utils.timezone import now as timezone_now
|
|
|
|
from zerver.lib.i18n import get_default_language_for_new_user
|
|
from zerver.lib.onboarding_steps import copy_onboarding_steps
|
|
from zerver.lib.timezone import canonicalize_timezone
|
|
from zerver.lib.upload import copy_avatar
|
|
from zerver.models import (
|
|
Realm,
|
|
RealmUserDefault,
|
|
Recipient,
|
|
Stream,
|
|
Subscription,
|
|
UserBaseSettings,
|
|
UserProfile,
|
|
)
|
|
from zerver.models.realms import get_fake_email_domain
|
|
|
|
|
|
def copy_default_settings(
|
|
settings_source: UserProfile | RealmUserDefault, target_profile: UserProfile
|
|
) -> None:
|
|
# Important note: Code run from here to configure the user's
|
|
# settings should not send events, as that would cause clients
|
|
# to throw an exception (we haven't sent the realm_user/add event
|
|
# yet, so that event will include the updated details of target_profile).
|
|
#
|
|
# Note that this function will do at least one save() on target_profile.
|
|
for settings_name in UserBaseSettings.property_types:
|
|
if settings_name in ["default_language", "enable_login_emails"] and isinstance(
|
|
settings_source, RealmUserDefault
|
|
):
|
|
continue
|
|
|
|
if settings_name == "email_address_visibility":
|
|
# For email_address_visibility, the value selected in registration form
|
|
# is preferred over the realm-level default value and value of source
|
|
# profile.
|
|
continue
|
|
value = getattr(settings_source, settings_name)
|
|
setattr(target_profile, settings_name, value)
|
|
|
|
if isinstance(settings_source, RealmUserDefault):
|
|
target_profile.save()
|
|
return
|
|
|
|
target_profile.full_name = settings_source.full_name
|
|
target_profile.timezone = canonicalize_timezone(settings_source.timezone)
|
|
target_profile.save()
|
|
|
|
if settings_source.avatar_source == UserProfile.AVATAR_FROM_USER:
|
|
from zerver.actions.user_settings import do_change_avatar_fields
|
|
|
|
copy_avatar(settings_source, target_profile)
|
|
do_change_avatar_fields(
|
|
target_profile,
|
|
UserProfile.AVATAR_FROM_USER,
|
|
skip_notify=True,
|
|
acting_user=target_profile,
|
|
)
|
|
|
|
copy_onboarding_steps(settings_source, target_profile)
|
|
|
|
|
|
def get_dummy_email_address_for_display_regex(realm: Realm) -> str:
|
|
"""
|
|
Returns a regex that matches the format of dummy email addresses we
|
|
generate for the .email of users with limit email_address_visibility.
|
|
|
|
The reason we need a regex is that we want something that we can use both
|
|
for generating the dummy email addresses and recognizing them together with extraction
|
|
of the user ID.
|
|
"""
|
|
|
|
# We can't directly have (\d+) in the username passed to Address, because it gets
|
|
# mutated by the underlying logic for escaping special characters.
|
|
# So we use a trick by using $ as a placeholder which will be preserved, and then
|
|
# replace it with (\d+) to obtain our intended regex.
|
|
address_template = Address(username="user$", domain=get_fake_email_domain(realm.host)).addr_spec
|
|
regex = re.escape(address_template).replace(r"\$", r"(\d+)", 1)
|
|
return regex
|
|
|
|
|
|
def get_display_email_address(user_profile: UserProfile) -> str:
|
|
if not user_profile.email_address_is_realm_public():
|
|
# The format of the dummy email address created here needs to stay consistent
|
|
# with get_dummy_email_address_for_display_regex.
|
|
return Address(
|
|
username=f"user{user_profile.id}", domain=get_fake_email_domain(user_profile.realm.host)
|
|
).addr_spec
|
|
return user_profile.delivery_email
|
|
|
|
|
|
# create_user_profile is based on Django's User.objects.create_user,
|
|
# except that we don't save to the database so it can used in
|
|
# bulk_creates
|
|
#
|
|
# Only use this for bulk_create -- for normal usage one should use
|
|
# create_user (below) which will also make the Subscription and
|
|
# Recipient objects
|
|
def create_user_profile(
|
|
realm: Realm,
|
|
email: str,
|
|
password: str | None,
|
|
active: bool,
|
|
bot_type: int | None,
|
|
full_name: str,
|
|
bot_owner: UserProfile | None,
|
|
is_mirror_dummy: bool,
|
|
tos_version: str | None,
|
|
timezone: str,
|
|
default_language: str,
|
|
force_id: int | None = None,
|
|
force_date_joined: datetime | None = None,
|
|
*,
|
|
email_address_visibility: int,
|
|
) -> UserProfile:
|
|
if force_date_joined is None:
|
|
date_joined = timezone_now()
|
|
else:
|
|
date_joined = force_date_joined
|
|
|
|
email = UserManager.normalize_email(email)
|
|
|
|
extra_kwargs = {}
|
|
if force_id is not None:
|
|
extra_kwargs["id"] = force_id
|
|
|
|
user_profile = UserProfile(
|
|
is_staff=False,
|
|
is_active=active,
|
|
full_name=full_name,
|
|
last_login=date_joined,
|
|
date_joined=date_joined,
|
|
realm=realm,
|
|
is_bot=bool(bot_type),
|
|
bot_type=bot_type,
|
|
bot_owner=bot_owner,
|
|
is_mirror_dummy=is_mirror_dummy,
|
|
tos_version=tos_version,
|
|
timezone=timezone,
|
|
default_language=default_language,
|
|
delivery_email=email,
|
|
email_address_visibility=email_address_visibility,
|
|
**extra_kwargs,
|
|
)
|
|
if bot_type or not active:
|
|
password = None
|
|
if user_profile.email_address_is_realm_public():
|
|
# If emails are visible to everyone, we can set this here and save a DB query
|
|
user_profile.email = get_display_email_address(user_profile)
|
|
user_profile.set_password(password)
|
|
return user_profile
|
|
|
|
|
|
def create_user(
|
|
email: str,
|
|
password: str | None,
|
|
realm: Realm,
|
|
full_name: str,
|
|
active: bool = True,
|
|
role: int | None = None,
|
|
bot_type: int | None = None,
|
|
bot_owner: UserProfile | None = None,
|
|
tos_version: str | None = None,
|
|
timezone: str = "",
|
|
avatar_source: str = UserProfile.AVATAR_FROM_GRAVATAR,
|
|
is_mirror_dummy: bool = False,
|
|
default_language: str | None = None,
|
|
default_sending_stream: Stream | None = None,
|
|
default_events_register_stream: Stream | None = None,
|
|
default_all_public_streams: bool | None = None,
|
|
source_profile: UserProfile | None = None,
|
|
force_id: int | None = None,
|
|
force_date_joined: datetime | None = None,
|
|
create_personal_recipient: bool = True,
|
|
enable_marketing_emails: bool | None = None,
|
|
email_address_visibility: int | None = None,
|
|
) -> UserProfile:
|
|
realm_user_default = RealmUserDefault.objects.get(realm=realm)
|
|
if bot_type is None:
|
|
if email_address_visibility is not None:
|
|
user_email_address_visibility = email_address_visibility
|
|
else:
|
|
user_email_address_visibility = realm_user_default.email_address_visibility
|
|
else:
|
|
# There is no privacy motivation for limiting access to bot email addresses,
|
|
# so we hardcode them to EMAIL_ADDRESS_VISIBILITY_EVERYONE.
|
|
user_email_address_visibility = UserProfile.EMAIL_ADDRESS_VISIBILITY_EVERYONE
|
|
|
|
# Users created via the API or LDAP/SAML syncing code paths will
|
|
# usually not have a default_language value, and should fall back
|
|
# to the realm default.
|
|
if default_language is None:
|
|
default_language = get_default_language_for_new_user(realm, request=None)
|
|
|
|
user_profile = create_user_profile(
|
|
realm,
|
|
email,
|
|
password,
|
|
active,
|
|
bot_type,
|
|
full_name,
|
|
bot_owner,
|
|
is_mirror_dummy,
|
|
tos_version,
|
|
timezone,
|
|
default_language,
|
|
force_id=force_id,
|
|
force_date_joined=force_date_joined,
|
|
email_address_visibility=user_email_address_visibility,
|
|
)
|
|
user_profile.avatar_source = avatar_source
|
|
user_profile.timezone = timezone
|
|
user_profile.default_sending_stream = default_sending_stream
|
|
user_profile.default_events_register_stream = default_events_register_stream
|
|
if role is not None:
|
|
user_profile.role = role
|
|
# Allow the ORM default to be used if not provided
|
|
if default_all_public_streams is not None:
|
|
user_profile.default_all_public_streams = default_all_public_streams
|
|
# If a source profile was specified, we copy settings from that
|
|
# user. Note that this is positioned in a way that overrides
|
|
# other arguments passed in, which is correct for most defaults
|
|
# like time zone where the source profile likely has a better value
|
|
# than the guess. As we decide on details like avatars and full
|
|
# names for this feature, we may want to move it.
|
|
if source_profile is not None:
|
|
# copy_default_settings saves the attribute values so a secondary
|
|
# save is not required.
|
|
copy_default_settings(source_profile, user_profile)
|
|
elif bot_type is None:
|
|
copy_default_settings(realm_user_default, user_profile)
|
|
else:
|
|
# This will be executed only for bots.
|
|
user_profile.save()
|
|
|
|
if bot_type is None and enable_marketing_emails is not None:
|
|
user_profile.enable_marketing_emails = enable_marketing_emails
|
|
user_profile.save(update_fields=["enable_marketing_emails"])
|
|
|
|
if not user_profile.email_address_is_realm_public():
|
|
# With restricted access to email addresses, we can't generate
|
|
# the fake email addresses we use for display purposes without
|
|
# a User ID, which isn't generated until the .save() above.
|
|
user_profile.email = get_display_email_address(user_profile)
|
|
user_profile.save(update_fields=["email"])
|
|
|
|
if not create_personal_recipient:
|
|
return user_profile
|
|
|
|
recipient = Recipient.objects.create(type_id=user_profile.id, type=Recipient.PERSONAL)
|
|
user_profile.recipient = recipient
|
|
user_profile.save(update_fields=["recipient"])
|
|
|
|
Subscription.objects.create(
|
|
user_profile=user_profile, recipient=recipient, is_user_active=user_profile.is_active
|
|
)
|
|
return user_profile
|