fix(auth): Create new org membership in automatic IdP migration (#30356)

Previously, users attaching an AuthIdentity to an existing account via
idpmigration.py could do so only if they already were a member of the
provider's organization and would hit an OrganizationMember.DoesNotExist
exception otherwise. Modify the flow to dispatch correctly to
handle_attach_identity, creating new entities of both AuthIdentity and
OrganizationMember when needed.

Refactor AuthIdentityHandler to take the identity mapping on init and
store the User with a matching email address as an instance variable.
Introduce a property to distinguish that user from the authenticated
user attached to the request, if any.

src/sentry/auth/helper.py

@@ -1,10 +1,11 @@
 import logging
 from dataclasses import dataclass
-from typing import Any, Dict, Mapping, Optional, Tuple
+from typing import Any, Dict, Mapping, Optional, Tuple, Union
 from uuid import uuid4
 
 from django.conf import settings
 from django.contrib import messages
+from django.contrib.auth.models import AnonymousUser
 from django.db import IntegrityError, transaction
 from django.db.models import F
 from django.http import HttpResponseRedirect
@@ -90,20 +91,19 @@ def _using_okta_migration_workaround(
     return has_flag and has_provider
 
 
-Identity = Mapping[str, Any]
-
-
-@dataclass(eq=True, frozen=True)
+@dataclass
 class AuthIdentityHandler:
 
     auth_provider: Optional[AuthProvider]
     provider: Provider
     organization: Organization
     request: HttpRequest
+    identity: Mapping[str, Any]
 
-    @property
-    def user(self) -> Any:
-        return self.request.user
+    def __post_init__(self) -> None:
+        self.user: Union[User, AnonymousUser] = (
+            self._find_user_from_email(self.identity.get("email")) or self.request.user
+        )
 
     class _NotCompletedSecurityChecks(Exception):
         pass
@@ -129,13 +129,12 @@ class AuthIdentityHandler:
         self,
         state: AuthHelperSessionStore,
         auth_identity: AuthIdentity,
-        identity: Identity,
     ) -> HttpResponseRedirect:
         # TODO(dcramer): this is very similar to attach
         now = timezone.now()
         auth_identity.update(
             data=self.provider.update_identity(
-                new_data=identity.get("data", {}), current_data=auth_identity.data
+                new_data=self.identity.get("data", {}), current_data=auth_identity.data
             ),
             last_verified=now,
             last_synced=now,
@@ -235,16 +234,12 @@ class AuthIdentityHandler:
             return None
 
     @transaction.atomic
-    def handle_attach_identity(
-        self,
-        identity: Identity,
-        member: Optional[OrganizationMember] = None,
-    ) -> AuthIdentity:
+    def handle_attach_identity(self, member: Optional[OrganizationMember] = None) -> AuthIdentity:
         """
         Given an already authenticated user, attach or re-attach an identity.
         """
         # prioritize identifying by the SSO provider's user ID
-        auth_identity = self._get_auth_identity(ident=identity["id"])
+        auth_identity = self._get_auth_identity(ident=self.identity["id"])
         if auth_identity is None:
             # otherwise look for an already attached identity
             # this can happen if the SSO provider's internal ID changes
@@ -255,8 +250,8 @@ class AuthIdentityHandler:
             auth_identity = AuthIdentity.objects.create(
                 auth_provider=self.auth_provider,
                 user=self.user,
-                ident=identity["id"],
-                data=identity.get("data", {}),
+                ident=self.identity["id"],
+                data=self.identity.get("data", {}),
             )
         else:
             auth_is_new = False
@@ -273,9 +268,9 @@ class AuthIdentityHandler:
             now = timezone.now()
             auth_identity.update(
                 user=self.user,
-                ident=identity["id"],
+                ident=self.identity["id"],
                 data=self.provider.update_identity(
-                    new_data=identity.get("data", {}), current_data=auth_identity.data
+                    new_data=self.identity.get("data", {}), current_data=auth_identity.data
                 ),
                 last_verified=now,
                 last_synced=now,
@@ -289,8 +284,8 @@ class AuthIdentityHandler:
                     "user_id": self.user.id,
                     "auth_identity_user_id": auth_identity.user.id,
                     "auth_provider_id": self.auth_provider.id,
-                    "idp_identity_id": identity["id"],
-                    "idp_identity_email": identity.get("email"),
+                    "idp_identity_id": self.identity["id"],
+                    "idp_identity_email": self.identity.get("email"),
                 },
             )
 
@@ -348,20 +343,15 @@ class AuthIdentityHandler:
             return self._handle_new_membership(auth_identity)
 
     @staticmethod
-    def _get_user(identity: Identity) -> Optional[User]:
-        email = identity.get("email")
+    def _find_user_from_email(email: Optional[str]) -> Optional[User]:
         if email is None:
             return None
 
         # TODO(dcramer): its possible they have multiple accounts and at
         # least one is managed (per the check below)
-        try:
-            return User.objects.filter(
-                id__in=UserEmail.objects.filter(email__iexact=email).values("user"),
-                is_active=True,
-            ).first()
-        except IndexError:
-            return None
+        return User.objects.filter(
+            id__in=UserEmail.objects.filter(email__iexact=email).values("user"), is_active=True
+        ).first()
 
     def _respond(
         self,
@@ -392,18 +382,22 @@ class AuthIdentityHandler:
 
         return response
 
-    def has_verified_account(self, identity: Identity, verification_value: Dict[str, Any]) -> bool:
-        acting_user = self._get_user(identity)
-
+    def has_verified_account(self, verification_value: Dict[str, Any]) -> bool:
         return (
-            verification_value["email"] == identity["email"]
-            and verification_value["user_id"] == acting_user.id
+            verification_value["email"] == self.identity["email"]
+            and verification_value["user_id"] == self.user.id
         )
 
+    def _has_usable_password(self):
+        return isinstance(self.user, User) and self.user.has_usable_password()
+
+    @property
+    def _logged_in_user(self) -> Optional[User]:
+        return self.request.user if self.request.user.is_authenticated else None
+
     def handle_unknown_identity(
         self,
         state: AuthHelperSessionStore,
-        identity: Identity,
     ) -> HttpResponseRedirect:
         """
         Flow is activated upon a user logging in to where an AuthIdentity is
@@ -421,64 +415,66 @@ class AuthIdentityHandler:
         - Should I create a new user based on this identity?
         """
         op = self.request.POST.get("op")
-        if not self.user.is_authenticated:
-            acting_user = self._get_user(identity)
-            login_form = AuthenticationForm(
+        login_form = (
+            None
+            if self._logged_in_user
+            else AuthenticationForm(
                 self.request,
                 self.request.POST if self.request.POST.get("op") == "login" else None,
-                initial={"username": acting_user.username if acting_user else None},
+                initial={"username": self.user.username if isinstance(self.user, User) else None},
             )
-        else:
-            acting_user = self.user
-            login_form = None
+        )
         # we don't trust all IDP email verification, so users can also confirm via one time email link
         is_account_verified = False
         if self.request.session.get("confirm_account_verification_key"):
             verification_key = self.request.session.get("confirm_account_verification_key")
             verification_value = get_verification_value_from_key(verification_key)
             if verification_value:
-                is_account_verified = self.has_verified_account(identity, verification_value)
+                is_account_verified = self.has_verified_account(verification_value)
 
-        if acting_user and identity.get("email_verified") or is_account_verified:
+        is_new_account = not self.user.is_authenticated  # stateful
+        if self.identity.get("email_verified") or is_account_verified:
             # we only allow this flow to happen if the existing user has
             # membership, otherwise we short circuit because it might be
             # an attempt to hijack membership of another organization
             has_membership = OrganizationMember.objects.filter(
-                user=acting_user, organization=self.organization
+                user=self.user, organization=self.organization
             ).exists()
             if has_membership:
                 try:
-                    self._login(acting_user)
+                    self._login(self.user)
                 except self._NotCompletedSecurityChecks:
                     # adding is_account_verified to the check below in order to redirect
                     # to 2fa when the user migrates their idp but has 2fa enabled,
                     # otherwise it would stop them from linking their sso provider
                     if (
-                        acting_user.has_usable_password()
+                        self._has_usable_password()
                         or is_account_verified
                         or _using_okta_migration_workaround(
-                            self.organization, acting_user, self.auth_provider
+                            self.organization, self.user, self.auth_provider
                         )
                     ):
                         return self._post_login_redirect()
                     else:
-                        acting_user = None
+                        is_new_account = True
                 else:
                     # assume they've confirmed they want to attach the identity
                     op = "confirm"
+            elif is_account_verified:
+                op = "confirm"
             else:
                 # force them to create a new account
-                acting_user = None
-        # without a usable password they can't login, so let's clear the acting_user
-        elif acting_user and not acting_user.has_usable_password():
-            acting_user = None
+                is_new_account = True
+        # without a usable password they can't login, so default to a new account
+        elif not self._has_usable_password():
+            is_new_account = True
 
         auth_identity = None
-        if op == "confirm" and self.user.is_authenticated:
-            auth_identity = self.handle_attach_identity(identity)
+        if op == "confirm" and self.user.is_authenticated or is_account_verified:
+            auth_identity = self.handle_attach_identity()
         elif op == "newuser":
-            auth_identity = self.handle_new_user(identity)
-        elif op == "login" and not self.user.is_authenticated:
+            auth_identity = self.handle_new_user()
+        elif op == "login" and not self._logged_in_user:
             # confirm authentication, login
             op = None
             if login_form.is_valid():
@@ -499,14 +495,14 @@ class AuthIdentityHandler:
             op = None
 
         if not op:
-            existing_user, template = self._dispatch_to_confirmation(identity)
+            existing_user, template = self._dispatch_to_confirmation(is_new_account)
 
             context = {
-                "identity": identity,
+                "identity": self.identity,
                 "provider": self.provider_name,
-                "identity_display_name": identity.get("name") or identity.get("email"),
-                "identity_identifier": identity.get("email") or identity.get("id"),
-                "existing_user": existing_user or acting_user,
+                "identity_display_name": self.identity.get("name") or self.identity.get("email"),
+                "identity_identifier": self.identity.get("email") or self.identity.get("id"),
+                "existing_user": existing_user,
             }
             if login_form:
                 context["login_form"] = login_form
@@ -536,28 +532,29 @@ class AuthIdentityHandler:
             # A blank character is needed to prevent an HTML span from collapsing
             return " "
 
-    def _dispatch_to_confirmation(self, identity: Identity) -> Tuple[Optional[User], str]:
-        if self.user.is_authenticated:
-            return self.user, "auth-confirm-link"
+    def _dispatch_to_confirmation(self, is_new_account: bool) -> Tuple[Optional[User], str]:
+        if self._logged_in_user:
+            return self._logged_in_user, "auth-confirm-link"
 
         if features.has("organizations:idp-automatic-migration", self.organization):
-            existing_user = self._get_user(identity)
-            if existing_user and not existing_user.has_usable_password():
+            if not self._has_usable_password():
                 send_one_time_account_confirm_link(
-                    existing_user,
+                    self.user,
                     self.organization,
                     self.auth_provider,
-                    identity["email"],
-                    identity["id"],
+                    self.identity["email"],
+                    self.identity["id"],
                 )
-                return existing_user, "auth-confirm-account"
+                return self.user, "auth-confirm-account"
 
         self.request.session.set_test_cookie()
-        return None, "auth-confirm-identity"
+        return None if is_new_account else self.user, "auth-confirm-identity"
 
-    def handle_new_user(self, identity: Identity) -> AuthIdentity:
+    def handle_new_user(self) -> AuthIdentity:
         user = User.objects.create(
-            username=uuid4().hex, email=identity["email"], name=identity.get("name", "")[:200]
+            username=uuid4().hex,
+            email=self.identity["email"],
+            name=self.identity.get("name", "")[:200],
         )
 
         if settings.TERMS_URL and settings.PRIVACY_URL:
@@ -568,12 +565,12 @@ class AuthIdentityHandler:
                 auth_identity = AuthIdentity.objects.create(
                     auth_provider=self.auth_provider,
                     user=user,
-                    ident=identity["id"],
-                    data=identity.get("data", {}),
+                    ident=self.identity["id"],
+                    data=self.identity.get("data", {}),
                 )
         except IntegrityError:
-            auth_identity = self._get_auth_identity(ident=identity["id"])
-            auth_identity.update(user=user, data=identity.get("data", {}))
+            auth_identity = self._get_auth_identity(ident=self.identity["id"])
+            auth_identity.update(user=user, data=self.identity.get("data", {}))
 
         user.send_confirm_emails(is_new_user=True)
         provider = self.auth_provider.provider if self.auth_provider else None
@@ -698,14 +695,13 @@ class AuthHelper(Pipeline):
 
         return response
 
-    @property
-    def auth_handler(self):
+    def auth_handler(self, identity: Mapping[str, Any]):
         return AuthIdentityHandler(
-            self.provider_model, self.provider, self.organization, self.request
+            self.provider_model, self.provider, self.organization, self.request, identity
         )
 
     @transaction.atomic
-    def _finish_login_pipeline(self, identity: Identity):
+    def _finish_login_pipeline(self, identity: Mapping[str, Any]):
         """
         The login flow executes both with anonymous and authenticated users.
 
@@ -741,6 +737,7 @@ class AuthHelper(Pipeline):
                 except AuthIdentity.DoesNotExist:
                     auth_identity = None
 
+            auth_handler = self.auth_handler(identity)
             if not auth_identity:
                 if _using_okta_migration_workaround(
                     self.organization, self.request.user, auth_provider
@@ -758,7 +755,7 @@ class AuthHelper(Pipeline):
                         },
                     )
 
-                return self.auth_handler.handle_unknown_identity(self.state, identity)
+                return auth_handler.handle_unknown_identity(self.state)
 
             # If the User attached to this AuthIdentity is not active,
             # we want to clobber the old account and take it over, rather than
@@ -767,13 +764,13 @@ class AuthHelper(Pipeline):
                 # Current user is also not logged in, so we have to
                 # assume unknown.
                 if not self.request.user.is_authenticated:
-                    return self.auth_handler.handle_unknown_identity(self.state, identity)
-                auth_identity = self.auth_handler.handle_attach_identity(identity)
+                    return auth_handler.handle_unknown_identity(self.state)
+                auth_identity = auth_handler.handle_attach_identity()
 
-            return self.auth_handler.handle_existing_identity(self.state, auth_identity, identity)
+            return auth_handler.handle_existing_identity(self.state, auth_identity)
 
     @transaction.atomic
-    def _finish_setup_pipeline(self, identity: Identity):
+    def _finish_setup_pipeline(self, identity: Mapping[str, Any]):
         """
         The setup flow creates the auth provider as well as an identity linked
         to the active user.
@@ -801,7 +798,7 @@ class AuthHelper(Pipeline):
             organization=self.organization, provider=self.provider.key, config=config
         )
 
-        self.auth_handler.handle_attach_identity(identity, om)
+        self.auth_handler(identity).handle_attach_identity(om)
 
         auth.mark_sso_complete(request, self.organization.id)
 

src/sentry/auth/idpmigration.py

@@ -84,9 +84,13 @@ class AccountConfirmLink:
 
     def store_in_redis(self) -> None:
         cluster = get_redis_cluster()
-        member_id = OrganizationMember.objects.get(
-            organization=self.organization, user=self.user
-        ).id
+
+        try:
+            member_id = OrganizationMember.objects.get(
+                organization=self.organization, user=self.user
+            ).id
+        except OrganizationMember.DoesNotExist:
+            member_id = None
 
         verification_value = {
             "user_id": self.user.id,

src/sentry/web/frontend/idp_email_verification.py

@@ -2,7 +2,7 @@ from django.http.response import HttpResponse
 from rest_framework.request import Request
 
 from sentry.auth.idpmigration import SSO_VERIFICATION_KEY, get_verification_value_from_key
-from sentry.models import OrganizationMember
+from sentry.models import Organization
 from sentry.web.frontend.base import BaseView
 from sentry.web.helpers import render_to_response
 
@@ -17,13 +17,7 @@ class AccountConfirmationView(BaseView):
         if not verification_value:
             return render_to_response("sentry/idp_account_not_verified.html", request=request)
 
-        try:
-            org = OrganizationMember.objects.get(
-                id=verification_value["member_id"]
-            ).organization.slug
-        except OrganizationMember.DoesNotExist:
-            org = None
-
+        org = self._recover_org_slug(verification_value)
         context = {"org": org}
 
         if verification_value and org:
@@ -32,3 +26,14 @@ class AccountConfirmationView(BaseView):
                 "sentry/idp_account_verified.html", context=context, request=request
             )
         return render_to_response("sentry/idp_account_not_verified.html", request=request)
+
+    @staticmethod
+    def _recover_org_slug(verification_value):
+        organization_id = verification_value.get("organization_id")
+        if organization_id is None:
+            return None
+        try:
+            org = Organization.objects.get(id=organization_id)
+        except Organization.DoesNotExist:
+            return None
+        return org.slug

tests/sentry/auth/test_helper.py

@@ -20,6 +20,7 @@ from sentry.models import (
     InviteStatus,
     OrganizationMember,
     OrganizationMemberTeam,
+    UserEmail,
 )
 from sentry.testutils import TestCase
 from sentry.utils import json
@@ -50,12 +51,21 @@ class AuthIdentityHandlerTest(TestCase):
             "data": {"foo": "bar"},
         }
 
-        self.handler = AuthIdentityHandler(
-            self.auth_provider, Provider(self.provider), self.organization, self.request
-        )
-
         self.state = AuthHelperSessionStore(self.request, "pipeline")
 
+    @property
+    def handler(self):
+        return self._handler_with(self.identity)
+
+    def _handler_with(self, identity):
+        return AuthIdentityHandler(
+            self.auth_provider,
+            Provider(self.provider),
+            self.organization,
+            self.request,
+            identity,
+        )
+
     def set_up_user(self):
         """Set up a persistent user and associate it to the request.
 
@@ -80,7 +90,7 @@ class HandleNewUserTest(AuthIdentityHandlerTest):
     @mock.patch("sentry.analytics.record")
     def test_simple(self, mock_record):
 
-        auth_identity = self.handler.handle_new_user(self.identity)
+        auth_identity = self.handler.handle_new_user()
         user = auth_identity.user
 
         assert user.email == self.email
@@ -100,7 +110,7 @@ class HandleNewUserTest(AuthIdentityHandlerTest):
     def test_associated_existing_member_invite_by_email(self):
         member = OrganizationMember.objects.create(organization=self.organization, email=self.email)
 
-        auth_identity = self.handler.handle_new_user(self.identity)
+        auth_identity = self.handler.handle_new_user()
 
         assigned_member = OrganizationMember.objects.get(
             organization=self.organization, user=auth_identity.user
@@ -115,7 +125,7 @@ class HandleNewUserTest(AuthIdentityHandlerTest):
             invite_status=InviteStatus.REQUESTED_TO_BE_INVITED.value,
         )
 
-        auth_identity = self.handler.handle_new_user(self.identity)
+        auth_identity = self.handler.handle_new_user()
 
         assert OrganizationMember.objects.filter(
             organization=self.organization,
@@ -136,7 +146,7 @@ class HandleNewUserTest(AuthIdentityHandlerTest):
             {"memberId": member.id, "token": member.token, "url": ""}
         )
 
-        auth_identity = self.handler.handle_new_user(self.identity)
+        auth_identity = self.handler.handle_new_user()
 
         assigned_member = OrganizationMember.objects.get(
             organization=self.organization, user=auth_identity.user
@@ -151,7 +161,7 @@ class HandleExistingIdentityTest(AuthIdentityHandlerTest):
         mock_auth.get_login_redirect.return_value = "test_login_url"
         user, auth_identity = self.set_up_user_identity()
 
-        redirect = self.handler.handle_existing_identity(self.state, auth_identity, self.identity)
+        redirect = self.handler.handle_existing_identity(self.state, auth_identity)
 
         assert redirect.url == mock_auth.get_login_redirect.return_value
         assert mock_auth.get_login_redirect.called_with(self.request)
@@ -174,9 +184,7 @@ class HandleExistingIdentityTest(AuthIdentityHandlerTest):
             mock_auth.get_login_redirect.return_value = "test_login_url"
             user, auth_identity = self.set_up_user_identity()
 
-            redirect = self.handler.handle_existing_identity(
-                self.state, auth_identity, self.identity
-            )
+            redirect = self.handler.handle_existing_identity(self.state, auth_identity)
 
             assert redirect.url == mock_auth.get_login_redirect.return_value
             assert mock_auth.get_login_redirect.called_with(self.request)
@@ -196,7 +204,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
     def test_new_identity(self, mock_messages):
         self.set_up_user()
 
-        auth_identity = self.handler.handle_attach_identity(self.identity)
+        auth_identity = self.handler.handle_attach_identity()
         assert auth_identity.ident == self.identity["id"]
         assert auth_identity.data == self.identity["data"]
 
@@ -226,7 +234,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
         user = self.set_up_user()
         existing_om = OrganizationMember.objects.create(user=user, organization=self.organization)
 
-        auth_identity = self.handler.handle_attach_identity(self.identity)
+        auth_identity = self.handler.handle_attach_identity()
         assert auth_identity.ident == self.identity["id"]
         assert auth_identity.data == self.identity["data"]
 
@@ -242,7 +250,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
     def test_existing_identity(self, mock_messages):
         user, existing_identity = self.set_up_user_identity()
 
-        returned_identity = self.handler.handle_attach_identity(self.identity)
+        returned_identity = self.handler.handle_attach_identity()
         assert returned_identity == existing_identity
         assert not mock_messages.add_message.called
 
@@ -255,7 +263,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
         )
         OrganizationMember.objects.create(user=other_user, organization=self.organization)
 
-        returned_identity = self.handler.handle_attach_identity(self.identity)
+        returned_identity = self.handler.handle_attach_identity()
         assert returned_identity.user == request_user
         assert returned_identity.ident == self.identity["id"]
         assert returned_identity.data == self.identity["data"]
@@ -278,7 +286,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
 
 class HandleUnknownIdentityTest(AuthIdentityHandlerTest):
     def _test_simple(self, mock_render, expected_template):
-        redirect = self.handler.handle_unknown_identity(self.state, self.identity)
+        redirect = self.handler.handle_unknown_identity(self.state)
 
         assert redirect is mock_render.return_value
         template, context, request = mock_render.call_args.args
@@ -400,24 +408,21 @@ class HasVerifiedAccountTest(AuthIdentityHandlerTest):
             "identity_id": self.identity_id,
         }
 
-    @mock.patch("sentry.auth.helper.AuthIdentityHandler._get_user")
-    def test_has_verified_account_success(self, mock_get_user):
-        mock_get_user.return_value = self.user
-        assert self.handler.has_verified_account(self.identity, self.verification_value) is True
+    def test_has_verified_account_success(self):
+        UserEmail.objects.create(email=self.email, user=self.user)
+        assert self.handler.has_verified_account(self.verification_value) is True
 
-    @mock.patch("sentry.auth.helper.AuthIdentityHandler._get_user")
-    def test_has_verified_account_fail_email(self, mock_get_user):
-        mock_get_user.return_value = self.user
+    def test_has_verified_account_fail_email(self):
+        UserEmail.objects.create(email=self.email, user=self.user)
         identity = {
             "id": "1234",
             "email": "b@test.com",
             "name": "Morty",
             "data": {"foo": "bar"},
         }
-        assert self.handler.has_verified_account(identity, self.verification_value) is False
+        assert self._handler_with(identity).has_verified_account(self.verification_value) is False
 
-    @mock.patch("sentry.auth.helper.AuthIdentityHandler._get_user")
-    def test_has_verified_account_fail_user_id(self, mock_get_user):
-        mock_get_user.return_value = self.create_user()
-        self.wrong_user_flag = True
-        assert self.handler.has_verified_account(self.identity, self.verification_value) is False
+    def test_has_verified_account_fail_user_id(self):
+        wrong_user = self.create_user()
+        UserEmail.objects.create(email=self.email, user=wrong_user)
+        assert self.handler.has_verified_account(self.verification_value) is False

tests/sentry/auth/test_idpmigration.py

@@ -5,6 +5,7 @@ from django.urls import reverse
 import sentry.auth.idpmigration as idpmigration
 from sentry.models import AuthProvider, OrganizationMember
 from sentry.testutils import TestCase
+from sentry.utils import json
 
 
 class IDPMigrationTests(TestCase):
@@ -15,17 +16,40 @@ class IDPMigrationTests(TestCase):
         self.email = "test@example.com"
         self.org = self.create_organization()
         self.provider = AuthProvider.objects.create(organization=self.org, provider="dummy")
-        OrganizationMember.objects.create(organization=self.org, user=self.user)
+
+    IDENTITY_ID = "drgUQCLzOyfHxmTyVs0G"
 
     def test_send_one_time_account_confirm_link(self):
+        om = OrganizationMember.objects.create(organization=self.org, user=self.user)
         link = idpmigration.send_one_time_account_confirm_link(
-            self.user, self.org, self.provider, self.email, "drgUQCLzOyfHxmTyVs0G"
+            self.user, self.org, self.provider, self.email, self.IDENTITY_ID
         )
         assert re.match(r"auth:one-time-key:\w{32}", link.verification_key)
 
+        value = json.loads(idpmigration.get_redis_cluster().get(link.verification_key))
+        assert value["user_id"] == self.user.id
+        assert value["email"] == self.email
+        assert value["member_id"] == om.id
+        assert value["organization_id"] == self.org.id
+        assert value["identity_id"] == self.IDENTITY_ID
+        assert value["provider"] == "dummy"
+
+    def test_send_without_org_membership(self):
+        link = idpmigration.send_one_time_account_confirm_link(
+            self.user, self.org, self.provider, self.email, self.IDENTITY_ID
+        )
+
+        value = json.loads(idpmigration.get_redis_cluster().get(link.verification_key))
+        assert value["user_id"] == self.user.id
+        assert value["email"] == self.email
+        assert value["member_id"] is None
+        assert value["organization_id"] == self.org.id
+        assert value["identity_id"] == self.IDENTITY_ID
+        assert value["provider"] == "dummy"
+
     def test_verify_account(self):
         link = idpmigration.send_one_time_account_confirm_link(
-            self.user, self.org, self.provider, self.email, "drgUQCLzOyfHxmTyVs0G"
+            self.user, self.org, self.provider, self.email, self.IDENTITY_ID
         )
         path = reverse(
             "sentry-idp-email-verification",
@@ -39,7 +63,7 @@ class IDPMigrationTests(TestCase):
 
     def test_verify_account_wrong_key(self):
         idpmigration.send_one_time_account_confirm_link(
-            self.user, self.org, self.provider, self.email, "drgUQCLzOyfHxmTyVs0G"
+            self.user, self.org, self.provider, self.email, self.IDENTITY_ID
         )
         path = reverse(
             "sentry-idp-email-verification",

tests/sentry/web/frontend/test_auth_organization_login.py

@@ -6,7 +6,6 @@ from django.urls import reverse
 from exam import fixture
 
 from sentry.auth.authenticators import RecoveryCodeInterface, TotpInterface
-from sentry.auth.provider import MigratingIdentityId
 from sentry.models import (
     AuthIdentity,
     AuthProvider,
@@ -783,7 +782,6 @@ class OrganizationAuthLoginTest(AuthProviderTestCase):
     @override_settings(SENTRY_SINGLE_ORGANIZATION=True)
     @with_feature({"organizations:create": False})
     def test_basic_auth_flow_as_invited_user(self):
-
         user = self.create_user("foor@example.com")
         self.create_member(organization=self.organization, user=user)
         member = OrganizationMember.objects.get(organization=self.organization, user=user)
@@ -1003,10 +1001,8 @@ class OrganizationAuthLoginNoPasswordTest(AuthProviderTestCase):
         UserEmail.objects.filter(user=self.user, email="bar@example.com").update(is_verified=False)
 
     @with_feature("organizations:idp-automatic-migration")
-    @mock.patch("sentry.auth.helper.send_one_time_account_confirm_link")
-    def test_flow_verify_and_link_without_password_sends_email(
-        self, mock_send_one_time_account_confirm_link
-    ):
+    @mock.patch("sentry.auth.idpmigration.MessageBuilder")
+    def test_flow_verify_and_link_without_password_sends_email(self, email):
         assert not self.user.has_usable_password()
         self.create_member(organization=self.organization, user=self.user)
 
@@ -1016,17 +1012,74 @@ class OrganizationAuthLoginNoPasswordTest(AuthProviderTestCase):
         assert self.provider.TEMPLATE in resp.content.decode("utf-8")
 
         resp = self.client.post(self.auth_sso_path, {"email": "bar@example.com"})
-        mock_send_one_time_account_confirm_link.assert_called_with(
-            self.user,
-            self.organization,
-            self.auth_provider,
-            "bar@example.com",
-            MigratingIdentityId(id="bar@example.com", legacy_id=None),
-        )
         self.assertTemplateUsed(resp, "sentry/auth-confirm-account.html")
         assert resp.status_code == 200
         assert resp.context["existing_user"] == self.user
 
+        _, message = email.call_args
+        context = message["context"]
+        assert context["user"] == self.user
+        assert context["email"] == self.user.email
+        assert context["organization"] == self.organization.name
+        email.return_value.send_async.assert_called_with([self.user.email])
+
+        path = reverse("sentry-idp-email-verification", args=[context["verification_key"]])
+        resp = self.client.get(path)
+        assert resp.templates[0].name == "sentry/idp_account_verified.html"
+        assert resp.status_code == 200
+
+        path = reverse("sentry-auth-organization", args=[self.organization.slug])
+        resp = self.client.post(path, follow=True)
+        assert resp.redirect_chain == [
+            (reverse("sentry-login"), 302),
+            ("/organizations/foo/issues/", 302),
+        ]
+
+        auth_identity = AuthIdentity.objects.get(auth_provider=self.auth_provider)
+        assert self.user == auth_identity.user
+
+    @with_feature("organizations:idp-automatic-migration")
+    @mock.patch("sentry.auth.idpmigration.MessageBuilder")
+    def test_flow_verify_without_org_membership(self, email):
+        assert not self.user.has_usable_password()
+        assert not OrganizationMember.objects.filter(
+            organization=self.organization, user=self.user
+        ).exists()
+
+        resp = self.client.post(self.path, {"init": True})
+
+        assert resp.status_code == 200
+        assert self.provider.TEMPLATE in resp.content.decode("utf-8")
+
+        resp = self.client.post(self.auth_sso_path, {"email": "bar@example.com"})
+        self.assertTemplateUsed(resp, "sentry/auth-confirm-account.html")
+        assert resp.status_code == 200
+        assert resp.context["existing_user"] == self.user
+
+        _, message = email.call_args
+        context = message["context"]
+        assert context["organization"] == self.organization.name
+
+        path = reverse("sentry-idp-email-verification", args=[context["verification_key"]])
+        resp = self.client.get(path)
+        assert resp.templates[0].name == "sentry/idp_account_verified.html"
+        assert resp.status_code == 200
+
+        path = reverse("sentry-auth-organization", args=[self.organization.slug])
+        resp = self.client.post(path, follow=True)
+        assert resp.redirect_chain == [
+            (reverse("sentry-login"), 302),
+            ("/organizations/foo/issues/", 302),
+        ]
+
+        auth_identity = AuthIdentity.objects.get(auth_provider=self.auth_provider)
+        assert self.user == auth_identity.user
+
+        # Check that OrganizationMember was created as a side effect
+        assert OrganizationMember.objects.filter(
+            organization=self.organization, user=self.user
+        ).exists()
+
     @with_feature("organizations:idp-automatic-migration")
     @mock.patch("sentry.auth.idpmigration.MessageBuilder")
     def test_flow_verify_and_link_without_password_login_success(self, email):