fix(hybridcloud) Move SSO setup to control silo (#67468)

Move the HTML and redirect flows for SSO setup to control silo. This
should resolve lost state issues in the setup pipeline that are causing
issues.

Part two of https://github.com/getsentry/sentry/pull/67451

src/sentry/models/authprovider.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Mapping
+from typing import Any
 
 from django.db import models
 from django.utils import timezone
@@ -68,6 +70,20 @@ class AuthProvider(ReplicatedControlModel):
             auth_provider=serialized, region_name=region_name
         )
 
+    @classmethod
+    def handle_async_deletion(
+        cls,
+        identifier: int,
+        region_name: str,
+        shard_identifier: int,
+        payload: Mapping[str, Any] | None,
+    ) -> None:
+        from sentry.services.hybrid_cloud.replica.service import region_replica_service
+
+        region_replica_service.delete_replicated_auth_provider(
+            auth_provider_id=identifier, region_name=region_name
+        )
+
     class flags(TypedClassBitField):
         # WARNING: Only add flags to the bottom of this list
         # bitfield flags are dependent on their order and inserting/removing

src/sentry/models/outbox.py

@@ -81,6 +81,7 @@ class OutboxCategory(IntEnum):
     PROVISION_ORGANIZATION = 17
     POST_ORGANIZATION_PROVISION = 18
     UNUSED_ONE = 19
+    # No longer in use.
     DISABLE_AUTH_PROVIDER = 20
     RESET_IDP_FLAGS = 21
     MARK_INVALID_SSO = 22

src/sentry/receivers/outbox/region.py

@@ -75,5 +75,6 @@ def process_organization_mapping_customer_id_update(
 
 @receiver(process_region_outbox, sender=OutboxCategory.DISABLE_AUTH_PROVIDER)
 def process_disable_auth_provider(object_identifier: int, shard_identifier: int, **kwds: Any):
+    # Deprecated
     auth_service.disable_provider(provider_id=object_identifier)
     AuthProviderReplica.objects.filter(auth_provider_id=object_identifier).delete()

src/sentry/web/frontend/organization_auth_settings.py

@@ -4,8 +4,6 @@ import logging
 
 from django import forms
 from django.contrib import messages
-from django.db import router, transaction
-from django.db.models import F
 from django.http import HttpResponseRedirect
 from django.http.response import HttpResponse, HttpResponseBadRequest, HttpResponseBase
 from django.urls import reverse
@@ -17,14 +15,12 @@ from sentry.auth import manager
 from sentry.auth.helper import AuthHelper
 from sentry.models.authprovider import AuthProvider
 from sentry.models.organization import Organization
-from sentry.models.organizationmember import OrganizationMember
-from sentry.models.outbox import OutboxCategory, OutboxScope, RegionOutbox, outbox_context
 from sentry.plugins.base import Response
 from sentry.services.hybrid_cloud.auth import RpcAuthProvider, auth_service
 from sentry.services.hybrid_cloud.organization import RpcOrganization, organization_service
-from sentry.tasks.auth import email_missing_links, email_unlink_notifications
+from sentry.tasks.auth import email_missing_links
 from sentry.utils.http import absolute_uri
-from sentry.web.frontend.base import OrganizationView, region_silo_view
+from sentry.web.frontend.base import ControlSiloOrganizationView, control_silo_view
 
 ERR_NO_SSO = _("The SSO feature is not enabled for this organization.")
 
@@ -83,8 +79,8 @@ def auth_provider_settings_form(provider, auth_provider, organization, request):
     return form
 
 
-@region_silo_view
-class OrganizationAuthSettingsView(OrganizationView):
+@control_silo_view
+class OrganizationAuthSettingsView(ControlSiloOrganizationView):
     # We restrict auth settings to org:write as it allows a non-owner to
     # escalate members to own by disabling the default role.
     required_scope = "org:write"
@@ -92,33 +88,23 @@ class OrganizationAuthSettingsView(OrganizationView):
     def _disable_provider(
         self, request: Request, organization: RpcOrganization, auth_provider: RpcAuthProvider
     ):
-        with outbox_context(transaction.atomic(router.db_for_write(OrganizationMember))):
-            self.create_audit_entry(
-                request,
-                organization=organization,
-                target_object=auth_provider.id,
-                event=audit_log.get_event_id("SSO_DISABLE"),
-                data=auth_provider.get_audit_log_data(),
-            )
-
-            OrganizationMember.objects.filter(organization_id=organization.id).update(
-                flags=F("flags")
-                .bitand(~OrganizationMember.flags["sso:linked"])
-                .bitand(~OrganizationMember.flags["sso:invalid"])
-            )
-
-            RegionOutbox(
-                shard_scope=OutboxScope.ORGANIZATION_SCOPE,
-                shard_identifier=organization.id,
-                category=OutboxCategory.DISABLE_AUTH_PROVIDER,
-                object_identifier=auth_provider.id,
-            ).save()
-            transaction.on_commit(
-                lambda: email_unlink_notifications.delay(
-                    organization.id, request.user.id, auth_provider.provider
-                ),
-                router.db_for_write(OrganizationMember),
-            )
+        user = request.user
+        sending_email = ""
+        if hasattr(user, "email"):
+            sending_email = user.email
+        organization_service.send_sso_unlink_emails(
+            organization_id=organization.id,
+            sending_user_email=sending_email,
+            provider_key=auth_provider.provider,
+        )
+        auth_service.disable_provider(provider_id=auth_provider.id)
+        self.create_audit_entry(
+            request,
+            organization=organization,
+            target_object=auth_provider.id,
+            event=audit_log.get_event_id("SSO_DISABLE"),
+            data=auth_provider.get_audit_log_data(),
+        )
 
     def handle_existing_provider(
         self, request: Request, organization: RpcOrganization, auth_provider: RpcAuthProvider
@@ -191,11 +177,9 @@ class OrganizationAuthSettingsView(OrganizationView):
                 },
             )
 
