users: Factor out do_send_password_reset_email.

zerver/actions/users.py

@@ -4,8 +4,13 @@ from email.headerregistry import Address
 from typing import Any
 
 from django.conf import settings
+from django.contrib.auth.tokens import PasswordResetTokenGenerator, default_token_generator
 from django.db import transaction
+from django.http import HttpRequest
+from django.urls import reverse
+from django.utils.http import urlsafe_base64_encode
 from django.utils.timezone import now as timezone_now
+from django.utils.translation import get_language
 
 from zerver.actions.user_groups import (
     do_send_user_group_members_update_event,
@@ -17,8 +22,9 @@ from zerver.lib.cache import bot_dict_fields
 from zerver.lib.create_user import create_user
 from zerver.lib.invites import revoke_invites_generated_by_user
 from zerver.lib.remote_server import maybe_enqueue_audit_log_upload
-from zerver.lib.send_email import clear_scheduled_emails
+from zerver.lib.send_email import FromAddress, clear_scheduled_emails, send_email
 from zerver.lib.sessions import delete_user_sessions
+from zerver.lib.soft_deactivation import queue_soft_reactivation
 from zerver.lib.stream_subscription import bulk_get_subscriber_peer_info
 from zerver.lib.stream_traffic import get_streams_traffic
 from zerver.lib.streams import get_streams_for_user, stream_to_dict
@@ -693,3 +699,63 @@ def get_owned_bot_dicts(
         }
         for botdict in result
     ]
+
+
+def generate_password_reset_url(
+    user_profile: UserProfile, token_generator: PasswordResetTokenGenerator
+) -> str:
+    token = token_generator.make_token(user_profile)
+    uid = urlsafe_base64_encode(str(user_profile.id).encode())
+    endpoint = reverse("password_reset_confirm", kwargs=dict(uidb64=uid, token=token))
+    return f"{user_profile.realm.url}{endpoint}"
+
+
+def do_send_password_reset_email(
+    email: str,
+    realm: Realm,
+    user_profile: UserProfile | None,
+    *,
+    token_generator: PasswordResetTokenGenerator = default_token_generator,
+    request: HttpRequest | None = None,
+) -> None:
+    context: dict[str, object] = {
+        "email": email,
+        "realm_url": realm.url,
+        "realm_name": realm.name,
+    }
+    if user_profile is not None and not user_profile.is_active:
+        context["user_deactivated"] = True
+        user_profile = None
+
+    if user_profile is not None:
+        queue_soft_reactivation(user_profile.id)
+        context["active_account_in_realm"] = True
+        context["reset_url"] = generate_password_reset_url(user_profile, token_generator)
+        send_email(
+            "zerver/emails/password_reset",
+            to_user_ids=[user_profile.id],
+            from_name=FromAddress.security_email_from_name(user_profile=user_profile),
+            from_address=FromAddress.tokenized_no_reply_address(),
+            context=context,
+            realm=realm,
+            request=request,
+        )
+    else:
+        context["active_account_in_realm"] = False
+        active_accounts_in_other_realms = UserProfile.objects.filter(
+            delivery_email__iexact=email, is_active=True
+        )
+        if active_accounts_in_other_realms:
+            context["active_accounts_in_other_realms"] = active_accounts_in_other_realms
+        language = get_language()
+
+        send_email(
+            "zerver/emails/password_reset",
+            to_emails=[email],
+            from_name=FromAddress.security_email_from_name(language=language),
+            from_address=FromAddress.tokenized_no_reply_address(),
+            language=language,
+            context=context,
+            realm=realm,
+            request=request,
+        )

zerver/forms.py

@@ -12,16 +12,15 @@ from django.contrib.auth.tokens import PasswordResetTokenGenerator, default_toke
 from django.core.exceptions import ValidationError
 from django.core.validators import validate_email
 from django.http import HttpRequest
-from django.urls import reverse
-from django.utils.http import urlsafe_base64_encode
-from django.utils.translation import get_language, gettext_lazy
 from django.utils.translation import gettext as _
+from django.utils.translation import gettext_lazy
 from markupsafe import Markup
 from two_factor.forms import AuthenticationTokenForm as TwoFactorAuthenticationTokenForm
 from two_factor.utils import totp_digits
 from typing_extensions import override
 
 from zerver.actions.user_settings import do_change_password
