ref(auth): Refactor auth/helper.py for clarity (#26571)

src/sentry/auth/helper.py

@@ -1,4 +1,6 @@
 import logging
+from dataclasses import dataclass
+from typing import Any, Mapping, Optional
 from uuid import uuid4
 
 from django.conf import settings
@@ -6,6 +8,8 @@ from django.contrib import messages
 from django.db import IntegrityError, transaction
 from django.db.models import F
 from django.http import HttpResponseRedirect
+from django.http.request import HttpRequest
+from django.http.response import HttpResponse
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
@@ -15,7 +19,7 @@ from sentry import features
 from sentry.api.invite_helper import ApiInviteHelper, remove_invite_cookie
 from sentry.app import locks
 from sentry.auth.exceptions import IdentityNotValid
-from sentry.auth.provider import MigratingIdentityId
+from sentry.auth.provider import MigratingIdentityId, Provider
 from sentry.auth.superuser import is_active_superuser
 from sentry.models import (
     AuditLogEntry,
@@ -117,412 +121,468 @@ class RedisBackedState:
 def handle_existing_identity(
     auth_provider, provider, organization, request, state, auth_identity, identity
 ):
-    # TODO(dcramer): this is very similar to attach
-    now = timezone.now()
-    auth_identity.update(
-        data=provider.update_identity(
-            new_data=identity.get("data", {}), current_data=auth_identity.data
-        ),
-        last_verified=now,
-        last_synced=now,
-    )
-
-    try:
-        member = OrganizationMember.objects.get(user=auth_identity.user, organization=organization)
-    except OrganizationMember.DoesNotExist:
-        # this is likely the case when someone was removed from the org
-        # but still has access to rejoin
-        member = handle_new_membership(auth_provider, organization, request, auth_identity)
-    else:
-        if getattr(member.flags, "sso:invalid") or not getattr(member.flags, "sso:linked"):
-            setattr(member.flags, "sso:invalid", False)
-            setattr(member.flags, "sso:linked", True)
-            member.save()
+    # Deprecated. For backwards compatibility. Prefer AuthIdentityHandler interface.
+    # TODO(RyanSkonnord): Delete after calls from getsentry are cleaned up
+    return AuthIdentityHandler(
+        auth_provider, provider, organization, request
+    ).handle_existing_identity(state, auth_identity, identity)
 
-    user = auth_identity.user
-    user.backend = settings.AUTHENTICATION_BACKENDS[0]
-
-    if not auth.login(
-        request, user, after_2fa=request.build_absolute_uri(), organization_id=organization.id
-    ):
-        return HttpResponseRedirect(auth.get_login_redirect(request))
-
-    state.clear()
-    metrics.incr(
-        "sso.login-success",
-        tags={
-            "provider": provider.key,
-            "organization_id": organization.id,
-            "user_id": request.user.id,
-        },
-        skip_internal=False,
-        sample_rate=1.0,
-    )
-
-    if not is_active_superuser(request):
-        # set activeorg to ensure correct redirect upon logging in
-        request.session["activeorg"] = organization.slug
-    return HttpResponseRedirect(auth.get_login_redirect(request))
-
-
-def handle_new_membership(auth_provider, organization, request, auth_identity):
-    user = auth_identity.user
-
-    # If the user is either currently *pending* invite acceptance (as indicated
-    # from the pending-invite cookie) OR an existing invite exists on this
-    # organziation for the email provided by the identity provider.
-    invite_helper = ApiInviteHelper.from_cookie_or_email(
-        request=request, organization=organization, email=user.email
-    )
-
-    # If we are able to accept an existing invite for the user for this
-    # organization, do so, otherwise handle new membership
-    if invite_helper:
-        if invite_helper.invite_approved:
-            invite_helper.accept_invite(user)
-            return
 
-        # It's possible the user has an _invite request_ that hasn't been approved yet,
-        # and is able to join the organization without an invite through the SSO flow.
-        # In that case, delete the invite request and create a new membership.
-        invite_helper.handle_invite_not_approved()
-
-    # Otherwise create a new membership
-    om = OrganizationMember.objects.create(
-        organization=organization,
-        role=organization.default_role,
-        user=user,
-        flags=OrganizationMember.flags["sso:linked"],
-    )
-
-    default_teams = auth_provider.default_teams.all()
-    for team in default_teams:
-        OrganizationMemberTeam.objects.create(team=team, organizationmember=om)
-
-    AuditLogEntry.objects.create(
-        organization=organization,
-        actor=user,
-        ip_address=request.META["REMOTE_ADDR"],
-        target_object=om.id,
-        target_user=om.user,
-        event=AuditLogEntryEvent.MEMBER_ADD,
-        data=om.get_audit_log_data(),
-    )
-
-    return om
-
-
-@transaction.atomic
 def handle_attach_identity(auth_provider, request, organization, provider, identity, member=None):
-    """
-    Given an already authenticated user, attach or re-attach an identity.
-    """
-    user = request.user
+    # Deprecated. For backwards compatibility. Prefer AuthIdentityHandler interface.
+    # TODO(RyanSkonnord): Delete after calls from getsentry are cleaned up
+    return AuthIdentityHandler(
+        auth_provider, provider, organization, request
+    ).handle_attach_identity(identity, member)
 
-    try:
-        try:
-            # prioritize identifying by the SSO provider's user ID
-            auth_identity = AuthIdentity.objects.get(
-                auth_provider=auth_provider, ident=identity["id"]
-            )
-        except AuthIdentity.DoesNotExist:
-            # otherwise look for an already attached identity
-            # this can happen if the SSO provider's internal ID changes
-            auth_identity = AuthIdentity.objects.get(auth_provider=auth_provider, user=user)
-    except AuthIdentity.DoesNotExist:
-        auth_identity = AuthIdentity.objects.create(
-            auth_provider=auth_provider,
-            user=user,
-            ident=identity["id"],
-            data=identity.get("data", {}),
+
+def handle_unknown_identity(request, organization, auth_provider, provider, state, identity):
+    # Deprecated. For backwards compatibility. Prefer AuthIdentityHandler interface.
+    # TODO(RyanSkonnord): Delete after calls from getsentry are cleaned up
+    return AuthIdentityHandler(
+        auth_provider, provider, organization, request
+    ).handle_unknown_identity(state, identity)
+
+
+def handle_new_user(auth_provider, organization, request, identity):
+    # Deprecated. For backwards compatibility. Prefer AuthIdentityHandler interface.
+    # TODO(RyanSkonnord): Delete after calls from getsentry are cleaned up
+    return AuthIdentityHandler(auth_provider, None, organization, request).handle_new_user(identity)
+
+
+Identity = Mapping[str, Any]
+
+
+@dataclass(eq=True, frozen=True)
+class AuthIdentityHandler:
+
+    auth_provider: Optional[AuthProvider]
+    provider: Provider
+    organization: Organization
+    request: HttpRequest
+
+    @property
+    def user(self) -> Any:
+        return self.request.user
+
+    class _NotCompletedSecurityChecks(Exception):
+        pass
+
+    def _login(self, user: Any) -> None:
+        user_was_logged_in = auth.login(
+            self.request,
+            user,
+            after_2fa=self.request.build_absolute_uri(),
+            organization_id=self.organization.id,
         )
-        auth_is_new = True
-    else:
-        now = timezone.now()
+        if not user_was_logged_in:
+            raise self._NotCompletedSecurityChecks()
 
-        # TODO(dcramer): this might leave the user with duplicate accounts,
-        # and in that kind of situation its very reasonable that we could
-        # test email addresses + is_managed to determine if we can auto
-        # merge
-        if auth_identity.user != user:
-            # it's possible the user has an existing identity, let's wipe it out
-            # so that the new identifier gets used (other we'll hit a constraint)
-            # violation since one might exist for (provider, user) as well as
-            # (provider, ident)
-            AuthIdentity.objects.exclude(id=auth_identity.id).filter(
-                auth_provider=auth_provider, user=user
-            ).delete()
-
-            # since we've identify an identity which is no longer valid
-            # lets preemptively mark it as such
-            try:
-                other_member = OrganizationMember.objects.get(
-                    user=auth_identity.user_id, organization=organization
-                )
-            except OrganizationMember.DoesNotExist:
-                pass
-            else:
-                other_member.flags["sso:invalid"] = True
-                other_member.flags["sso:linked"] = False
-                other_member.save()
+    @staticmethod
+    def _set_linked_flag(member: OrganizationMember) -> None:
+        if getattr(member.flags, "sso:invalid") or not getattr(member.flags, "sso:linked"):
+            setattr(member.flags, "sso:invalid", False)
+            setattr(member.flags, "sso:linked", True)
+            member.save()
 
+    def handle_existing_identity(
+        self,
+        state: RedisBackedState,
+        auth_identity: AuthIdentity,
+        identity: Identity,
+    ) -> HttpResponseRedirect:
+        # TODO(dcramer): this is very similar to attach
+        now = timezone.now()
         auth_identity.update(
-            user=user,
-            ident=identity["id"],
-            data=provider.update_identity(
+            data=self.provider.update_identity(
                 new_data=identity.get("data", {}), current_data=auth_identity.data
             ),
             last_verified=now,
             last_synced=now,
         )
-        auth_is_new = False
 
-    if member is None:
         try:
-            member = OrganizationMember.objects.get(user=user, organization=organization)
-        except OrganizationMember.DoesNotExist:
-            member = OrganizationMember.objects.create(
-                organization=organization,
-                role=organization.default_role,
-                user=user,
-                flags=OrganizationMember.flags["sso:linked"],
+            member = OrganizationMember.objects.get(
+                user=auth_identity.user, organization=self.organization
             )
+        except OrganizationMember.DoesNotExist:
+            # this is likely the case when someone was removed from the org
+            # but still has access to rejoin
+            member = self._handle_new_membership(auth_identity)
+        else:
+            self._set_linked_flag(member)
 
-            default_teams = auth_provider.default_teams.all()
-            for team in default_teams:
-                OrganizationMemberTeam.objects.create(team=team, organizationmember=member)
+        user = auth_identity.user
+        user.backend = settings.AUTHENTICATION_BACKENDS[0]
 
-            AuditLogEntry.objects.create(
-                organization=organization,
-                actor=user,
-                ip_address=request.META["REMOTE_ADDR"],
-                target_object=member.id,
-                target_user=user,
-                event=AuditLogEntryEvent.MEMBER_ADD,
-                data=member.get_audit_log_data(),
-            )
-    if getattr(member.flags, "sso:invalid") or not getattr(member.flags, "sso:linked"):
-        setattr(member.flags, "sso:invalid", False)
-        setattr(member.flags, "sso:linked", True)
-        member.save()
+        try:
+            self._login(user)
+        except self._NotCompletedSecurityChecks:
+            return HttpResponseRedirect(auth.get_login_redirect(self.request))
+
+        state.clear()
+        metrics.incr(
+            "sso.login-success",
+            tags={
+                "provider": self.provider.key,
+                "organization_id": self.organization.id,
+                "user_id": self.user.id,
+            },
+            skip_internal=False,
+            sample_rate=1.0,
+        )
+
+        if not is_active_superuser(self.request):
+            # set activeorg to ensure correct redirect upon logging in
+            self.request.session["activeorg"] = self.organization.slug
+        return HttpResponseRedirect(auth.get_login_redirect(self.request))
+
+    def _handle_new_membership(self, auth_identity: AuthIdentity) -> Optional[OrganizationMember]:
+        user = auth_identity.user
+
+        # If the user is either currently *pending* invite acceptance (as indicated
+        # from the pending-invite cookie) OR an existing invite exists on this
+        # organziation for the email provided by the identity provider.
+        invite_helper = ApiInviteHelper.from_cookie_or_email(
+            request=self.request, organization=self.organization, email=user.email
+        )
+
+        # If we are able to accept an existing invite for the user for this
+        # organization, do so, otherwise handle new membership
+        if invite_helper:
+            if invite_helper.invite_approved:
+                invite_helper.accept_invite(user)
+                return None
+
+            # It's possible the user has an _invite request_ that hasn't been approved yet,
+            # and is able to join the organization without an invite through the SSO flow.
+            # In that case, delete the invite request and create a new membership.
+            invite_helper.handle_invite_not_approved()
+
+        # Otherwise create a new membership
+        om = OrganizationMember.objects.create(
+            organization=self.organization,
+            role=self.organization.default_role,
+            user=user,
+            flags=OrganizationMember.flags["sso:linked"],
+        )
+
+        default_teams = self.auth_provider.default_teams.all()
+        for team in default_teams:
+            OrganizationMemberTeam.objects.create(team=team, organizationmember=om)
 
-    if auth_is_new:
         AuditLogEntry.objects.create(
-            organization=organization,
+            organization=self.organization,
             actor=user,
-            ip_address=request.META["REMOTE_ADDR"],
-            target_object=auth_identity.id,
-            event=AuditLogEntryEvent.SSO_IDENTITY_LINK,
-            data=auth_identity.get_audit_log_data(),
+            ip_address=self.request.META["REMOTE_ADDR"],
+            target_object=om.id,
+            target_user=om.user,
+            event=AuditLogEntryEvent.MEMBER_ADD,
+            data=om.get_audit_log_data(),
         )
 
-        messages.add_message(request, messages.SUCCESS, OK_LINK_IDENTITY)
+        return om
 
-    return auth_identity
+    def _get_auth_identity(self, **params) -> Optional[AuthIdentity]:
+        try:
+            return AuthIdentity.objects.get(auth_provider=self.auth_provider, **params)
+        except AuthIdentity.DoesNotExist:
+            return None
 
+    @transaction.atomic
+    def handle_attach_identity(
+        self,
+        identity: Identity,
+        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"])
+        if auth_identity is None:
+            # otherwise look for an already attached identity
+            # this can happen if the SSO provider's internal ID changes
+            auth_identity = self._get_auth_identity(user=self.user)
 
-def get_display_name(identity):
-    return identity.get("name") or identity.get("email")
+        if auth_identity is None:
+            auth_is_new = True
+            auth_identity = AuthIdentity.objects.create(
+                auth_provider=self.auth_provider,
+                user=self.user,
+                ident=identity["id"],
+                data=identity.get("data", {}),
+            )
+        else:
+            auth_is_new = False
+
+            # TODO(dcramer): this might leave the user with duplicate accounts,
+            # and in that kind of situation its very reasonable that we could
+            # test email addresses + is_managed to determine if we can auto
+            # merge
+            if auth_identity.user != self.user:
+                self._wipe_existing_identity(auth_identity)
+
+            now = timezone.now()
+            auth_identity.update(
+                user=self.user,
+                ident=identity["id"],
+                data=self.provider.update_identity(
+                    new_data=identity.get("data", {}), current_data=auth_identity.data
+                ),
+                last_verified=now,
+                last_synced=now,
+            )
+
+        if member is None:
+            member = self._get_organization_member()
+        self._set_linked_flag(member)
+
+        if auth_is_new:
+            AuditLogEntry.objects.create(
+                organization=self.organization,
+                actor=self.user,
+                ip_address=self.request.META["REMOTE_ADDR"],
+                target_object=auth_identity.id,
+                event=AuditLogEntryEvent.SSO_IDENTITY_LINK,
+                data=auth_identity.get_audit_log_data(),
+            )
+
+            messages.add_message(self.request, messages.SUCCESS, OK_LINK_IDENTITY)
+
+        return auth_identity
 
+    def _wipe_existing_identity(self, auth_identity: AuthIdentity) -> None:
+        # it's possible the user has an existing identity, let's wipe it out
+        # so that the new identifier gets used (other we'll hit a constraint)
+        # violation since one might exist for (provider, user) as well as
+        # (provider, ident)
+        AuthIdentity.objects.exclude(id=auth_identity.id).filter(
+            auth_provider=self.auth_provider, user=self.user
+        ).delete()
 
-def get_identifier(identity):
-    return identity.get("email") or identity.get("id")
+        # since we've identified an identity which is no longer valid
+        # lets preemptively mark it as such
+        try:
+            other_member = OrganizationMember.objects.get(
+                user=auth_identity.user_id, organization=self.organization
+            )
+        except OrganizationMember.DoesNotExist:
+            return
+        other_member.flags["sso:invalid"] = True
+        other_member.flags["sso:linked"] = False
+        other_member.save()
+
+    def _get_organization_member(self) -> OrganizationMember:
+        try:
+            return OrganizationMember.objects.get(user=self.user, organization=self.organization)
+        except OrganizationMember.DoesNotExist:
+            pass
 
+        member = OrganizationMember.objects.create(
+            organization=self.organization,
+            role=self.organization.default_role,
+            user=self.user,
+            flags=OrganizationMember.flags["sso:linked"],
+        )
 
-def respond(template, organization, request, context=None, status=200):
-    default_context = {"organization": organization}
-    if context:
-        default_context.update(context)
+        default_teams = self.auth_provider.default_teams.all()
+        for team in default_teams:
+            OrganizationMemberTeam.objects.create(team=team, organizationmember=member)
 
-    return render_to_response(template, default_context, request, status=status)
+        AuditLogEntry.objects.create(
+            organization=self.organization,
+            actor=self.user,
+            ip_address=self.request.META["REMOTE_ADDR"],
+            target_object=member.id,
+            target_user=self.user,
+            event=AuditLogEntryEvent.MEMBER_ADD,
+            data=member.get_audit_log_data(),
+        )
+        return member
 
+    def _respond(
+        self,
+        template: str,
+        context: Mapping[str, Any] = None,
+        status: int = 200,
+    ) -> HttpResponse:
+        default_context = {"organization": self.organization}
+        if context:
+            default_context.update(context)
 
-def post_login_redirect(request):
-    response = HttpResponseRedirect(auth.get_login_redirect(request))
+        return render_to_response(template, default_context, self.request, status=status)
 
-    # Always remove any pending invite cookies, pending invites will have been
-    # accepted during the SSO flow.
-    remove_invite_cookie(request, response)
+    def _post_login_redirect(self) -> HttpResponseRedirect:
+        response = HttpResponseRedirect(auth.get_login_redirect(self.request))
 
-    return response
+        # Always remove any pending invite cookies, pending invites will have been
+        # accepted during the SSO flow.
+        remove_invite_cookie(self.request, response)
 
+        return response
 
-def handle_unknown_identity(request, organization, auth_provider, provider, state, identity):
-    """
-    Flow is activated upon a user logging in to where an AuthIdentity is
-    not present.
+    def handle_unknown_identity(
+        self,
+        state: RedisBackedState,
+        identity: Identity,
+    ) -> HttpResponseRedirect:
+        """
+        Flow is activated upon a user logging in to where an AuthIdentity is
+        not present.
 
-    XXX(dcramer): this docstring is out of date
+        XXX(dcramer): this docstring is out of date
 
-    The flow will attempt to answer the following:
+        The flow will attempt to answer the following:
 
-    - Is there an existing user with the same email address? Should they be
-      merged?
+        - Is there an existing user with the same email address? Should they be
+          merged?
 
-    - Is there an existing user (via authentication) that should be merged?
+        - Is there an existing user (via authentication) that should be merged?
 
-    - Should I create a new user based on this identity?
-    """
-    op = request.POST.get("op")
-    if not request.user.is_authenticated:
-        # TODO(dcramer): its possible they have multiple accounts and at
-        # least one is managed (per the check below)
-        try:
-            acting_user = User.objects.filter(
-                id__in=UserEmail.objects.filter(email__iexact=identity["email"]).values("user"),
-                is_active=True,
-            ).first()
-        except IndexError:
-            acting_user = None
-        login_form = AuthenticationForm(
-            request,
-            request.POST if request.POST.get("op") == "login" else None,
-            initial={"username": acting_user.username if acting_user else None},
-        )
-    else:
-        acting_user = request.user
-
-    # If they already have an SSO account and the identity provider says
-    # the email matches we go ahead and let them merge it. This is the
-    # only way to prevent them having duplicate accounts, and because
-    # we trust identity providers, its considered safe.
-    # Note: we do not trust things like SAML, so the SSO implementation needs
-    # to consider if 'email_verified' can be trusted or not
-    if acting_user and identity.get("email_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=organization
-        ).exists()
-        if has_membership:
-            if not auth.login(
-                request,
-                acting_user,
-                after_2fa=request.build_absolute_uri(),
-                organization_id=organization.id,
-            ):
-                if acting_user.has_usable_password():
-                    return post_login_redirect(request)
+        - Should I create a new user based on this identity?
+        """
+        op = self.request.POST.get("op")
+        if not self.user.is_authenticated:
+            # TODO(dcramer): its possible they have multiple accounts and at
+            # least one is managed (per the check below)
+            try:
+                acting_user = User.objects.filter(
+                    id__in=UserEmail.objects.filter(email__iexact=identity["email"]).values("user"),
+                    is_active=True,
+                ).first()
+            except IndexError:
+                acting_user = None
+            login_form = 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},
+            )
+        else:
+            acting_user = self.user
+
+        # If they already have an SSO account and the identity provider says
+        # the email matches we go ahead and let them merge it. This is the
+        # only way to prevent them having duplicate accounts, and because
+        # we trust identity providers, its considered safe.
+        # Note: we do not trust things like SAML, so the SSO implementation needs
+        # to consider if 'email_verified' can be trusted or not
+        if acting_user and identity.get("email_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
+            ).exists()
+            if has_membership:
+                try:
+                    self._login(acting_user)
+                except self._NotCompletedSecurityChecks:
+                    if acting_user.has_usable_password():
+                        return self._post_login_redirect()
+                    else:
+                        acting_user = None
                 else:
-                    acting_user = None
+                    # assume they've confirmed they want to attach the identity
+                    op = "confirm"
             else:
-                # assume they've confirmed they want to attach the identity
-                op = "confirm"
-        else:
-            # force them to create a new account
+                # force them to create a new account
+                acting_user = None
+        # without a usable password they cant login, so let's clear the acting_user
+        elif acting_user and not acting_user.has_usable_password():
             acting_user = None
-    # without a usable password they cant login, so let's clear the acting_user
-    elif acting_user and not acting_user.has_usable_password():
-        acting_user = None
 
-    if op == "confirm" and request.user.is_authenticated:
-        auth_identity = handle_attach_identity(
-            auth_provider, request, organization, provider, identity
-        )
-    elif op == "newuser":
-        auth_identity = handle_new_user(auth_provider, organization, request, identity)
-    elif op == "login" and not request.user.is_authenticated:
-        # confirm authentication, login
-        op = None
-        if login_form.is_valid():
-            # This flow is special.  If we are going through a 2FA
-            # flow here (login returns False) we want to instruct the
-            # system to return upon completion of the 2fa flow to the
-            # current URL and continue with the dialog.
-            #
-            # If there is no 2fa we don't need to do this and can just
-            # go on.
-            if not auth.login(
-                request,
-                login_form.get_user(),
-                after_2fa=request.build_absolute_uri(),
-                organization_id=organization.id,
-            ):
-                return post_login_redirect(request)
+        if op == "confirm" and self.user.is_authenticated:
+            auth_identity = self.handle_attach_identity(identity)
+        elif op == "newuser":
+            auth_identity = self.handle_new_user(identity)
+        elif op == "login" and not self.user.is_authenticated:
+            # confirm authentication, login
+            op = None
+            if login_form.is_valid():
+                # This flow is special.  If we are going through a 2FA
+                # flow here (login returns False) we want to instruct the
+                # system to return upon completion of the 2fa flow to the
+                # current URL and continue with the dialog.
+                #
+                # If there is no 2fa we don't need to do this and can just
+                # go on.
+                try:
+                    self._login(login_form.get_user())
+                except self._NotCompletedSecurityChecks:
+                    return self._post_login_redirect()
+            else:
+                auth.log_auth_failure(self.request, self.request.POST.get("username"))
         else:
-            auth.log_auth_failure(request, request.POST.get("username"))
-    else:
-        op = None
-
-    if not op:
-        # A blank character is needed to prevent the HTML span from collapsing
-        provider_name = auth_provider.get_provider().name if auth_provider else " "
-
-        if request.user.is_authenticated:
-            return respond(
-                "sentry/auth-confirm-link.html",
-                organization,
-                request,
-                {
-                    "identity": identity,
-                    "provider": provider_name,
-                    "existing_user": request.user,
-                    "identity_display_name": get_display_name(identity),
-                    "identity_identifier": get_identifier(identity),
-                },
-            )
+            op = None
 
-        return respond(
-            "sentry/auth-confirm-identity.html",
-            organization,
-            request,
-            {
-                "existing_user": acting_user,
+        if not op:
+            # A blank character is needed to prevent the HTML span from collapsing
+            provider_name = self.auth_provider.get_provider().name if self.auth_provider else " "
+
+            context = {
                 "identity": identity,
                 "provider": provider_name,
-                "login_form": login_form,
-                "identity_display_name": get_display_name(identity),
-                "identity_identifier": get_identifier(identity),
-            },
-        )
-
-    user = auth_identity.user
-    user.backend = settings.AUTHENTICATION_BACKENDS[0]
+                "identity_display_name": identity.get("name") or identity.get("email"),
+                "identity_identifier": identity.get("email") or identity.get("id"),
+            }
+            if self.user.is_authenticated:
+                template = "sentry/auth-confirm-link.html"
+                context.update({"existing_user": self.user})
+            else:
+                template = "sentry/auth-confirm-identity.html"
+                context.update({"existing_user": acting_user, "login_form": login_form})
+            return self._respond(template, context)
 
-    # XXX(dcramer): this is repeated from above
-    if not auth.login(
-        request, user, after_2fa=request.build_absolute_uri(), organization_id=organization.id
-    ):
-        return post_login_redirect(request)
+        user = auth_identity.user
+        user.backend = settings.AUTHENTICATION_BACKENDS[0]
 
-    state.clear()
+        # XXX(dcramer): this is repeated from above
+        try:
+            self._login(user)
+        except self._NotCompletedSecurityChecks:
+            return self._post_login_redirect()
 
-    if not is_active_superuser(request):
-        # set activeorg to ensure correct redirect upon logging in
-        request.session["activeorg"] = organization.slug
-    return post_login_redirect(request)
+        state.clear()
 
+        if not is_active_superuser(self.request):
+            # set activeorg to ensure correct redirect upon logging in
+            self.request.session["activeorg"] = self.organization.slug
+        return self._post_login_redirect()
 
-def handle_new_user(auth_provider, organization, request, identity):
-    user = User.objects.create(
-        username=uuid4().hex, email=identity["email"], name=identity.get("name", "")[:200]
-    )
+    def handle_new_user(self, identity: Identity) -> AuthIdentity:
+        user = User.objects.create(
+            username=uuid4().hex, email=identity["email"], name=identity.get("name", "")[:200]
+        )
 
-    if settings.TERMS_URL and settings.PRIVACY_URL:
-        user.update(flags=F("flags").bitor(User.flags.newsletter_consent_prompt))
+        if settings.TERMS_URL and settings.PRIVACY_URL:
+            user.update(flags=F("flags").bitor(User.flags.newsletter_consent_prompt))
 
-    try:
-        with transaction.atomic():
-            auth_identity = AuthIdentity.objects.create(
-                auth_provider=auth_provider,
-                user=user,
-                ident=identity["id"],
-                data=identity.get("data", {}),
-            )
-    except IntegrityError:
-        auth_identity = AuthIdentity.objects.get(auth_provider=auth_provider, ident=identity["id"])
-        auth_identity.update(user=user, data=identity.get("data", {}))
-
-    user.send_confirm_emails(is_new_user=True)
-    provider = auth_provider.provider if auth_provider else None
-    user_signup.send_robust(
-        sender=handle_new_user, user=user, source="sso", provider=provider, referrer="in-app"
-    )
+        try:
+            with transaction.atomic():
+                auth_identity = AuthIdentity.objects.create(
+                    auth_provider=self.auth_provider,
+                    user=user,
+                    ident=identity["id"],
+                    data=identity.get("data", {}),
+                )
+        except IntegrityError:
+            auth_identity = self._get_auth_identity(ident=identity["id"])
+            auth_identity.update(user=user, data=identity.get("data", {}))
+
+        user.send_confirm_emails(is_new_user=True)
+        provider = self.auth_provider.provider if self.auth_provider else None
+        user_signup.send_robust(
+            sender=self.handle_new_user,
+            user=user,
+            source="sso",
+            provider=provider,
+            referrer="in-app",
+        )
 
-    handle_new_membership(auth_provider, organization, request, auth_identity)
+        self._handle_new_membership(auth_identity)
 
-    return auth_identity
+        return auth_identity
 
 
 class AuthHelper:
@@ -671,8 +731,14 @@ class AuthHelper:
 
         return response
 
+    @property
+    def auth_handler(self):
+        return AuthIdentityHandler(
+            self.auth_provider, self.provider, self.organization, self.request
+        )
+
     @transaction.atomic
-    def _finish_login_pipeline(self, identity):
+    def _finish_login_pipeline(self, identity: Identity):
         """
         The login flow executes both with anonymous and authenticated users.
 
@@ -726,14 +792,7 @@ class AuthHelper:
                         },
                     )
 
-                return handle_unknown_identity(
-                    self.request,
-                    self.organization,
-                    self.auth_provider,
-                    self.provider,
-                    self.state,
-                    identity,
-                )
+                return self.auth_handler.handle_unknown_identity(self.state, identity)
 
             # If the User attached to this AuthIdentity is not active,
             # we want to clobber the old account and take it over, rather than
@@ -742,30 +801,13 @@ class AuthHelper:
                 # Current user is also not logged in, so we have to
                 # assume unknown.
                 if not self.request.user.is_authenticated:
-                    return handle_unknown_identity(
-                        self.request,
-                        self.organization,
-                        self.auth_provider,
-                        self.provider,
-                        self.state,
-                        identity,
-                    )
-                auth_identity = handle_attach_identity(
-                    self.auth_provider, self.request, self.organization, self.provider, identity
-                )
+                    return self.auth_handler.handle_unknown_identity(self.state, identity)
+                auth_identity = self.auth_handler.handle_attach_identity(identity)
 
-            return handle_existing_identity(
-                self.auth_provider,
-                self.provider,
-                self.organization,
-                self.request,
-                self.state,
-                auth_identity,
-                identity,
-            )
+            return self.auth_handler.handle_existing_identity(self.state, auth_identity, identity)
 
     @transaction.atomic
-    def _finish_setup_pipeline(self, identity):
+    def _finish_setup_pipeline(self, identity: Identity):
         """
         The setup flow creates the auth provider as well as an identity linked
         to the active user.
@@ -793,9 +835,7 @@ class AuthHelper:
             organization=self.organization, provider=self.provider.key, config=config
         )
 
-        handle_attach_identity(
-            self.auth_provider, self.request, self.organization, self.provider, identity, om
-        )
+        self.auth_handler.handle_attach_identity(identity, om)
 
         auth.mark_sso_complete(request, self.organization.id)
 

tests/sentry/auth/test_helper.py

@@ -4,15 +4,7 @@ from django.contrib import messages
 from django.contrib.auth.models import AnonymousUser
 from django.test import Client, RequestFactory
 
-from sentry.auth.helper import (
-    OK_LINK_IDENTITY,
-    AuthHelper,
-    RedisBackedState,
-    handle_attach_identity,
-    handle_existing_identity,
-    handle_new_user,
-    handle_unknown_identity,
-)
+from sentry.auth.helper import OK_LINK_IDENTITY, AuthHelper, AuthIdentityHandler, RedisBackedState
 from sentry.auth.provider import Provider
 from sentry.models import (
     AuditLogEntry,
@@ -41,6 +33,8 @@ class AuthIdentityHandlerTest(TestCase):
         self.provider = "dummy"
         self.provider_obj = Provider(self.provider)
         self.request = _set_up_request()
+        self.request.user = AnonymousUser()
+        self.request.session = Client().session
 
         self.auth_provider = AuthProvider.objects.create(
             organization=self.organization, provider=self.provider
@@ -52,6 +46,10 @@ class AuthIdentityHandlerTest(TestCase):
             "data": {"foo": "bar"},
         }
 
+        self.handler = AuthIdentityHandler(
+            self.auth_provider, Provider(self.provider), self.organization, self.request
+        )
+
         self.state = RedisBackedState(self.request)
 
     def set_up_user(self):
@@ -75,13 +73,10 @@ class AuthIdentityHandlerTest(TestCase):
 
 
 class HandleNewUserTest(AuthIdentityHandlerTest):
-    def _handle_new_user(self):
-        return handle_new_user(self.auth_provider, self.organization, self.request, self.identity)
-
     @mock.patch("sentry.analytics.record")
     def test_simple(self, mock_record):
 
-        auth_identity = self._handle_new_user()
+        auth_identity = self.handler.handle_new_user(self.identity)
         user = auth_identity.user
 
         assert user.email == self.identity["email"]
@@ -103,7 +98,7 @@ class HandleNewUserTest(AuthIdentityHandlerTest):
             organization=self.organization, email=self.identity["email"]
         )
 
-        auth_identity = self._handle_new_user()
+        auth_identity = self.handler.handle_new_user(self.identity)
 
         assigned_member = OrganizationMember.objects.get(
             organization=self.organization, user=auth_identity.user
@@ -118,7 +113,7 @@ class HandleNewUserTest(AuthIdentityHandlerTest):
             invite_status=InviteStatus.REQUESTED_TO_BE_INVITED.value,
         )
 
-        auth_identity = self._handle_new_user()
+        auth_identity = self.handler.handle_new_user(self.identity)
 
         assert OrganizationMember.objects.filter(
             organization=self.organization,
@@ -139,7 +134,7 @@ class HandleNewUserTest(AuthIdentityHandlerTest):
             {"memberId": member.id, "token": member.token, "url": ""}
         )
 
-        auth_identity = self._handle_new_user()
+        auth_identity = self.handler.handle_new_user(self.identity)
 
         assigned_member = OrganizationMember.objects.get(
             organization=self.organization, user=auth_identity.user
@@ -154,15 +149,7 @@ class HandleExistingIdentityTest(AuthIdentityHandlerTest):
         mock_auth.get_login_redirect.return_value = "test_login_url"
         user, auth_identity = self.set_up_user_identity()
 
-        redirect = handle_existing_identity(
-            self.auth_provider,
-            self.provider_obj,
-            self.organization,
-            self.request,
-            self.state,
-            auth_identity,
-            self.identity,
-        )
+        redirect = self.handler.handle_existing_identity(self.state, auth_identity, self.identity)
 
         assert redirect.url == mock_auth.get_login_redirect.return_value
         assert mock_auth.get_login_redirect.called_with(self.request)
@@ -180,16 +167,11 @@ class HandleExistingIdentityTest(AuthIdentityHandlerTest):
 
 
 class HandleAttachIdentityTest(AuthIdentityHandlerTest):
-    def _handle_attach_identity(self):
-        return handle_attach_identity(
-            self.auth_provider, self.request, self.organization, self.provider_obj, self.identity
-        )
-
     @mock.patch("sentry.auth.helper.messages")
     def test_new_identity(self, mock_messages):
         self.set_up_user()
 
-        auth_identity = self._handle_attach_identity()
+        auth_identity = self.handler.handle_attach_identity(self.identity)
         assert auth_identity.ident == self.identity["id"]
         assert auth_identity.data == self.identity["data"]
 
@@ -219,7 +201,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
         user = self.set_up_user()
         existing_om = OrganizationMember.objects.create(user=user, organization=self.organization)
 
-        auth_identity = self._handle_attach_identity()
+        auth_identity = self.handler.handle_attach_identity(self.identity)
         assert auth_identity.ident == self.identity["id"]
         assert auth_identity.data == self.identity["data"]
 
@@ -235,7 +217,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
     def test_existing_identity(self, mock_messages):
         user, existing_identity = self.set_up_user_identity()
 
-        returned_identity = self._handle_attach_identity()
+        returned_identity = self.handler.handle_attach_identity(self.identity)
         assert returned_identity == existing_identity
         assert not mock_messages.add_message.called
 
@@ -248,7 +230,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
         )
         OrganizationMember.objects.create(user=other_user, organization=self.organization)
 
-        returned_identity = self._handle_attach_identity()
+        returned_identity = self.handler.handle_attach_identity(self.identity)
         assert returned_identity.user == request_user
         assert returned_identity.ident == self.identity["id"]
         assert returned_identity.data == self.identity["data"]
@@ -271,14 +253,7 @@ class HandleAttachIdentityTest(AuthIdentityHandlerTest):
 
 class HandleUnknownIdentityTest(AuthIdentityHandlerTest):
     def _test_simple(self, mock_render, expected_template):
-        redirect = handle_unknown_identity(
-            self.request,
-            self.organization,
-            self.auth_provider,
-            self.provider_obj,
-            self.state,
-            self.identity,
-        )
+        redirect = self.handler.handle_unknown_identity(self.state, self.identity)
 
         assert redirect is mock_render.return_value
         template, context, request = mock_render.call_args.args