diff --git a/src/sentry/users/api/endpoints/user_details.py b/src/sentry/users/api/endpoints/user_details.py index 4f455132b8ea..4d75a93d2c59 100644 --- a/src/sentry/users/api/endpoints/user_details.py +++ b/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 diff --git a/src/sentry/users/models/user.py b/src/sentry/users/models/user.py index 33c5570f084e..ccdecee947ce 100644 --- a/src/sentry/users/models/user.py +++ b/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,12 +35,10 @@ 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 from sentry.integrations.types import EXTERNAL_PROVIDERS, ExternalProviders -from sentry.locks import locks from sentry.models.organizationmapping import OrganizationMapping from sentry.models.organizationmembermapping import OrganizationMemberMapping from sentry.models.orgauthtoken import OrgAuthToken @@ -51,7 +50,6 @@ from sentry.users.models.useremail import UserEmail from sentry.users.services.user import RpcUser from sentry.utils.http import absolute_uri -from sentry.utils.retries import TimedRetryPolicy audit_logger = logging.getLogger("sentry.audit.user") logger = logging.getLogger(__name__) @@ -482,8 +480,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 +533,35 @@ 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) + | Q(email_unique__iexact=username) + ) + .exists() + ) + if username_or_primary_taken: + return False + verified_email_taken = ( + UserEmail.objects.filter(email__iexact=username, is_verified=True) + .exclude(user_id=exclude_user_id) + .exists() + ) + return not verified_email_taken + def write_relocation_import( self, scope: ImportScope, flags: ImportFlags ) -> tuple[int, ImportKind] | None: @@ -551,7 +576,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 +595,27 @@ 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", - ) - - # Perform the remainder of the write while we're still holding the lock. - return do_write() + 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): + base = self.username[: MAX_USERNAME_LENGTH - 13] + for _ in range(3): + # 3 chances to create a random suffix - randomly selecting an exact duplicate should be almost impossible + 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") + + return do_write() @classmethod def sanitize_relocation_json( diff --git a/src/sentry/users/web/accounts_form.py b/src/sentry/users/web/accounts_form.py index 6aa5590f61dc..1a39d0c8c5fb 100644 --- a/src/sentry/users/web/accounts_form.py +++ b/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() diff --git a/tests/sentry/backup/test_imports.py b/tests/sentry/backup/test_imports.py index d90d639f7c73..7c851b3669a1 100644 --- a/tests/sentry/backup/test_imports.py +++ b/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") diff --git a/tests/sentry/users/api/endpoints/test_user_details.py b/tests/sentry/users/api/endpoints/test_user_details.py index f1be8361f237..ff2e45efa73e 100644 --- a/tests/sentry/users/api/endpoints/test_user_details.py +++ b/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", diff --git a/tests/sentry/users/web/test_accounts_form.py b/tests/sentry/users/web/test_accounts_form.py index 417514509c72..f14e0f5ee871 100644 --- a/tests/sentry/users/web/test_accounts_form.py +++ b/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)