fix(auth): do not update email in UserDetailsEndpoint

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

@@ -36,7 +36,6 @@
 from sentry.users.api.serializers.user import DetailedSelfUserSerializer
 from sentry.users.models.user import User
 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 get_timezone_choices
 
@@ -160,27 +159,10 @@ class BaseUserSerializer(CamelSnakeModelSerializer[User]):
     def validate_username(self, value: str) -> str:
         assert isinstance(self.instance, User), "Should be a single record not a sequence"
 
-        if (
-            User.objects.filter(username__iexact=value)
-            # Django throws an exception if `id` is `None`, which it will be when we're importing
-            # new users via the relocation logic on the `User` model. So we cast `None` to `0` to
-            # make Django happy here.
-            .exclude(id=self.instance.id if hasattr(self.instance, "id") else 0)
-            .exists()
-        ):
+        if not User.is_username_available(value, exclude_user_id=self.instance.id or 0):
             raise serializers.ValidationError("That username is already in use.")
-        return value
-
-    def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
-        attrs = super().validate(attrs)
 
-        assert isinstance(self.instance, User), "Should be a single record not a sequence"
-        if self.instance.email == self.instance.username:
-            if attrs.get("username", self.instance.email) != self.instance.email:
-                # ... this probably needs to handle newsletters and such?
-                attrs.setdefault("email", attrs["username"])
-
-        return attrs
+        return value
 
     def update(self, instance: User, validated_data: dict[str, Any]) -> User:
         if "isActive" not in validated_data:
@@ -319,17 +301,8 @@ def put(self, request: Request, user: User) -> Response:
         :param string default_issue_event: Event displayed by default, "recommended", "latest" or "oldest"
         :auth: required
         """
-        email = None
-        if "email" in request.data and len(request.data["email"]) > 0:
-            email = request.data["email"]
-        elif "username" in request.data and len(request.data["username"]) > 0:
-            email = request.data["username"]
-        if email:
-            verified_email_found = UserEmail.objects.filter(
-                user_id=user.id, email=email, is_verified=True
-            ).exists()
-            if not verified_email_found:
-                return Response({"detail": "Verified email address is not found."}, status=400)
+        # `email` is not writable here: primary email changes go through UserEmailsEndpoint
+        # `username` is only editable if it does not match `email`, changing `username` does not change `email`.
 
         # We want to prevent superusers from setting users to superuser or staff
         # because this is only done through _admin. This will always be enforced

src/sentry/users/models/user.py

@@ -10,13 +10,14 @@
 from django.contrib.auth.models import UserManager as DjangoUserManager
 from django.contrib.auth.signals import user_logged_out
 from django.db import IntegrityError, models, router, transaction
-from django.db.models import Count, Subquery
+from django.db.models import Count, Q, Subquery
 from django.db.models.query import QuerySet
 from django.dispatch import receiver
 from django.forms import model_to_dict
 from django.http.request import HttpRequest
 from django.urls import reverse
 from django.utils import timezone
+from django.utils.crypto import get_random_string
 from django.utils.translation import gettext_lazy as _
 
 from bitfield import TypedClassBitField
@@ -34,7 +35,6 @@
 from sentry.db.models import Model, control_silo_model, sane_repr
 from sentry.db.models.manager.base import BaseManager
 from sentry.db.models.manager.base_query_set import BaseQuerySet
-from sentry.db.models.utils import unique_db_instance
 from sentry.db.postgres.transactions import enforce_constraints
 from sentry.hybridcloud.models.outbox import ControlOutboxBase, outbox_context
 from sentry.hybridcloud.outbox.category import OutboxCategory
@@ -482,8 +482,6 @@ def set_password(self, raw_password: str | None) -> None:
         self.is_password_expired = False
 
     def refresh_session_nonce(self, request: HttpRequest | None = None) -> None:
-        from django.utils.crypto import get_random_string
-
         self.session_nonce = get_random_string(12)
         if request is not None:
             request.session["_nonce"] = self.session_nonce
@@ -537,6 +535,29 @@ def normalize_before_relocation_import(
 
         return old_pk
 
+    @classmethod
+    def is_username_available(cls, username: str, *, exclude_user_id: int = 0) -> bool:
+        """
+        A username is available only if it is not already in use as another account's
+        username, primary email, or verified email. Login resolves a username before
+        falling back to email (see find_users in sentry/utils/auth.py), so a username
+        equal to an email another account owns would collide with its login identity.
+
+        `exclude_user_id` is the user being updated (excluded from the check); leave it
+        as the default when checking a not-yet-saved user (e.g. a relocation import).
+        """
+        username_or_primary_taken = (
+            cls.objects.exclude(id=exclude_user_id)
+            .filter(Q(username__iexact=username) | Q(email__iexact=username))
+            .exists()
+        )
+        verified_email_taken = (
+            UserEmail.objects.filter(email__iexact=username, is_verified=True)
+            .exclude(user_id=exclude_user_id)
+            .exists()
+        )
+        return not (username_or_primary_taken or verified_email_taken)
+
     def write_relocation_import(
         self, scope: ImportScope, flags: ImportFlags
     ) -> tuple[int, ImportKind] | None:
@@ -551,7 +572,7 @@ def do_write() -> tuple[int, ImportKind]:
                 DatabaseLostPasswordHashService,
             )
 
-            serializer_cls = BaseUserSerializer
+            serializer_cls: type[BaseUserSerializer]
             if scope not in {ImportScope.Config, ImportScope.Global}:
                 serializer_cls = UserSerializer
             else:
@@ -570,29 +591,30 @@ def do_write() -> tuple[int, ImportKind]:
             # that actually goes, and how to prevent it from happening during the validation pass.
             return (self.pk, ImportKind.Inserted)
 
-        # If there is no existing user with this `username`, no special renaming or merging
-        # shenanigans are needed, as we can just insert this exact model directly.
-        existing = User.objects.filter(username=self.username).first()
-        if not existing:
-            return do_write()
-
-        # Re-use the existing user if merging is enabled.
+        # Merging is enabled only for SAAS_TO_SAAS relocations, where the data is
+        # server-generated and `username` is unique, so matching on username alone is safe.
+        # Reuse the existing account instead of inserting.
         if flags.merge_users:
-            return (existing.pk, ImportKind.Existing)
-
-        # We already have a user with this `username`, but merging users has not been enabled. In
-        # this case, add a random suffix to the importing username.
-        lock = locks.get(f"user:username:{self.id}", duration=10, name="username")
-        with TimedRetryPolicy(10)(lock.acquire):
-            unique_db_instance(
-                self,
-                self.username,
-                max_length=MAX_USERNAME_LENGTH,
-                field_name="username",
-            )
+            username_match = User.objects.filter(username__iexact=self.username).first()
+            if username_match:
+                return (username_match.pk, ImportKind.Existing)
+
+        # Suffix until the username collides with neither an existing username nor
+        # email; the standard slug helper (unique_db_instance) does not support cross-column checking.
+        if not User.is_username_available(self.username):
+            lock = locks.get(f"user:username:{self.id}", duration=10, name="username")
+            with TimedRetryPolicy(10)(lock.acquire):
+                base = self.username[: MAX_USERNAME_LENGTH - 13]
+                for _ in range(3):
+                    suffix = get_random_string(
+                        12, allowed_chars="abcdefghijklmnopqrstuvwxyz0123456789"
+                    )
+                    self.username = f"{base}-{suffix}"
+                    if User.is_username_available(self.username):
+                        return do_write()
+                raise RuntimeError("Could not generate a unique username during relocation import")
 
-            # Perform the remainder of the write while we're still holding the lock.
-            return do_write()
+        return do_write()
 
     @classmethod
     def sanitize_relocation_json(

src/sentry/users/web/accounts_form.py

@@ -121,7 +121,7 @@ def clean_username(self) -> str | None:
         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():
+        if not User.is_username_available(value, exclude_user_id=self.user.id):
             raise forms.ValidationError(_("An account is already registered with that username."))
         return value.lower()
 

tests/sentry/backup/test_imports.py

@@ -354,6 +354,23 @@ def test_generate_suffix_for_already_taken_username(self) -> None:
                 assert User.objects.filter(username__iexact="min_user").count() == 1
                 assert User.objects.filter(username__icontains="min_user-").count() == 2
 
+    def test_generate_suffix_when_username_matches_existing_email(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            self.create_user(email="min_user", username="existing_sso", is_test_user=False)
+            tmp_path = Path(tmp_dir).joinpath(f"{self._testMethodName}.json")
+            with open(tmp_path, "wb+") as tmp_file:
+                # creates a User with username = "min_user"
+                models = self.json_of_exhaustive_user_with_minimum_privileges()
+                tmp_file.write(orjson.dumps(self.sort_in_memory_json(models)))
+
+            with open(tmp_path, "rb") as tmp_file:
+                import_in_user_scope(tmp_file, printer=NOOP_PRINTER)
+
+            with assume_test_silo_mode(SiloMode.CONTROL):
+                assert User.objects.filter(username__iexact="min_user").count() == 0
+                assert User.objects.filter(username__icontains="min_user-").count() == 1
+                assert User.objects.filter(username="existing_sso").count() == 1
+
     def test_bad_invalid_user(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             tmp_path = Path(tmp_dir).joinpath(f"{self._testMethodName}.json")

tests/sentry/users/api/endpoints/test_user_details.py

@@ -173,44 +173,72 @@ def test_managed_fields(self) -> None:
             user = User.objects.get(id=self.user.id)
             assert user
 
-    def test_change_username_when_different(self) -> None:
-        # if email != username and we change username, only username should change
+    def test_change_username_with_different_email(self) -> None:
         user = self.create_user(email="c@example.com", username="diff@example.com")
         self.login_as(user=user, superuser=False)
 
-        self.create_useremail(user, "new@example.com", is_verified=True)
-        response = self.get_success_response("me", username="new@example.com")
+        response = self.get_success_response("me", username="john")
         user = User.objects.get(id=user.id)
 
         assert user.email == "c@example.com"
         assert response.data["email"] == "c@example.com"
-        assert user.username == "new@example.com"
+        assert user.username == "john"
 
-    def test_change_username_when_same(self) -> None:
-        # if email == username and we change username,
-        # keep email in sync
+    def test_change_username_with_same_email(self) -> None:
         user = self.create_user(email="c@example.com", username="c@example.com")
         self.login_as(user=user)
 
-        self.create_useremail(user, "new@example.com", is_verified=True)
         self.get_success_response("me", username="new@example.com")
 
         user = User.objects.get(id=user.id)
 
-        assert user.email == "new@example.com"
+        assert user.email == "c@example.com"
         assert user.username == "new@example.com"
 
-    def test_cannot_change_username_to_non_verified(self) -> None:
-        user = self.create_user(email="c@example.com", username="c@example.com")
+    def test_cannot_edit_email_through_this_endpoint(self) -> None:
+        user = self.create_user(email="c@example.com", username="diff@example.com")
         self.login_as(user=user)
+        self.create_useremail(user, "new@example.com", is_verified=True)
 
-        self.create_useremail(user, "new@example.com", is_verified=False)
-        resp = self.get_error_response("me", username="new@example.com", status_code=400)
-        assert resp.data["detail"] == "Verified email address is not found."
+        self.get_success_response("me", email="new@example.com")
 
         user = User.objects.get(id=user.id)
 
         assert user.email == "c@example.com"
+        assert user.username == "diff@example.com"
+
+    def test_cannot_take_another_users_email_as_username(self) -> None:
+        self.create_user(email="user_a@example.com", username="sso-abc123")
+        user_b = self.create_user(email="b_user@example.com", username="user_b@example.com")
+        self.login_as(user=user_b)
+
+        resp = self.get_error_response("me", username="user_a@example.com", status_code=400)
+        assert resp.data["username"] == ["That username is already in use."]
+
+        user_b = User.objects.get(id=user_b.id)
+        assert user_b.username == "user_b@example.com"
+
+    def test_change_username_to_own_secondary_email(self) -> None:
+        user = self.create_user(email="c@example.com", username="c@example.com")
+        self.create_useremail(user, "secondary@example.com", is_verified=True)
+        self.login_as(user=user)
+
+        self.get_success_response("me", username="secondary@example.com")
+
+        user = User.objects.get(id=user.id)
+        assert user.username == "secondary@example.com"
+        assert user.email == "c@example.com"
+
+    def test_cannot_take_another_users_verified_secondary_email_as_username(self) -> None:
+        other = self.create_user(email="other@example.com", username="other@example.com")
+        self.create_useremail(other, "shared@example.com", is_verified=True)
+        user = self.create_user(email="c@example.com", username="c@example.com")
+        self.login_as(user=user)
+
+        resp = self.get_error_response("me", username="shared@example.com", status_code=400)
+        assert resp.data["username"] == ["That username is already in use."]
+
+        user = User.objects.get(id=user.id)
         assert user.username == "c@example.com"
 
     @override_settings(SENTRY_MODE=SentryMode.SAAS)
@@ -599,9 +627,6 @@ def test_superuser_can_update_non_privileged_fields_without_permission(self) ->
         """Test that superuser can update non-privileged fields even without users.admin permission"""
         self.login_as(user=self.superuser, superuser=True)
 
-        # Create a verified email for the user before updating username
-        self.create_useremail(self.user, "newemail@example.com", is_verified=True)
-
         resp = self.get_success_response(
             self.user.id,
             name="New Name",

tests/sentry/users/web/test_accounts_form.py

@@ -1,3 +1,6 @@
+import pytest
+from django import forms
+
 from sentry.testutils.cases import TestCase
 from sentry.testutils.silo import control_silo_test
 from sentry.users.web.accounts_form import RelocationForm
@@ -36,6 +39,24 @@ def test_clean_username_forces_lowercase(self) -> None:
 
         assert relocation_form.clean_username() == "new_username"
 
+    def test_clean_username_rejects_existing_username(self) -> None:
+        self.create_user(username="taken_username")
+        user = self.create_user(username="test_user")
+        relocation_form = RelocationForm(user=user)
+        relocation_form.cleaned_data = {"username": "taken_username"}
+
+        with pytest.raises(forms.ValidationError):
+            relocation_form.clean_username()
+
+    def test_clean_username_rejects_another_users_email(self) -> None:
+        self.create_user(email="mail@example.com", username="sso-abc123")
+        user = self.create_user(username="test_user")
+        relocation_form = RelocationForm(user=user)
+        relocation_form.cleaned_data = {"username": "mail@example.com"}
+
+        with pytest.raises(forms.ValidationError):
+            relocation_form.clean_username()
+
     def test_clean_password(self) -> None:
         username = "test_user"
         user = self.create_user(username=username)