+from zerver.actions.users import do_send_password_reset_email
 from zerver.lib.email_validation import (
     email_allowed_for_realm,
     email_reserved_for_system_bots_error,
@@ -31,8 +30,6 @@ from zerver.lib.exceptions import JsonableError, RateLimitedError
 from zerver.lib.i18n import get_language_list
 from zerver.lib.name_restrictions import is_reserved_subdomain
 from zerver.lib.rate_limiter import RateLimitedObject, rate_limit_request_by_ip
-from zerver.lib.send_email import FromAddress, send_email
-from zerver.lib.soft_deactivation import queue_soft_reactivation
 from zerver.lib.subdomains import get_subdomain, is_root_domain_available
 from zerver.lib.users import check_full_name
 from zerver.models import Realm, UserProfile
@@ -357,15 +354,6 @@ class LoggingSetPasswordForm(SetPasswordForm):
         return self.user
 
 
-def generate_password_reset_url(
-    user_profile: UserProfile, token_generator: PasswordResetTokenGenerator
-) -> str:
-    token = token_generator.make_token(user_profile)
-    uid = urlsafe_base64_encode(str(user_profile.id).encode())
-    endpoint = reverse("password_reset_confirm", kwargs=dict(uidb64=uid, token=token))
-    return f"{user_profile.realm.url}{endpoint}"
-
-
 class ZulipPasswordResetForm(PasswordResetForm):
     @override
     def save(
@@ -431,48 +419,9 @@ class ZulipPasswordResetForm(PasswordResetForm):
         except UserProfile.DoesNotExist:
             user = None
 
-        context = {
-            "email": email,
-            "realm_url": realm.url,
-            "realm_name": realm.name,
-        }
-
-        if user is not None and not user.is_active:
-            context["user_deactivated"] = True
-            user = None
-
-        if user is not None:
-            context["active_account_in_realm"] = True
-            context["reset_url"] = generate_password_reset_url(user, token_generator)
-            queue_soft_reactivation(user.id)
-            send_email(
-                "zerver/emails/password_reset",
-                to_user_ids=[user.id],
-                from_name=FromAddress.security_email_from_name(user_profile=user),
-                from_address=FromAddress.tokenized_no_reply_address(),
-                context=context,
-                realm=realm,
-                request=request,
-            )
-        else:
-            context["active_account_in_realm"] = False
-            active_accounts_in_other_realms = UserProfile.objects.filter(
-                delivery_email__iexact=email, is_active=True
-            )
-            if active_accounts_in_other_realms:
-                context["active_accounts_in_other_realms"] = active_accounts_in_other_realms
-            language = get_language()
-
-            send_email(
-                "zerver/emails/password_reset",
-                to_emails=[email],
-                from_name=FromAddress.security_email_from_name(language=language),
-                from_address=FromAddress.tokenized_no_reply_address(),
-                language=language,
-                context=context,
-                realm=realm,
-                request=request,
-            )
+        do_send_password_reset_email(
+            email, realm, user, token_generator=token_generator, request=request
+        )
 
 
 class RateLimitedPasswordResetByEmail(RateLimitedObject):

zerver/management/commands/send_password_reset_email.py

@@ -1,14 +1,12 @@
 from argparse import ArgumentParser
 from typing import Any
 
-from django.contrib.auth.tokens import default_token_generator
 from django.core.management.base import CommandError
 from django.db.models import QuerySet
 from typing_extensions import override
 
-from zerver.forms import generate_password_reset_url
+from zerver.actions.users import do_send_password_reset_email
 from zerver.lib.management import ZulipBaseCommand
-from zerver.lib.send_email import FromAddress, send_email
 from zerver.models import UserProfile
 
 
@@ -59,17 +57,6 @@ class Command(ZulipBaseCommand):
     def send(self, users: QuerySet[UserProfile]) -> None:
         """Sends one-use only links for resetting password to target users"""
         for user_profile in users:
-            context = {
-                "email": user_profile.delivery_email,
-                "reset_url": generate_password_reset_url(user_profile, default_token_generator),
-                "realm_url": user_profile.realm.url,
-                "realm_name": user_profile.realm.name,
-                "active_account_in_realm": True,
-            }
-            send_email(
-                "zerver/emails/password_reset",
-                to_user_ids=[user_profile.id],
-                from_address=FromAddress.tokenized_no_reply_address(),
-                from_name=FromAddress.security_email_from_name(user_profile=user_profile),
-                context=context,
+            do_send_password_reset_email(
+                user_profile.delivery_email, user_profile.realm, user_profile
             )

zerver/views/user_settings.py

@@ -30,8 +30,8 @@ from zerver.actions.user_settings import (
     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
-from zerver.forms import generate_password_reset_url
 from zerver.lib.avatar import avatar_url
 from zerver.lib.email_validation import (
     get_realm_email_validator,