-        pending_links_count = OrganizationMember.objects.filter(
-            organization_id=organization.id,
-            flags=F("flags").bitand(~OrganizationMember.flags["sso:linked"]),
-        ).count()
-
+        pending_links_count = organization_service.count_members_without_sso(
+            organization_id=organization.id
+        )
         context = {
             "form": form,
             "pending_links_count": pending_links_count,

static/app/data/controlsiloUrlPatterns.ts

@@ -7,13 +7,14 @@ const patterns: RegExp[] = [
   new RegExp('^remote/github/marketplace/purchase/$'),
   new RegExp('^remote/hubspot/webhook/$'),
   new RegExp('^remote/channel-provision/account/$'),
+  new RegExp('^orgredirect/try-business/$'),
   new RegExp('^orgredirect/'),
   new RegExp('^api/0/staff-auth/$'),
   new RegExp('^api/0/signup/$'),
   new RegExp('^api/0/audit-logs/$'),
   new RegExp('^api/0/_admin/options/$'),
   new RegExp('^api/0/billingadmins/$'),
-  new RegExp('^api/0/superuseradmins/$'),
+  new RegExp('^api/0/employees/$'),
   new RegExp('^api/0/beacons/$'),
   new RegExp('^api/0/beacons/[^/]+/$'),
   new RegExp('^api/0/beacons/[^/]+/checkins/$'),
@@ -92,7 +93,6 @@ const patterns: RegExp[] = [
   new RegExp('^api/0/sentry-apps/[^/]+/avatar/$'),
   new RegExp('^api/0/sentry-apps/[^/]+/api-tokens/$'),
   new RegExp('^api/0/sentry-apps/[^/]+/api-tokens/[^/]+/$'),
-  new RegExp('^api/0/sentry-apps/[^/]+/api-tokens/[^/]+/$'),
   new RegExp('^api/0/sentry-apps/[^/]+/stats/$'),
   new RegExp('^api/0/sentry-apps/[^/]+/publish-request/$'),
   new RegExp('^api/0/sentry-app-installations/[^/]+/$'),
@@ -130,7 +130,6 @@ const patterns: RegExp[] = [
   new RegExp('^api/0/internal/mail/$'),
   new RegExp('^api/0/internal/integration-proxy/$'),
   new RegExp('^api/0/internal/rpc/[^/]+/[^/]+/$'),
-  new RegExp('^api/0/internal/seer-rpc/[^/]+/$'),
   new RegExp('^api/0/internal/feature-flags/$'),
   new RegExp('^api/hooks/mailgun/inbound/'),
   new RegExp('^oauth/authorize/$'),
@@ -151,7 +150,9 @@ const patterns: RegExp[] = [
   new RegExp('^account/user-confirm/[^/]+/$'),
   new RegExp('^account/settings/identities/associate/[^/]+/[^/]+/[^/]+/$'),
   new RegExp('^account/settings/wizard/[^/]+/$'),
+  new RegExp('^settings/organization/auth/configure/$'),
   new RegExp('^disabled-member/'),
+  new RegExp('^organizations/[^/]+/auth/configure/$'),
   new RegExp('^organizations/[^/]+/integrations/[^/]+/setup/$'),
   new RegExp('^organizations/[^/]+/disabled-member/$'),
   new RegExp('^avatar/[^/]+/$'),

tests/sentry/web/frontend/test_auth_saml2.py

@@ -229,9 +229,7 @@ class AuthSAML2Test(AuthProviderTestCase):
 
         data = {"init": True, "provider": self.provider_name}
 
-        with Feature(["organizations:sso-basic", "organizations:sso-saml2"]), assume_test_silo_mode(
-            SiloMode.REGION
-        ):
+        with Feature(["organizations:sso-basic", "organizations:sso-saml2"]):
             setup = self.client.post(self.setup_path, data)
 
         assert setup.status_code == 302

tests/sentry/web/frontend/test_organization_auth_settings.py

@@ -1,6 +1,7 @@
 from unittest.mock import patch
 
 import pytest
+from django.core import mail
 from django.db import models
 from django.urls import reverse
 
@@ -14,6 +15,7 @@ from sentry.auth.providers.saml2.provider import Attributes
 from sentry.models.auditlogentry import AuditLogEntry
 from sentry.models.authidentity import AuthIdentity
 from sentry.models.authprovider import AuthProvider
+from sentry.models.authproviderreplica import AuthProviderReplica
 from sentry.models.integrations.sentry_app_installation_for_provider import (
     SentryAppInstallationForProvider,
 )
@@ -26,21 +28,21 @@ from sentry.silo import SiloMode
 from sentry.testutils.cases import AuthProviderTestCase, PermissionTestCase
 from sentry.testutils.helpers.features import with_feature
 from sentry.testutils.outbox import outbox_runner
-from sentry.testutils.silo import assume_test_silo_mode, region_silo_test
+from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
 from sentry.web.frontend.organization_auth_settings import get_scim_url
 
 
-@region_silo_test
+@control_silo_test
 class OrganizationAuthSettingsPermissionTest(PermissionTestCase):
     def setUp(self):
         super().setUp()
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            self.auth_provider_inst = AuthProvider.objects.create(
-                organization_id=self.organization.id, provider="dummy"
-            )
-            AuthIdentity.objects.create(
-                user=self.user, ident="foo", auth_provider=self.auth_provider_inst
-            )
+
+        self.auth_provider_inst = AuthProvider.objects.create(
+            organization_id=self.organization.id, provider="dummy"
+        )
+        AuthIdentity.objects.create(
+            user=self.user, ident="foo", auth_provider=self.auth_provider_inst
+        )
         self.login_as(self.user, organization_id=self.organization.id)
         self.path = reverse(
             "sentry-organization-auth-provider-settings", args=[self.organization.slug]
@@ -51,13 +53,11 @@ class OrganizationAuthSettingsPermissionTest(PermissionTestCase):
         self.create_member(
             user=user, organization=self.organization, role="owner", teams=[self.team]
         )
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            AuthIdentity.objects.create(
-                user=user, ident="foo2", auth_provider=self.auth_provider_inst
-            )
-        om = OrganizationMember.objects.get(user_id=user.id, organization=self.organization)
-        setattr(om.flags, "sso:linked", True)
-        om.save()
+        AuthIdentity.objects.create(user=user, ident="foo2", auth_provider=self.auth_provider_inst)
+        with assume_test_silo_mode(SiloMode.REGION):
+            om = OrganizationMember.objects.get(user_id=user.id, organization=self.organization)
+            setattr(om.flags, "sso:linked", True)
+            om.save()
         return user
 
     def create_manager_and_attach_identity(self):
@@ -65,13 +65,11 @@ class OrganizationAuthSettingsPermissionTest(PermissionTestCase):
         self.create_member(
             user=user, organization=self.organization, role="manager", teams=[self.team]
         )
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            AuthIdentity.objects.create(
-                user=user, ident="foo3", auth_provider=self.auth_provider_inst
-            )
-        om = OrganizationMember.objects.get(user_id=user.id, organization=self.organization)
-        setattr(om.flags, "sso:linked", True)
-        om.save()
+        AuthIdentity.objects.create(user=user, ident="foo3", auth_provider=self.auth_provider_inst)
+        with assume_test_silo_mode(SiloMode.REGION):
+            om = OrganizationMember.objects.get(user_id=user.id, organization=self.organization)
+            setattr(om.flags, "sso:linked", True)
+            om.save()
         return user
 
     def test_teamless_admin_cannot_load(self):
@@ -112,22 +110,22 @@ class OrganizationAuthSettingsPermissionTest(PermissionTestCase):
             assert resp.status_code == 200
 
 
-@region_silo_test
+@control_silo_test
 class OrganizationAuthSettingsTest(AuthProviderTestCase):
     def enroll_user_and_require_2fa(self, user, organization):
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            TotpInterface().enroll(user)
-        organization.update(flags=models.F("flags").bitor(Organization.flags.require_2fa))
+        TotpInterface().enroll(user)
+        with assume_test_silo_mode(SiloMode.REGION):
+            organization.update(flags=models.F("flags").bitor(Organization.flags.require_2fa))
         assert organization.flags.require_2fa.is_set
 
     def assert_require_2fa_disabled(self, user, organization, logger):
-        organization = Organization.objects.get(id=organization.id)
-        assert not organization.flags.require_2fa.is_set
+        with assume_test_silo_mode(SiloMode.REGION):
+            organization = Organization.objects.get(id=organization.id)
+            assert not organization.flags.require_2fa.is_set
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            event = AuditLogEntry.objects.get(
-                target_object=organization.id, event=audit_log.get_event_id("ORG_EDIT"), actor=user
-            )
+        event = AuditLogEntry.objects.get(
+            target_object=organization.id, event=audit_log.get_event_id("ORG_EDIT"), actor=user
+        )
         audit_log_event = audit_log.get(event.event)
         assert "require_2fa to False when enabling SSO" in audit_log_event.render(event)
         logger.info.assert_called_once_with(
@@ -145,9 +143,8 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
             assert resp.status_code == 200
             assert PLACEHOLDER_TEMPLATE in resp.content.decode("utf-8")
 
-            with assume_test_silo_mode(SiloMode.CONTROL):
-                path = reverse("sentry-auth-sso")
-                resp = self.client.post(path, {"email": user.email})
+            path = reverse("sentry-auth-sso")
+            resp = self.client.post(path, {"email": user.email})
 
         settings_path = reverse("sentry-organization-auth-settings", args=[organization.slug])
 
@@ -157,38 +154,36 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
         else:
             self.assertRedirects(resp, configure_path)
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.get(
-                organization_id=organization.id, provider="dummy"
-            )
-            auth_identity = AuthIdentity.objects.get(auth_provider=auth_provider)
-            assert user == auth_identity.user
+        auth_provider = AuthProvider.objects.get(organization_id=organization.id, provider="dummy")
+        auth_identity = AuthIdentity.objects.get(auth_provider=auth_provider)
+        assert user == auth_identity.user
 
-        member = OrganizationMember.objects.get(organization=organization, user_id=user.id)
+        with assume_test_silo_mode(SiloMode.REGION):
+            member = OrganizationMember.objects.get(organization=organization, user_id=user.id)
 
-        assert getattr(member.flags, "sso:linked")
-        assert not getattr(member.flags, "sso:invalid")
+            assert getattr(member.flags, "sso:linked")
+            assert not getattr(member.flags, "sso:invalid")
 
     def create_org_and_auth_provider(self, provider_name="dummy"):
         if provider_name == "Fly.io":
             auth.register("Fly.io", FlyOAuth2Provider)
             self.addCleanup(auth.unregister, "Fly.io", FlyOAuth2Provider)
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            self.user.update(is_managed=True)
-        organization = self.create_organization(name="foo", owner=self.user)
+        self.user.update(is_managed=True)
+        with assume_test_silo_mode(SiloMode.REGION):
+            organization = self.create_organization(name="foo", owner=self.user)
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.create(
-                organization_id=organization.id, provider=provider_name
-            )
-            AuthIdentity.objects.create(user=self.user, ident="foo", auth_provider=auth_provider)
+        auth_provider = AuthProvider.objects.create(
+            organization_id=organization.id, provider=provider_name
+        )
+        AuthIdentity.objects.create(user=self.user, ident="foo", auth_provider=auth_provider)
         return organization, auth_provider
 
     def create_om_and_link_sso(self, organization):
-        om = OrganizationMember.objects.get(user_id=self.user.id, organization=organization)
-        setattr(om.flags, "sso:linked", True)
-        om.save()
+        with assume_test_silo_mode(SiloMode.REGION):
+            om = OrganizationMember.objects.get(user_id=self.user.id, organization=organization)
+            setattr(om.flags, "sso:linked", True)
+            om.save()
         return om
 
     def test_can_start_auth_flow(self):
@@ -224,11 +219,10 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
         self.login_as(user)
         self.assert_basic_flow(user, organization)
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            # disable require 2fa logs not called
-            assert not AuditLogEntry.objects.filter(
-                target_object=organization.id, event=audit_log.get_event_id("ORG_EDIT"), actor=user
-            ).exists()
+        # disable require 2fa logs not called
+        assert not AuditLogEntry.objects.filter(
+            target_object=organization.id, event=audit_log.get_event_id("ORG_EDIT"), actor=user
+        ).exists()
         assert not logger.info.called
 
     @with_feature("organizations:customer-domains")
@@ -270,34 +264,43 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
         self.assert_basic_flow(user, organization)
         self.assert_require_2fa_disabled(user, organization, logger)
 
-    @patch("sentry.web.frontend.organization_auth_settings.email_unlink_notifications")
-    def test_disable_provider(self, email_unlink_notifications):
+    def test_disable_provider(self):
         organization, auth_provider = self.create_org_and_auth_provider()
         om = self.create_om_and_link_sso(organization)
         path = reverse("sentry-organization-auth-provider-settings", args=[organization.slug])
+        with assume_test_silo_mode(SiloMode.REGION):
+            assert AuthProviderReplica.objects.filter(organization_id=organization.id).exists()
 
         self.login_as(self.user, organization_id=organization.id)
 
-        with self.feature("organizations:sso-basic"):
+        with self.tasks(), self.feature("organizations:sso-basic"):
             resp = self.client.post(path, {"op": "disable"})
 
         assert resp.status_code == 302
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert not AuthProvider.objects.filter(organization_id=organization.id).exists()
-            assert not AuthProvider.objects.filter(id=auth_provider.id).exists()
+        assert not AuthProvider.objects.filter(organization_id=organization.id).exists()
+        assert not AuthProvider.objects.filter(id=auth_provider.id).exists()
+        assert AuditLogEntry.objects.filter(event=audit_log.get_event_id("SSO_DISABLE")).exists()
 
-        om = OrganizationMember.objects.get(id=om.id)
+        with assume_test_silo_mode(SiloMode.REGION):
+            om = OrganizationMember.objects.get(id=om.id)
 
+        # No more linked members, users are not managed either.
         assert not getattr(om.flags, "sso:linked")
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert not User.objects.get(id=om.user_id).is_managed
+        assert not User.objects.get(id=om.user_id).is_managed
 
-        assert email_unlink_notifications.delay.called
+        # Replica record should be removed too
+        with assume_test_silo_mode(SiloMode.REGION):
+            assert not AuthProviderReplica.objects.filter(organization_id=organization.id).exists()
+
+        # We should send emails about SSO changes
+        assert len(mail.outbox) == 1
+        message = mail.outbox[0]
+        assert "Action Required" in message.subject
+        assert "Single Sign-On has been disabled" in message.body
 
-    @patch("sentry.web.frontend.organization_auth_settings.email_unlink_notifications")
     @with_feature("organizations:sso-basic")
-    def test_disable_partner_provider(self, email_unlink_notifications):
+    def test_disable_partner_provider(self):
         organization, auth_provider = self.create_org_and_auth_provider("Fly.io")
         self.create_om_and_link_sso(organization)
         path = reverse("sentry-organization-auth-provider-settings", args=[organization.slug])
@@ -309,18 +312,15 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
 
     def test_disable__scim_missing(self):
         organization, auth_provider = self.create_org_and_auth_provider()
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider.flags.scim_enabled = True
-            auth_provider.save()
+        auth_provider.flags.scim_enabled = True
+        auth_provider.save()
 
         member = self.create_om_and_link_sso(organization)
-        member.flags["idp:provisioned"] = True
-        member.save()
+        with assume_test_silo_mode(SiloMode.REGION):
+            member.flags["idp:provisioned"] = True
+            member.save()
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert not SentryAppInstallationForProvider.objects.filter(
-                provider=auth_provider
-            ).exists()
+        assert not SentryAppInstallationForProvider.objects.filter(provider=auth_provider).exists()
 
         path = reverse("sentry-organization-auth-provider-settings", args=[organization.slug])
         self.login_as(self.user, organization_id=organization.id)
@@ -332,14 +332,13 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
         assert resp.redirect_chain == [
             ("/settings/foo/auth/", 302),
         ]
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert not AuthProvider.objects.filter(organization_id=organization.id).exists()
+        assert not AuthProvider.objects.filter(organization_id=organization.id).exists()
 
-        member.refresh_from_db()
-        assert not member.flags["idp:provisioned"], "member should not be idp controlled now"
+        with assume_test_silo_mode(SiloMode.REGION):
+            member.refresh_from_db()
+            assert not member.flags["idp:provisioned"], "member should not be idp controlled now"
 
-    @patch("sentry.web.frontend.organization_auth_settings.email_unlink_notifications")
-    def test_superuser_disable_provider(self, email_unlink_notifications):
+    def test_superuser_disable_provider(self):
         organization, auth_provider = self.create_org_and_auth_provider()
         with self.feature("organizations:sso-scim"), assume_test_silo_mode(SiloMode.CONTROL):
             auth_provider.enable_scim(self.user)
@@ -351,26 +350,24 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
         superuser = self.create_user(is_superuser=True)
         self.login_as(superuser, superuser=True)
 
-        with self.feature({"organizations:sso-basic": False}):
+        with self.feature({"organizations:sso-basic": False}), self.tasks():
             resp = self.client.post(path, {"op": "disable"})
 
         assert resp.status_code == 302
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert not AuthProvider.objects.filter(organization_id=organization.id).exists()
-            assert not AuthProvider.objects.filter(id=auth_provider.id).exists()
+        assert not AuthProvider.objects.filter(organization_id=organization.id).exists()
+        assert not AuthProvider.objects.filter(id=auth_provider.id).exists()
+        assert AuditLogEntry.objects.filter(event=audit_log.get_event_id("SSO_DISABLE")).exists()
 
-        om = OrganizationMember.objects.get(id=om.id)
+        with assume_test_silo_mode(SiloMode.REGION):
+            om = OrganizationMember.objects.get(id=om.id)
 
         assert not getattr(om.flags, "sso:linked")
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert not User.objects.get(id=om.user_id).is_managed
+        assert not User.objects.get(id=om.user_id).is_managed
 
-        assert email_unlink_notifications.delay.called
+        assert len(mail.outbox)
 
-        with pytest.raises(SentryAppInstallationForProvider.DoesNotExist), assume_test_silo_mode(
-            SiloMode.CONTROL
-        ):
+        with pytest.raises(SentryAppInstallationForProvider.DoesNotExist):
             SentryAppInstallationForProvider.objects.get(
                 organization_id=self.organization.id, provider="dummy_scim"
             )
@@ -391,19 +388,19 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
 
         assert resp.status_code == 200
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.get(organization_id=organization.id)
+        auth_provider = AuthProvider.objects.get(organization_id=organization.id)
         assert getattr(auth_provider.flags, "allow_unlinked")
-        organization = Organization.objects.get(id=organization.id)
-        assert organization.default_role == "owner"
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            result = AuditLogEntry.objects.filter(
-                organization_id=organization.id,
-                target_object=auth_provider.id,
-                event=audit_log.get_event_id("SSO_EDIT"),
-                actor=self.user,
-            ).first()
+        with assume_test_silo_mode(SiloMode.REGION):
+            organization = Organization.objects.get(id=organization.id)
+            assert organization.default_role == "owner"
+
+        result = AuditLogEntry.objects.filter(
+            organization_id=organization.id,
+            target_object=auth_provider.id,
+            event=audit_log.get_event_id("SSO_EDIT"),
+            actor=self.user,
+        ).first()
 
         assert result.data == {"require_link": "to False", "default_role": "to owner"}
 
@@ -423,19 +420,18 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
 
         assert resp.status_code == 200
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.get(organization_id=organization.id)
+        auth_provider = AuthProvider.objects.get(organization_id=organization.id)
         assert getattr(auth_provider.flags, "allow_unlinked")
-        organization = Organization.objects.get(id=organization.id)
-        assert organization.default_role == "member"
+        with assume_test_silo_mode(SiloMode.REGION):
+            organization = Organization.objects.get(id=organization.id)
+            assert organization.default_role == "member"
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            result = AuditLogEntry.objects.filter(
-                organization_id=organization.id,
-                target_object=auth_provider.id,
-                event=audit_log.get_event_id("SSO_EDIT"),
-                actor=self.user,
-            ).first()
+        result = AuditLogEntry.objects.filter(
+            organization_id=organization.id,
+            target_object=auth_provider.id,
+            event=audit_log.get_event_id("SSO_EDIT"),
+            actor=self.user,
+        ).first()
 
         assert result.data == {"require_link": "to False"}
 
@@ -455,20 +451,18 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
 
         assert resp.status_code == 200
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.get(organization_id=organization.id)
+        auth_provider = AuthProvider.objects.get(organization_id=organization.id)
         assert not getattr(auth_provider.flags, "allow_unlinked")
-        organization = Organization.objects.get(id=organization.id)
-        assert organization.default_role == "owner"
-
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            result = AuditLogEntry.objects.filter(
-                organization_id=organization.id,
-                target_object=auth_provider.id,
-                event=audit_log.get_event_id("SSO_EDIT"),
-                actor=self.user,
-            ).first()
-
+        with assume_test_silo_mode(SiloMode.REGION):
+            organization = Organization.objects.get(id=organization.id)
+            assert organization.default_role == "owner"
+
+        result = AuditLogEntry.objects.filter(
+            organization_id=organization.id,
+            target_object=auth_provider.id,
+            event=audit_log.get_event_id("SSO_EDIT"),
+            actor=self.user,
+        ).first()
         assert result.data == {"default_role": "to owner"}
 
     def test_edit_sso_settings__no_change(self):
@@ -487,16 +481,15 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
 
         assert resp.status_code == 200
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.get(organization_id=organization.id)
+        auth_provider = AuthProvider.objects.get(organization_id=organization.id)
         assert not getattr(auth_provider.flags, "allow_unlinked")
-        organization = Organization.objects.get(id=organization.id)
-        assert organization.default_role == "member"
+        with assume_test_silo_mode(SiloMode.REGION):
+            organization = Organization.objects.get(id=organization.id)
+            assert organization.default_role == "member"
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert not AuditLogEntry.objects.filter(
-                organization_id=organization.id, event=audit_log.get_event_id("SSO_EDIT")
-            ).exists()
+        assert not AuditLogEntry.objects.filter(
+            organization_id=organization.id, event=audit_log.get_event_id("SSO_EDIT")
+        ).exists()
 
     def test_edit_sso_settings__scim(self):
         organization, auth_provider = self.create_org_and_auth_provider()
@@ -520,32 +513,36 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
 
         assert resp.status_code == 200
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.get(organization_id=organization.id)
+        auth_provider = AuthProvider.objects.get(organization_id=organization.id)
         assert getattr(auth_provider.flags, "scim_enabled")
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            assert auth_provider.get_scim_token() is not None
+        assert auth_provider.get_scim_token() is not None
+
         org_member = organization_service.get_organization_by_id(id=auth_provider.organization_id)
         assert org_member is not None
         assert get_scim_url(auth_provider, org_member.organization) is not None
 
         # "add" some scim users
         u1 = self.create_user()
-        not_scim_member = OrganizationMember.objects.create(
-            user_id=u1.id, organization=organization
-        )
-        not_scim_member.save()
         u2 = self.create_user()
-        scim_member = OrganizationMember.objects.create(user_id=u2.id, organization=organization)
-        scim_member.flags["idp:provisioned"] = True
-        scim_member.save()
         u3 = self.create_user()
-        scim_role_restricted_user = OrganizationMember.objects.create(
-            user_id=u3.id, organization=organization
-        )
-        scim_role_restricted_user.flags["idp:provisioned"] = True
-        scim_role_restricted_user.flags["idp:role-restricted"] = True
-        scim_role_restricted_user.save()
+        with assume_test_silo_mode(SiloMode.REGION):
+            not_scim_member = OrganizationMember.objects.create(
+                user_id=u1.id, organization=organization
+            )
+            not_scim_member.save()
+
+            scim_member = OrganizationMember.objects.create(
+                user_id=u2.id, organization=organization
+            )
+            scim_member.flags["idp:provisioned"] = True
+            scim_member.save()
+
+            scim_role_restricted_user = OrganizationMember.objects.create(
+                user_id=u3.id, organization=organization
+            )
+            scim_role_restricted_user.flags["idp:provisioned"] = True
+            scim_role_restricted_user.flags["idp:role-restricted"] = True
+            scim_role_restricted_user.save()
 
         with self.feature({"organizations:sso-basic": True}):
             resp = self.client.post(
@@ -559,22 +556,20 @@ class OrganizationAuthSettingsTest(AuthProviderTestCase):
             )
 
         assert resp.status_code == 200
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            auth_provider = AuthProvider.objects.get(organization_id=organization.id)
+        auth_provider = AuthProvider.objects.get(organization_id=organization.id)
 
         assert not getattr(auth_provider.flags, "scim_enabled")
         org_member = organization_service.get_organization_by_id(id=auth_provider.organization_id)
         assert org_member is not None
         assert get_scim_url(auth_provider, org_member.organization) is None
-        with assume_test_silo_mode(SiloMode.CONTROL), pytest.raises(
-            SentryAppInstallationForProvider.DoesNotExist
-        ):
+        with pytest.raises(SentryAppInstallationForProvider.DoesNotExist):
             SentryAppInstallationForProvider.objects.get(
                 organization_id=self.organization.id, provider="dummy_scim"
             )
-        not_scim_member.refresh_from_db()
-        scim_member.refresh_from_db()
-        scim_role_restricted_user.refresh_from_db()
+        with assume_test_silo_mode(SiloMode.REGION):
+            not_scim_member.refresh_from_db()
+            scim_member.refresh_from_db()
+            scim_role_restricted_user.refresh_from_db()
         assert not any(
             (not_scim_member.flags["idp:provisioned"], not_scim_member.flags["idp:role-restricted"])
         )
@@ -615,7 +610,7 @@ class DummySAML2Provider(GenericSAML2Provider):
         return dummy_provider_config
 
 
-@region_silo_test
+@control_silo_test
 class OrganizationAuthSettingsSAML2Test(AuthProviderTestCase):
     provider = DummySAML2Provider
     provider_name = "saml2_dummy"
@@ -624,12 +619,11 @@ class OrganizationAuthSettingsSAML2Test(AuthProviderTestCase):
         super().setUp()
         self.user = self.create_user("foobar@sentry.io")
         self.organization = self.create_organization(owner=self.user, name="saml2-org")
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            self.auth_provider_inst = AuthProvider.objects.create(
-                provider=self.provider_name,
-                config=dummy_provider_config,
-                organization_id=self.organization.id,
-            )
+        self.auth_provider_inst = AuthProvider.objects.create(
+            provider=self.provider_name,
+            config=dummy_provider_config,
+            organization_id=self.organization.id,
+        )
 
     def test_update_generic_saml2_config(self):
         self.login_as(self.user, organization_id=self.organization.id)
@@ -659,10 +653,9 @@ class OrganizationAuthSettingsSAML2Test(AuthProviderTestCase):
         resp = self.client.post(configure_path, payload)
         assert resp.status_code == 200
 
-        with assume_test_silo_mode(SiloMode.CONTROL):
-            actual = AuthProvider.objects.get(id=self.auth_provider_inst.id)
-            assert actual.config == expected_provider_config
-            assert actual.config != self.auth_provider_inst.config
+        actual = AuthProvider.objects.get(id=self.auth_provider_inst.id)
+        assert actual.config == expected_provider_config
+        assert actual.config != self.auth_provider_inst.config
 
-            assert actual.provider == self.auth_provider_inst.provider
-            assert actual.flags == self.auth_provider_inst.flags
+        assert actual.provider == self.auth_provider_inst.provider
+        assert actual.flags == self.auth_provider_inst.flags