ref(users): move users related web resources to users directory

pyproject.toml

@@ -378,7 +378,6 @@ module = [
     "sentry.utils.committers",
     "sentry.utils.services",
     "sentry.web.forms.accounts",
-    "sentry.web.frontend.account_identity",
     "sentry.web.frontend.auth_close",
     "sentry.web.frontend.auth_login",
     "sentry.web.frontend.auth_logout",

src/sentry/middleware/auth.py

@@ -83,7 +83,7 @@ def process_exception(
         self, request: HttpRequest, exception: Exception
     ) -> HttpResponseBase | None:
         if isinstance(exception, AuthUserPasswordExpired):
-            from sentry.web.frontend.accounts import expired
+            from sentry.users.web.accounts import expired
 
             return expired(request, exception.user)
         else:

src/sentry/users/api/endpoints/user_details.py

@@ -1,6 +1,4 @@
 import logging
-import zoneinfo
-from datetime import datetime
 from typing import Any
 
 from django.conf import settings
@@ -31,27 +29,13 @@
 from sentry.users.models.user_option import UserOption
 from sentry.users.models.useremail import UserEmail
 from sentry.users.services.user.serial import serialize_generic_user
-from sentry.utils.dates import AVAILABLE_TIMEZONES
+from sentry.utils.dates import get_timezone_choices
 
 audit_logger = logging.getLogger("sentry.audit.user")
 delete_logger = logging.getLogger("sentry.deletions.api")
 
 
-def _get_timezone_choices() -> list[tuple[str, str]]:
-    build_results = []
-    for tz in AVAILABLE_TIMEZONES:
-        now = datetime.now(zoneinfo.ZoneInfo(tz))
-        offset = now.strftime("%z")
-        build_results.append((int(offset), tz, f"(UTC{offset}) {tz}"))
-    build_results.sort()
-
-    results: list[tuple[str, str]] = []
-    for item in build_results:
-        results.append(item[1:])
-    return results
-
-
-TIMEZONE_CHOICES = _get_timezone_choices()
+TIMEZONE_CHOICES = get_timezone_choices()
 
 
 class UserOptionsSerializer(serializers.Serializer[UserOption]):

src/sentry/web/frontend/account_identity.py → src/sentry/users/web/account_identity.py

@@ -1,10 +1,11 @@
+from django.http.request import HttpRequest
 from django.http.response import HttpResponseBase
 from django.urls import reverse
 from django.utils.decorators import method_decorator
 from django.views.decorators.cache import never_cache
-from rest_framework.request import Request
 
 from sentry.identity.pipeline import IdentityProviderPipeline
+from sentry.models.organization import Organization
 from sentry.users.models.identity import IdentityProvider
 from sentry.web.frontend.base import ControlSiloOrganizationView, control_silo_view
 from sentry.web.helpers import render_to_response
@@ -13,7 +14,9 @@
 @control_silo_view
 class AccountIdentityAssociateView(ControlSiloOrganizationView):
     @method_decorator(never_cache)
-    def handle(self, request: Request, organization, provider_key, external_id) -> HttpResponseBase:
+    def handle(
+        self, request: HttpRequest, organization: Organization, provider_key: str, external_id: str
+    ) -> HttpResponseBase:
         try:
             provider_model = IdentityProvider.objects.get(
                 type=provider_key, external_id=external_id

src/sentry/web/frontend/accounts.py → src/sentry/users/web/accounts.py

@@ -5,7 +5,7 @@
 from django.contrib.auth import authenticate
 from django.contrib.auth import login as login_user
 from django.db import router, transaction
-from django.http import HttpResponse, HttpResponseRedirect
+from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
 from django.urls import reverse
 from django.utils.translation import gettext as _
 from django.views.decorators.http import require_http_methods
@@ -21,9 +21,13 @@
 from sentry.users.models.useremail import UserEmail
 from sentry.users.services.lost_password_hash import lost_password_hash_service
 from sentry.users.services.user.service import user_service
+from sentry.users.web.accounts_form import (
+    ChangePasswordRecoverForm,
+    RecoverPasswordForm,
+    RelocationForm,
+)
 from sentry.utils import auth
 from sentry.web.decorators import login_required, set_referrer_policy
-from sentry.web.forms.accounts import ChangePasswordRecoverForm, RecoverPasswordForm, RelocationForm
 from sentry.web.helpers import render_to_response
 
 logger = logging.getLogger("sentry.accounts")
@@ -38,19 +42,19 @@ class InvalidRequest(Exception):
     pass
 
 
-def get_template(mode, name):
+def get_template(mode: str, name: str) -> str:
     return f"sentry/account/{mode}/{name}.html"
 
 
 @login_required
 @control_silo_function
-def login_redirect(request):
+def login_redirect(request: HttpRequest) -> HttpResponseRedirect:
     login_url = auth.get_login_redirect(request)
     return HttpResponseRedirect(login_url)
 
 
 @control_silo_function
-def expired(request, user):
+def expired(request: HttpRequest, user: User) -> HttpResponse:
     hash = lost_password_hash_service.get_or_create(user_id=user.id).hash
     LostPasswordHash.send_recover_password_email(user, hash, request.META["REMOTE_ADDR"])
 
@@ -59,7 +63,7 @@ def expired(request, user):
 
 
 @control_silo_function
-def recover(request):
+def recover(request: HttpRequest) -> HttpResponse:
     from sentry import ratelimits as ratelimiter
 
     extra = {
@@ -112,7 +116,7 @@ def recover(request):
 
 @set_referrer_policy("strict-origin-when-cross-origin")
 @control_silo_function
-def relocate_reclaim(request, user_id):
+def relocate_reclaim(request: HttpRequest, user_id: int) -> HttpResponse:
     """
     Ask to receive a new "claim this user" email.
     """
@@ -176,7 +180,9 @@ def relocate_reclaim(request, user_id):
 
 @set_referrer_policy("strict-origin-when-cross-origin")
 @control_silo_function
-def recover_confirm(request, user_id, hash, mode="recover"):
+def recover_confirm(
+    request: HttpRequest, user_id: int, hash: str, mode: str = "recover"
+) -> HttpResponse:
     from sentry import ratelimits as ratelimiter
 
     try:
@@ -285,7 +291,7 @@ def recover_confirm(request, user_id, hash, mode="recover"):
 @login_required
 @require_http_methods(["POST"])
 @control_silo_function
-def start_confirm_email(request):
+def start_confirm_email(request: HttpRequest) -> HttpResponse:
     from sentry import ratelimits as ratelimiter
 
     if ratelimiter.backend.is_limited(
@@ -299,6 +305,9 @@ def start_confirm_email(request):
             status=429,
         )
 
+    assert isinstance(
+        request.user, User
+    ), "User must have an associated email to send confirm emails to"
     if "primary-email" in request.POST:
         email = request.POST.get("email")
         try:
@@ -334,7 +343,7 @@ def start_confirm_email(request):
 @set_referrer_policy("strict-origin-when-cross-origin")
 @login_required
 @control_silo_function
-def confirm_email(request, user_id, hash):
+def confirm_email(request: HttpRequest, user_id: int, hash: str) -> HttpResponseRedirect:
     msg = _("Thanks for confirming your email")
     level = messages.SUCCESS
     try:

src/sentry/users/web/accounts_form.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from django import forms
+from django.conf import settings
+from django.utils.translation import gettext_lazy as _
+
+from sentry.auth import password_validation
+from sentry.users.models.user import User
+from sentry.utils.auth import find_users
+from sentry.utils.dates import get_timezone_choices
+from sentry.web.forms.fields import AllowedEmailField
+
+TIMEZONE_CHOICES = get_timezone_choices()
+
+
+class RecoverPasswordForm(forms.Form):
+    user = forms.CharField(
+        label=_("Account"),
+        max_length=128,
+        widget=forms.TextInput(attrs={"placeholder": _("username or email")}),
+    )
+
+    def clean_user(self) -> User | None:
+        value = (self.cleaned_data.get("user") or "").strip()
+        if not value:
+            return None
+        users = find_users(value, with_valid_password=False)
+        if not users:
+            return None
+
+        # If we find more than one user, we likely matched on email address.
+        # We silently bail here as we emailing the 'wrong' person isn't great.
+        # They will have to retry with their username which is guaranteed
+        # to be unique
+        if len(users) > 1:
+            return None
+
+        users = [u for u in users if not u.is_managed]
+        if not users:
+            raise forms.ValidationError(
+                _(
+                    "The account you are trying to recover is managed and does not support password recovery."
+                )
+            )
+        return users[0]
+
+
+class ChangePasswordRecoverForm(forms.Form):
+    password = forms.CharField(widget=forms.PasswordInput())
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.user = kwargs.pop("user", None)
+        super().__init__(*args, **kwargs)
+
+    def clean_password(self) -> str:
+        password = self.cleaned_data["password"]
+        password_validation.validate_password(password, user=self.user)
+        return password
+
+
+class EmailForm(forms.Form):
+    alt_email = AllowedEmailField(
+        label=_("New Email"),
+        required=False,
+        help_text="Designate an alternative email for this account",
+    )
+
+    password = forms.CharField(
+        label=_("Current password"),
+        widget=forms.PasswordInput(),
+        help_text=_("You will need to enter your current account password to make changes."),
+        required=True,
+    )
+
+    def __init__(self, user: User, *args: Any, **kwargs: Any) -> None:
+        self.user = user
+        super().__init__(*args, **kwargs)
+
+        needs_password = user.has_usable_password()
+
+        if not needs_password:
+            del self.fields["password"]
+
+    def clean_password(self) -> str:
+        value = self.cleaned_data.get("password")
+        if value and not self.user.check_password(value):
+            raise forms.ValidationError(_("The password you entered is not correct."))
+        elif not value:
+            raise forms.ValidationError(
+                _("You must confirm your current password to make changes.")
+            )
+        return value
+
+
+class RelocationForm(forms.Form):
+    username = forms.CharField(max_length=128, required=False, widget=forms.TextInput())
+    password = forms.CharField(widget=forms.PasswordInput())
+    tos_check = forms.BooleanField(
+        label=_(
+            f"I agree to the <a href={settings.TERMS_URL}>Terms of Service</a> and <a href={settings.PRIVACY_URL}>Privacy Policy</a>"
+        ),
+        widget=forms.CheckboxInput(),
+        required=False,
+        initial=False,
+    )
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.user = kwargs.pop("user", None)
+        super().__init__(*args, **kwargs)
+        self.fields["username"].widget.attrs.update(placeholder=self.user.username)
+
+    def clean_username(self) -> str | None:
+        value = self.cleaned_data.get("username") or self.user.username
+        value = re.sub(r"[ \n\t\r\0]*", "", value)
+        if not value:
+            return None
+        if User.objects.filter(username__iexact=value).exclude(id=self.user.id).exists():
+            raise forms.ValidationError(_("An account is already registered with that username."))
+        return value.lower()
+
+    def clean_password(self) -> str:
+        password = self.cleaned_data["password"]
+        password_validation.validate_password(password, user=self.user)
+        return password
+
+    def clean_tos_check(self) -> None:
+        value = self.cleaned_data.get("tos_check")
+        if not value:
+            raise forms.ValidationError(
+                _("You must agree to the Terms of Service and Privacy Policy before proceeding.")
+            )
+        return None

src/sentry/utils/dates.py

@@ -184,3 +184,17 @@ def outside_retention_with_modified_start(
     start = max(start, now - timedelta(days=retention))
 
     return start > end, start
+
+
+def get_timezone_choices() -> list[tuple[str, str]]:
+    build_results = []
+    for tz in AVAILABLE_TIMEZONES:
+        now = datetime.now(zoneinfo.ZoneInfo(tz))
+        offset = now.strftime("%z")
+        build_results.append((int(offset), tz, f"(UTC{offset}) {tz}"))
+    build_results.sort()
+
+    results: list[tuple[str, str]] = []
+    for item in build_results:
+        results.append(item[1:])
+    return results