mirror of
https://github.com/zulip/zulip.git
synced 2025-11-04 05:53:43 +00:00
fe993032a6
In user signup context, we are okay with there being an existing mirror dummy user with the matching email - at the end of the signup, that mirror dummy account will be activated and control of it given to the user doing this signup. However, in email change contexts (SCIM API and regular email change flow), we can't change an account's email address to the address that already belongs to an existing mirror dummy user. To avoid subtle bugs like this, we make callers have to explicitly specify whether existance of mirror dummies with the matching email address is okay or not.
495 lines
20 KiB
Python
495 lines
20 KiB
Python
from email.headerregistry import Address
|
|
from typing import Annotated, Any
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import authenticate, update_session_auth_hash
|
|
from django.contrib.auth.tokens import default_token_generator
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.files.uploadedfile import UploadedFile
|
|
from django.db import transaction
|
|
from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
|
|
from django.shortcuts import render
|
|
from django.urls import reverse
|
|
from django.utils.html import escape
|
|
from django.utils.safestring import SafeString
|
|
from django.utils.translation import gettext as _
|
|
from django.utils.translation import gettext_lazy
|
|
from pydantic import Json
|
|
from pydantic.functional_validators import AfterValidator
|
|
|
|
from confirmation.models import (
|
|
Confirmation,
|
|
ConfirmationKeyError,
|
|
get_object_from_key,
|
|
render_confirmation_key_error,
|
|
)
|
|
from zerver.actions.user_settings import (
|
|
check_change_full_name,
|
|
do_change_avatar_fields,
|
|
do_change_password,
|
|
do_change_user_delivery_email,
|
|
do_change_user_setting,
|
|
do_regenerate_api_key,
|
|
do_start_email_change_process,
|
|
)
|
|
from zerver.actions.users import generate_password_reset_url
|
|
from zerver.decorator import human_users_only, require_post
|
|
from zerver.lib.avatar import avatar_url
|
|
from zerver.lib.email_notifications import enqueue_welcome_emails
|
|
from zerver.lib.email_validation import (
|
|
get_realm_email_validator,
|
|
validate_email_is_valid,
|
|
validate_email_not_already_in_realm,
|
|
)
|
|
from zerver.lib.exceptions import JsonableError, RateLimitedError, UserDeactivatedError
|
|
from zerver.lib.i18n import get_available_language_codes
|
|
from zerver.lib.rate_limiter import RateLimitedUser
|
|
from zerver.lib.response import json_success
|
|
from zerver.lib.send_email import FromAddress, send_email
|
|
from zerver.lib.sounds import get_available_notification_sounds
|
|
from zerver.lib.typed_endpoint import typed_endpoint, typed_endpoint_without_parameters
|
|
from zerver.lib.typed_endpoint_validators import (
|
|
check_int_in_validator,
|
|
check_string_in_validator,
|
|
parse_enum_from_string_value,
|
|
timezone_validator,
|
|
)
|
|
from zerver.lib.upload import upload_avatar_image
|
|
from zerver.models import EmailChangeStatus, UserProfile
|
|
from zerver.models.realms import avatar_changes_disabled, name_changes_disabled
|
|
from zerver.models.users import ResolvedTopicNoticeAutoReadPolicyEnum
|
|
from zerver.views.auth import redirect_to_deactivation_notice
|
|
from zproject.backends import check_password_strength, email_belongs_to_ldap
|
|
|
|
AVATAR_CHANGES_DISABLED_ERROR = gettext_lazy("Avatar changes are disabled in this organization.")
|
|
|
|
|
|
def validate_email_change_request(user_profile: UserProfile, new_email: str) -> None:
|
|
if not user_profile.is_active:
|
|
raise UserDeactivatedError
|
|
|
|
if user_profile.realm.email_changes_disabled and not user_profile.is_realm_admin:
|
|
raise JsonableError(_("Email address changes are disabled in this organization."))
|
|
|
|
error = validate_email_is_valid(
|
|
new_email,
|
|
get_realm_email_validator(user_profile.realm),
|
|
)
|
|
if error:
|
|
raise JsonableError(error)
|
|
|
|
try:
|
|
validate_email_not_already_in_realm(
|
|
user_profile.realm,
|
|
new_email,
|
|
allow_inactive_mirror_dummies=False,
|
|
verbose=False,
|
|
)
|
|
except ValidationError as e:
|
|
raise JsonableError(e.message)
|
|
|
|
|
|
def confirm_email_change_get(request: HttpRequest, confirmation_key: str) -> HttpResponse:
|
|
try:
|
|
get_object_from_key(confirmation_key, [Confirmation.EMAIL_CHANGE], mark_object_used=False)
|
|
except ConfirmationKeyError as exception: # nocoverage
|
|
return render_confirmation_key_error(request, exception)
|
|
|
|
return render(
|
|
request,
|
|
"confirmation/redirect_to_post.html",
|
|
context={
|
|
"target_url": reverse("confirm_email_change"),
|
|
"key": confirmation_key,
|
|
},
|
|
)
|
|
|
|
|
|
@require_post
|
|
@typed_endpoint
|
|
def confirm_email_change(request: HttpRequest, *, key: str) -> HttpResponse:
|
|
with transaction.atomic(durable=True):
|
|
try:
|
|
email_change_object = get_object_from_key(
|
|
key, [Confirmation.EMAIL_CHANGE], mark_object_used=True
|
|
)
|
|
except ConfirmationKeyError as exception:
|
|
return render_confirmation_key_error(request, exception)
|
|
|
|
assert isinstance(email_change_object, EmailChangeStatus)
|
|
new_email = email_change_object.new_email
|
|
old_email = email_change_object.old_email
|
|
|
|
user_profile = UserProfile.objects.select_for_update().get(
|
|
id=email_change_object.user_profile_id
|
|
)
|
|
|
|
if user_profile.delivery_email != old_email:
|
|
# This is not expected to be possible, since we deactivate
|
|
# any previous email changes when we create a new one, but
|
|
# double-check.
|
|
return render_confirmation_key_error(
|
|
request, ConfirmationKeyError(ConfirmationKeyError.EXPIRED)
|
|
) # nocoverage
|
|
|
|
if user_profile.realm.deactivated:
|
|
return redirect_to_deactivation_notice()
|
|
try:
|
|
validate_email_change_request(user_profile, new_email)
|
|
except UserDeactivatedError:
|
|
context = {"realm_url": user_profile.realm.url}
|
|
return render(
|
|
request,
|
|
"zerver/portico_error_pages/user_deactivated.html",
|
|
context=context,
|
|
status=401,
|
|
)
|
|
do_change_user_delivery_email(user_profile, new_email, acting_user=user_profile)
|
|
|
|
user_profile = UserProfile.objects.get(id=email_change_object.user_profile_id)
|
|
context = {"realm_name": user_profile.realm.name, "new_email": new_email}
|
|
language = user_profile.default_language
|
|
|
|
if old_email == "":
|
|
# The assertions here are to help document the only circumstance under which
|
|
# this condition should be possible.
|
|
assert (
|
|
user_profile.realm.demo_organization_scheduled_deletion_date is not None
|
|
and user_profile.is_realm_owner
|
|
)
|
|
# Schedule onboarding emails for demo organization owner now that we have an
|
|
# email address for their account.
|
|
enqueue_welcome_emails(user_profile, demo_organization_creator=True)
|
|
# Because demo organizations are created without setting an email and password
|
|
# we want to redirect to setting a password after configuring and confirming
|
|
# an email for the owner's account.
|
|
reset_password_url = generate_password_reset_url(user_profile, default_token_generator)
|
|
return HttpResponseRedirect(reset_password_url)
|
|
|
|
send_email(
|
|
"zerver/emails/notify_change_in_email",
|
|
to_emails=[old_email],
|
|
from_name=FromAddress.security_email_from_name(user_profile=user_profile),
|
|
from_address=FromAddress.SUPPORT,
|
|
language=language,
|
|
context=context,
|
|
realm=user_profile.realm,
|
|
)
|
|
old_email_address = Address(addr_spec=old_email)
|
|
new_email_address = Address(addr_spec=new_email)
|
|
ctx = {
|
|
"new_email_html_tag": SafeString(
|
|
f'<a href="mailto:{escape(new_email)}">{escape(new_email_address.username)}@<wbr>{escape(new_email_address.domain)}</wbr></a>'
|
|
),
|
|
"old_email_html_tag": SafeString(
|
|
f'<a href="mailto:{escape(old_email)}">{escape(old_email_address.username)}@<wbr>{escape(old_email_address.domain)}</wbr></a>'
|
|
),
|
|
}
|
|
return render(request, "confirmation/confirm_email_change.html", context=ctx)
|
|
|
|
|
|
emojiset_choices = {emojiset["key"] for emojiset in UserProfile.emojiset_choices()}
|
|
web_home_view_options = ["recent_topics", "inbox", "all_messages"]
|
|
web_animate_image_previews_options = ["always", "on_hover", "never"]
|
|
|
|
|
|
def check_settings_values(
|
|
notification_sound: str | None,
|
|
email_notifications_batching_period_seconds: int | None,
|
|
default_language: str | None = None,
|
|
) -> None:
|
|
# We can't use typed_endpoint for this widget because
|
|
# get_available_language_codes requires provisioning to be
|
|
# complete.
|
|
if default_language is not None and default_language not in get_available_language_codes():
|
|
raise JsonableError(_("Invalid default_language"))
|
|
|
|
if (
|
|
notification_sound is not None
|
|
and notification_sound not in get_available_notification_sounds()
|
|
and notification_sound != "none"
|
|
):
|
|
raise JsonableError(
|
|
_("Invalid notification sound '{notification_sound}'").format(
|
|
notification_sound=notification_sound
|
|
)
|
|
)
|
|
|
|
if email_notifications_batching_period_seconds is not None and (
|
|
email_notifications_batching_period_seconds <= 0
|
|
or email_notifications_batching_period_seconds > 7 * 24 * 60 * 60
|
|
):
|
|
# We set a limit of one week for the batching period
|
|
raise JsonableError(
|
|
_("Invalid email batching period: {seconds} seconds").format(
|
|
seconds=email_notifications_batching_period_seconds
|
|
)
|
|
)
|
|
|
|
|
|
@human_users_only
|
|
@typed_endpoint
|
|
def json_change_settings(
|
|
request: HttpRequest,
|
|
user_profile: UserProfile,
|
|
*,
|
|
allow_private_data_export: Json[bool] | None = None,
|
|
automatically_follow_topics_policy: Annotated[
|
|
Json[int],
|
|
check_int_in_validator(UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_CHOICES),
|
|
]
|
|
| None = None,
|
|
automatically_follow_topics_where_mentioned: Json[bool] | None = None,
|
|
automatically_unmute_topics_in_muted_streams_policy: Annotated[
|
|
Json[int],
|
|
check_int_in_validator(UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_CHOICES),
|
|
]
|
|
| None = None,
|
|
color_scheme: Annotated[Json[int], check_int_in_validator(UserProfile.COLOR_SCHEME_CHOICES)]
|
|
| None = None,
|
|
default_language: str | None = None,
|
|
demote_inactive_streams: Annotated[
|
|
Json[int], check_int_in_validator(UserProfile.DEMOTE_STREAMS_CHOICES)
|
|
]
|
|
| None = None,
|
|
desktop_icon_count_display: Annotated[
|
|
Json[int], check_int_in_validator(UserProfile.DESKTOP_ICON_COUNT_DISPLAY_CHOICES)
|
|
]
|
|
| None = None,
|
|
display_emoji_reaction_users: Json[bool] | None = None,
|
|
email: str | None = None,
|
|
email_address_visibility: Annotated[
|
|
Json[int], check_int_in_validator(UserProfile.EMAIL_ADDRESS_VISIBILITY_TYPES)
|
|
]
|
|
| None = None,
|
|
email_notifications_batching_period_seconds: Json[int] | None = None,
|
|
emojiset: Annotated[str, check_string_in_validator(emojiset_choices)] | None = None,
|
|
enable_desktop_notifications: Json[bool] | None = None,
|
|
enable_digest_emails: Json[bool] | None = None,
|
|
enable_drafts_synchronization: Json[bool] | None = None,
|
|
enable_followed_topic_audible_notifications: Json[bool] | None = None,
|
|
enable_followed_topic_desktop_notifications: Json[bool] | None = None,
|
|
enable_followed_topic_email_notifications: Json[bool] | None = None,
|
|
enable_followed_topic_push_notifications: Json[bool] | None = None,
|
|
enable_followed_topic_wildcard_mentions_notify: Json[bool] | None = None,
|
|
enable_login_emails: Json[bool] | None = None,
|
|
enable_marketing_emails: Json[bool] | None = None,
|
|
enable_offline_email_notifications: Json[bool] | None = None,
|
|
enable_offline_push_notifications: Json[bool] | None = None,
|
|
enable_online_push_notifications: Json[bool] | None = None,
|
|
enable_sounds: Json[bool] | None = None,
|
|
enable_stream_audible_notifications: Json[bool] | None = None,
|
|
enable_stream_desktop_notifications: Json[bool] | None = None,
|
|
enable_stream_email_notifications: Json[bool] | None = None,
|
|
enable_stream_push_notifications: Json[bool] | None = None,
|
|
enter_sends: Json[bool] | None = None,
|
|
fluid_layout_width: Json[bool] | None = None,
|
|
full_name: str | None = None,
|
|
high_contrast_mode: Json[bool] | None = None,
|
|
hide_ai_features: Json[bool] | None = None,
|
|
left_side_userlist: Json[bool] | None = None,
|
|
message_content_in_email_notifications: Json[bool] | None = None,
|
|
new_password: str | None = None,
|
|
notification_sound: str | None = None,
|
|
old_password: str | None = None,
|
|
pm_content_in_desktop_notifications: Json[bool] | None = None,
|
|
presence_enabled: Json[bool] | None = None,
|
|
realm_name_in_email_notifications_policy: Annotated[
|
|
Json[int],
|
|
check_int_in_validator(UserProfile.REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_CHOICES),
|
|
]
|
|
| None = None,
|
|
receives_typing_notifications: Json[bool] | None = None,
|
|
resolved_topic_notice_auto_read_policy: Annotated[
|
|
str | None,
|
|
AfterValidator(
|
|
lambda val: parse_enum_from_string_value(
|
|
val,
|
|
"resolved_topic_notice_auto_read_policy",
|
|
ResolvedTopicNoticeAutoReadPolicyEnum,
|
|
)
|
|
),
|
|
] = None,
|
|
send_private_typing_notifications: Json[bool] | None = None,
|
|
send_read_receipts: Json[bool] | None = None,
|
|
send_stream_typing_notifications: Json[bool] | None = None,
|
|
starred_message_counts: Json[bool] | None = None,
|
|
timezone: Annotated[str, timezone_validator()] | None = None,
|
|
translate_emoticons: Json[bool] | None = None,
|
|
twenty_four_hour_time: Json[bool] | None = None,
|
|
user_list_style: Annotated[
|
|
Json[int], check_int_in_validator(UserProfile.USER_LIST_STYLE_CHOICES)
|
|
]
|
|
| None = None,
|
|
web_animate_image_previews: Annotated[
|
|
str, check_string_in_validator(web_animate_image_previews_options)
|
|
]
|
|
| None = None,
|
|
web_channel_default_view: Annotated[
|
|
Json[int], check_int_in_validator(UserProfile.WEB_CHANNEL_DEFAULT_VIEW_CHOICES)
|
|
]
|
|
| None = None,
|
|
web_escape_navigates_to_home_view: Json[bool] | None = None,
|
|
web_font_size_px: Json[int] | None = None,
|
|
web_home_view: Annotated[str, check_string_in_validator(web_home_view_options)] | None = None,
|
|
web_left_sidebar_unreads_count_summary: Json[bool] | None = None,
|
|
web_line_height_percent: Json[int] | None = None,
|
|
web_mark_read_on_scroll_policy: Annotated[
|
|
Json[int], check_int_in_validator(UserProfile.WEB_MARK_READ_ON_SCROLL_POLICY_CHOICES)
|
|
]
|
|
| None = None,
|
|
web_navigate_to_sent_message: Json[bool] | None = None,
|
|
web_stream_unreads_count_display_policy: Annotated[
|
|
Json[int],
|
|
check_int_in_validator(UserProfile.WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_CHOICES),
|
|
]
|
|
| None = None,
|
|
web_suggest_update_timezone: Json[bool] | None = None,
|
|
wildcard_mentions_notify: Json[bool] | None = None,
|
|
) -> HttpResponse:
|
|
# UserProfile object is being refetched here to make sure that we
|
|
# do not use stale object from cache which can happen when a
|
|
# previous request tried updating multiple settings in a single
|
|
# request.
|
|
#
|
|
# TODO: Change the cache flushing strategy to make sure cache
|
|
# does not contain stale objects.
|
|
user_profile = UserProfile.objects.get(id=user_profile.id)
|
|
if (
|
|
default_language is not None
|
|
or notification_sound is not None
|
|
or email_notifications_batching_period_seconds is not None
|
|
):
|
|
check_settings_values(
|
|
notification_sound, email_notifications_batching_period_seconds, default_language
|
|
)
|
|
|
|
if new_password is not None:
|
|
return_data: dict[str, Any] = {}
|
|
if email_belongs_to_ldap(user_profile.realm, user_profile.delivery_email):
|
|
raise JsonableError(_("Your Zulip password is managed in LDAP"))
|
|
|
|
try:
|
|
if not authenticate(
|
|
request,
|
|
username=user_profile.delivery_email,
|
|
password=old_password,
|
|
realm=user_profile.realm,
|
|
return_data=return_data,
|
|
):
|
|
raise JsonableError(_("Wrong password!"))
|
|
except RateLimitedError as e:
|
|
assert e.secs_to_freedom is not None
|
|
secs_to_freedom = int(e.secs_to_freedom)
|
|
raise JsonableError(
|
|
_("You're making too many attempts! Try again in {seconds} seconds.").format(
|
|
seconds=secs_to_freedom
|
|
),
|
|
)
|
|
|
|
if not check_password_strength(new_password):
|
|
raise JsonableError(_("New password is too weak!"))
|
|
|
|
do_change_password(user_profile, new_password)
|
|
# Password changes invalidates sessions, see
|
|
# https://docs.djangoproject.com/en/5.0/topics/auth/default/#session-invalidation-on-password-change
|
|
# for details. To avoid this logging the user out of their own
|
|
# session (which would provide a confusing UX at best), we
|
|
# update the session hash here.
|
|
update_session_auth_hash(request, user_profile)
|
|
# We also save the session to the DB immediately to mitigate
|
|
# race conditions. In theory, there is still a race condition
|
|
# and to completely avoid it we will have to use some kind of
|
|
# mutex lock in `django.contrib.auth.get_user` where session
|
|
# is verified. To make that lock work we will have to control
|
|
# the AuthenticationMiddleware which is currently controlled
|
|
# by Django,
|
|
request.session.save()
|
|
|
|
result: dict[str, Any] = {}
|
|
|
|
if email is not None:
|
|
new_email = email.strip()
|
|
if user_profile.delivery_email != new_email:
|
|
validate_email_change_request(user_profile, new_email)
|
|
|
|
ratelimited, time_until_free = RateLimitedUser(
|
|
user_profile, domain="email_change_by_user"
|
|
).rate_limit()
|
|
if ratelimited:
|
|
raise RateLimitedError(time_until_free)
|
|
|
|
do_start_email_change_process(user_profile, new_email)
|
|
|
|
if full_name is not None and user_profile.full_name != full_name:
|
|
if name_changes_disabled(user_profile.realm) and not user_profile.is_realm_admin:
|
|
# Failingly silently is fine -- they can't do it through the UI, so
|
|
# they'd have to be trying to break the rules.
|
|
pass
|
|
else:
|
|
# Note that check_change_full_name strips the passed name automatically
|
|
check_change_full_name(user_profile, full_name, user_profile)
|
|
|
|
# Loop over user_profile.property_types
|
|
request_settings = {k: v for k, v in locals().items() if k in user_profile.property_types}
|
|
for k, v in request_settings.items():
|
|
if v is not None and getattr(user_profile, k) != v:
|
|
do_change_user_setting(user_profile, k, v, acting_user=user_profile)
|
|
|
|
if timezone is not None and user_profile.timezone != timezone:
|
|
do_change_user_setting(user_profile, "timezone", timezone, acting_user=user_profile)
|
|
|
|
return json_success(request, data=result)
|
|
|
|
|
|
def set_avatar_backend(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
|
|
if len(request.FILES) != 1:
|
|
raise JsonableError(_("You must upload exactly one avatar."))
|
|
|
|
if avatar_changes_disabled(user_profile.realm) and not user_profile.is_realm_admin:
|
|
raise JsonableError(str(AVATAR_CHANGES_DISABLED_ERROR))
|
|
|
|
[user_file] = request.FILES.values()
|
|
assert isinstance(user_file, UploadedFile)
|
|
assert user_file.size is not None
|
|
if user_file.size > settings.MAX_AVATAR_FILE_SIZE_MIB * 1024 * 1024:
|
|
raise JsonableError(
|
|
_("Uploaded file is larger than the allowed limit of {max_size} MiB").format(
|
|
max_size=settings.MAX_AVATAR_FILE_SIZE_MIB,
|
|
)
|
|
)
|
|
upload_avatar_image(user_file, user_profile)
|
|
do_change_avatar_fields(user_profile, UserProfile.AVATAR_FROM_USER, acting_user=user_profile)
|
|
user_avatar_url = avatar_url(user_profile)
|
|
|
|
json_result = dict(
|
|
avatar_url=user_avatar_url,
|
|
)
|
|
return json_success(request, data=json_result)
|
|
|
|
|
|
def delete_avatar_backend(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
|
|
if avatar_changes_disabled(user_profile.realm) and not user_profile.is_realm_admin:
|
|
raise JsonableError(str(AVATAR_CHANGES_DISABLED_ERROR))
|
|
|
|
do_change_avatar_fields(
|
|
user_profile, UserProfile.AVATAR_FROM_GRAVATAR, acting_user=user_profile
|
|
)
|
|
gravatar_url = avatar_url(user_profile)
|
|
|
|
json_result = dict(
|
|
avatar_url=gravatar_url,
|
|
)
|
|
return json_success(request, data=json_result)
|
|
|
|
|
|
# We don't use @human_users_only here, because there are use cases for
|
|
# a bot regenerating its own API key.
|
|
@typed_endpoint_without_parameters
|
|
def regenerate_api_key(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
|
|
new_api_key = do_regenerate_api_key(user_profile, user_profile)
|
|
json_result = dict(
|
|
api_key=new_api_key,
|
|
)
|
|
return json_success(request, data=json_result)
|