|
6 | 6 | from django.contrib.auth.models import AnonymousUser |
7 | 7 | from django.db import IntegrityError, models, router, transaction |
8 | 8 | from django.http import HttpResponseRedirect |
| 9 | +from django.http.response import HttpResponseBase |
9 | 10 | from django.test import Client, RequestFactory |
10 | 11 |
|
11 | 12 | from sentry import audit_log |
@@ -829,6 +830,94 @@ def test_no_provider_key_continues_normally(self, mock_messages: mock.MagicMock) |
829 | 830 | assert "SSO" not in str(call) |
830 | 831 |
|
831 | 832 |
|
| 833 | +@control_silo_test |
| 834 | +class SetupPipelineIdentityLinkingTest(TestCase, HybridCloudTestMixin): |
| 835 | + """Tests that the SSO setup pipeline always links the identity to the authenticated admin.""" |
| 836 | + |
| 837 | + def setUp(self) -> None: |
| 838 | + super().setUp() |
| 839 | + self.provider = "dummy" |
| 840 | + self. admin_user = self. create_user( email="[email protected]") |
| 841 | + self.create_member(organization=self.organization, user=self.admin_user, role="owner") |
| 842 | + |
| 843 | + self.auth_key = "test_auth_key" |
| 844 | + self.request = _set_up_request() |
| 845 | + self.request.user = self.admin_user |
| 846 | + self.request.session["auth_key"] = self.auth_key |
| 847 | + |
| 848 | + def _run_setup_pipeline_with_identity_email(self, identity_email: str) -> HttpResponseBase: |
| 849 | + """Run the setup pipeline with a given email in the IdP assertion.""" |
| 850 | + initial_state = { |
| 851 | + "org_id": self.organization.id, |
| 852 | + "flow": FLOW_SETUP_PROVIDER, |
| 853 | + "provider_model_id": None, |
| 854 | + "provider_key": self.provider, |
| 855 | + "referrer": None, |
| 856 | + } |
| 857 | + local_client = clusters.get("default").get_local_client_for_key(self.auth_key) |
| 858 | + local_client.set(self.auth_key, json.dumps(initial_state)) |
| 859 | + |
| 860 | + helper = AuthHelper.get_for_request(self.request) |
| 861 | + assert helper is not None |
| 862 | + helper.initialize() |
| 863 | + |
| 864 | + helper.bind_state("email", identity_email) |
| 865 | + helper.bind_state("email_verified", True) |
| 866 | + |
| 867 | + helper.state.step_index = len(helper.pipeline_views) |
| 868 | + result = helper.current_step() |
| 869 | + return result |
| 870 | + |
| 871 | + @mock.patch("sentry.auth.helper.messages") |
| 872 | + def test_setup_links_to_admin_when_assertion_email_differs( |
| 873 | + self, mock_messages: mock.MagicMock |
| 874 | + ) -> None: |
| 875 | + """When the IdP assertion email belongs to a different org member, |
| 876 | + the identity is still linked to the admin performing setup.""" |
| 877 | + other_user = self. create_user( email="[email protected]") |
| 878 | + self.create_member(organization=self.organization, user=other_user) |
| 879 | + |
| 880 | + result = self. _run_setup_pipeline_with_identity_email( "[email protected]") |
| 881 | + |
| 882 | + assert result.status_code == 302 |
| 883 | + |
| 884 | + auth_provider = AuthProvider.objects.get( |
| 885 | + organization_id=self.organization.id, provider=self.provider |
| 886 | + ) |
| 887 | + auth_identity = AuthIdentity.objects.get(auth_provider=auth_provider) |
| 888 | + assert auth_identity.user_id == self.admin_user.id |
| 889 | + assert not AuthIdentity.objects.filter(user_id=other_user.id).exists() |
| 890 | + |
| 891 | + @mock.patch("sentry.auth.helper.messages") |
| 892 | + def test_setup_links_to_admin_when_emails_match(self, mock_messages: mock.MagicMock) -> None: |
| 893 | + """Setup succeeds normally when the IdP assertion email matches the admin.""" |
| 894 | + result = self. _run_setup_pipeline_with_identity_email( "[email protected]") |
| 895 | + |
| 896 | + assert result.status_code == 302 |
| 897 | + |
| 898 | + auth_provider = AuthProvider.objects.get( |
| 899 | + organization_id=self.organization.id, provider=self.provider |
| 900 | + ) |
| 901 | + auth_identity = AuthIdentity.objects.get(auth_provider=auth_provider) |
| 902 | + assert auth_identity.user_id == self.admin_user.id |
| 903 | + |
| 904 | + @mock.patch("sentry.auth.helper.messages") |
| 905 | + def test_setup_links_to_admin_when_email_matches_no_user( |
| 906 | + self, mock_messages: mock.MagicMock |
| 907 | + ) -> None: |
| 908 | + """When the IdP returns an email that doesn't match any Sentry user, |
| 909 | + the identity is still linked to the admin.""" |
| 910 | + result = self. _run_setup_pipeline_with_identity_email( "[email protected]") |
| 911 | + |
| 912 | + assert result.status_code == 302 |
| 913 | + |
| 914 | + auth_provider = AuthProvider.objects.get( |
| 915 | + organization_id=self.organization.id, provider=self.provider |
| 916 | + ) |
| 917 | + auth_identity = AuthIdentity.objects.get(auth_provider=auth_provider) |
| 918 | + assert auth_identity.user_id == self.admin_user.id |
| 919 | + |
| 920 | + |
832 | 921 | @control_silo_test |
833 | 922 | class InactiveUserIdentityTest(AuthIdentityHandlerTest): |
834 | 923 | """Tests that inactive-user AuthIdentity always routes through handle_unknown_identity.""" |
|
0 commit comments