ref(auth): Extract reusable methods from SSO identity handler

src/sentry/auth/email_verification.py

@@ -3,6 +3,7 @@
 import binascii
 import logging
 import time
+from collections.abc import Mapping
 from typing import Any
 
 from django.conf import settings
@@ -17,6 +18,8 @@
 from sentry.utils.signing import sign, unsign
 
 logger = logging.getLogger("sentry.auth.email_verification")
+TRUSTED_EMAIL_VERIFIED_PROVIDERS = frozenset(("github", "google"))
+DEFAULT_MAX_AGE_MINUTES = 120
 
 
 class SignupLinkExpired(SignatureExpired):
@@ -30,7 +33,13 @@ def hash_email(email: str) -> str:
     return sha256_text(email.lower()).hexdigest()
 
 
-DEFAULT_MAX_AGE_MINUTES = 120
+def is_email_verified_by_trusted_provider(provider_key: str, identity: Mapping[str, Any]) -> bool:
+    """If True: the provider's email_verified claim can be trusted, and they certify that the email is verified.
+    If False: the identity requires our own verification step.
+    """
+    return (
+        provider_key in TRUSTED_EMAIL_VERIFIED_PROVIDERS and identity.get("email_verified") is True
+    )
 
 
 def send_signup_verification_email(

src/sentry/auth/exceptions.py

@@ -13,3 +13,14 @@ class IdentityNotValid(Exception):
 
 class AuthIdentityUserMismatch(Exception):
     pass
+
+
+class PipelineStateExpired(Exception):
+    pass
+
+
+class ProviderMismatch(Exception):
+    def __init__(self, actual: str, expected: str) -> None:
+        self.actual = actual
+        self.expected = expected
+        super().__init__(f"Expected provider {expected}, got {actual}")

src/sentry/auth/helper.py

@@ -25,7 +25,12 @@
 from sentry.api.invite_helper import ApiInviteHelper, remove_invite_details_from_session
 from sentry.audit_log.services.log import AuditLogEvent, log_service
 from sentry.auth.email import AmbiguousUserFromEmail, resolve_email_to_user
-from sentry.auth.exceptions import AuthIdentityUserMismatch, IdentityNotValid
+from sentry.auth.exceptions import (
+    AuthIdentityUserMismatch,
+    IdentityNotValid,
+    PipelineStateExpired,
+    ProviderMismatch,
+)
 from sentry.auth.idpmigration import (
     SSO_VERIFICATION_KEY,
     get_verification_value_from_key,
@@ -454,9 +459,11 @@ def _respond(
 
         return render_to_response(template, default_context, self.request, status=status)
 
-    def _post_login_redirect(self) -> HttpResponseRedirect:
+    def _post_login_redirect(self, is_new_user: bool | None = None) -> HttpResponseRedirect:
         url = auth.get_login_redirect(self.request)
-        if self.request.POST.get("op") == "newuser":
+        if is_new_user is None:
+            is_new_user = self.request.POST.get("op") == "newuser"
+        if is_new_user:
             # add events that we can handle on the front end
             provider = self.auth_provider.provider if self.auth_provider else None
             params = {
@@ -654,20 +661,28 @@ def handle_unknown_identity(
             )
             return self._build_confirmation_response(is_new_account)
 
+        return self.complete_and_login(auth_identity, state, is_new_user=(op == "newuser"))
+
+    def complete_and_login(
+        self,
+        auth_identity: AuthIdentity,
+        state: AuthHelperSessionStore,
+        is_new_user: bool = False,
+    ) -> HttpResponseRedirect:
+        """Log in a user after identity resolution and clean up pipeline state."""
         user = auth_identity.user
         user.backend = settings.AUTHENTICATION_BACKENDS[0]
 
-        # XXX(dcramer): this is repeated from above
         try:
             self._login(user)
         except self._NotCompletedSecurityChecks:
-            return self._post_login_redirect()
+            return self._post_login_redirect(is_new_user=is_new_user)
 
         state.clear()
 
         if not is_active_superuser(self.request):
             auth.set_active_org(self.request, self.organization.slug)
-        return self._post_login_redirect()
+        return self._post_login_redirect(is_new_user=is_new_user)
 
     @property
     def provider_name(self) -> str:
@@ -696,7 +711,7 @@ def _dispatch_to_confirmation(
         self.request.session.set_test_cookie()
         return None if is_new_account else self.user, "auth-confirm-identity"
 
-    def handle_new_user(self) -> AuthIdentity:
+    def handle_new_user(self, skip_confirm_emails: bool = False) -> AuthIdentity:
         user = User.objects.create(
             username=uuid4().hex,
             email=self.identity["email"],
@@ -720,7 +735,8 @@ def handle_new_user(self) -> AuthIdentity:
             )
             auth_identity.update(user=user, data=self.identity.get("data", {}))
 
-        user.send_confirm_emails(is_new_user=True)
+        if not skip_confirm_emails:
+            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,
@@ -839,31 +855,41 @@ def get_initial_state(self) -> Mapping[str, Any]:
         state.update({"flow": self.flow, "referrer": self.referrer})
         return state
 
-    def finish_pipeline(self) -> HttpResponseBase:
-        data = self.fetch_state()
+    def resolve_identity(self) -> Mapping[str, Any]:
+        """Fetch pipeline state, validate provider, and build identity.
 
-        # The state data may have expired, in which case the state data will
-        # simply be None.
+        Raises PipelineStateExpired if state is missing/expired,
+        ProviderMismatch if the pipeline provider doesn't match the org's auth provider,
+        IdentityNotValid if the provider can't build a valid identity.
+        """
+        data: Mapping[str, Any] | None = self.fetch_state()
         if not data:
-            return self.error(ERR_INVALID_IDENTITY)
+            raise PipelineStateExpired()
 
-        # Check for provider mismatch - user authenticated with a different provider
-        # than what the organization requires. This can happen when a user has multiple
-        # SSO sessions in different tabs and completes the wrong one.
+        # Check for provider mismatch — user authenticated with a different
+        # provider than what the organization requires.  Can happen when a user
+        # has multiple SSO sessions in different tabs.
         provider_key = data.get("provider_key")
         if (
             self.state.flow == FLOW_LOGIN
             and self.provider_model
             and provider_key
             and provider_key != self.provider_model.provider
         ):
-            return self._handle_provider_mismatch(
-                provider_key=provider_key,
-                expected_provider_key=self.provider_model.provider,
-            )
+            raise ProviderMismatch(actual=provider_key, expected=self.provider_model.provider)
+
+        return self.provider.build_identity(data)
 
+    def finish_pipeline(self) -> HttpResponseBase:
         try:
-            identity = self.provider.build_identity(data)
+            identity = self.resolve_identity()
+        except PipelineStateExpired:
+            return self.error(ERR_INVALID_IDENTITY)
+        except ProviderMismatch as e:
+            return self._handle_provider_mismatch(
+                provider_key=e.actual,
+                expected_provider_key=e.expected,
+            )
         except IdentityNotValid as error:
             return self.error(str(error) or ERR_INVALID_IDENTITY)
 

tests/sentry/auth/test_email_verification.py

@@ -6,6 +6,7 @@
 from django.core.signing import BadSignature, SignatureExpired
 
 from sentry.auth.email_verification import (
+    is_email_verified_by_trusted_provider,
     send_signup_verification_email,
     verify_signup_link,
 )
@@ -88,3 +89,20 @@ def test_wrong_salt(self) -> None:
         signed = sign(salt="wrong-salt", email="a@b.com", expires_at=exp)
         with pytest.raises(BadSignature):
             verify_signup_link(signed)
+
+
+@control_silo_test
+class IsEmailVerifiedByTrustedProviderTest(TestCase):
+    def test_trusted_check(self) -> None:
+        cases = [
+            ("google", {"email_verified": True}, True),
+            ("github", {"email_verified": True}, True),
+            ("google", {"email_verified": False}, False),
+            ("google", {}, False),
+            ("saml2", {"email_verified": True}, False),
+            ("okta", {"email_verified": False}, False),
+        ]
+        for provider_key, identity, expected in cases:
+            assert is_email_verified_by_trusted_provider(provider_key, identity) is expected, (
+                f"{provider_key=} {identity=}"
+            )

tests/sentry/auth/test_helper.py

@@ -37,6 +37,7 @@
 from sentry.testutils.helpers.options import override_options
 from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
 from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
+from sentry.users.models.user import User
 from sentry.utils import json
 from sentry.utils.redis import clusters
 
@@ -171,6 +172,12 @@ def test_unauthenticated_with_no_email_match_resolves_to_anonymous(self) -> None
 
 @control_silo_test
 class HandleNewUserTest(AuthIdentityHandlerTest, HybridCloudTestMixin):
+    @mock.patch("sentry.analytics.record")
+    def test_skip_confirm_emails_suppresses_email(self, mock_record: mock.MagicMock) -> None:
+        with mock.patch.object(User, "send_confirm_emails") as mock_send:
+            self.handler.handle_new_user(skip_confirm_emails=True)
+        mock_send.assert_not_called()
+
     @mock.patch("sentry.analytics.record")
     def test_simple(self, mock_record: mock.MagicMock) -> None:
         auth_identity = self.handler.handle_new_user()