test_api.py 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128
  1. from datetime import datetime, timedelta, timezone
  2. from sentry.models.authidentity import AuthIdentity
  3. from sentry.models.authprovider import AuthProvider
  4. from sentry.silo import SiloMode
  5. from sentry.testutils.cases import AuthProviderTestCase
  6. from sentry.testutils.silo import assume_test_silo_mode
  7. from sentry.testutils.skips import requires_snuba
  8. from sentry.utils.auth import SSO_EXPIRY_TIME, SsoSession
  9. pytestmark = [requires_snuba]
  10. # TODO: move these into the tests/sentry/auth directory and remove deprecated logic
  11. class AuthenticationTest(AuthProviderTestCase):
  12. def setUp(self):
  13. self.organization = self.create_organization(name="foo")
  14. self.user = self.create_user("foobar@example.com", is_superuser=False)
  15. team = self.create_team(name="bar", organization=self.organization)
  16. self.project = self.create_project(name="baz", organization=self.organization, teams=[team])
  17. member = self.create_member(user=self.user, organization=self.organization, teams=[team])
  18. setattr(member.flags, "sso:linked", True)
  19. member.save()
  20. event = self.store_event(data={}, project_id=self.project.id)
  21. group_id = event.group_id
  22. with assume_test_silo_mode(SiloMode.CONTROL):
  23. auth_provider = AuthProvider.objects.create(
  24. organization_id=self.organization.id, provider="dummy", flags=0
  25. )
  26. AuthIdentity.objects.create(auth_provider=auth_provider, user=self.user)
  27. self.login_as(self.user)
  28. self.paths = (
  29. f"/api/0/organizations/{self.organization.slug}/",
  30. f"/api/0/projects/{self.organization.slug}/{self.project.slug}/",
  31. f"/api/0/teams/{self.organization.slug}/{self.team.slug}/",
  32. f"/api/0/issues/{group_id}/",
  33. # this uses the internal API, which once upon a time was broken
  34. f"/api/0/issues/{group_id}/events/latest/",
  35. )
  36. def test_sso_auth_required(self):
  37. # we should be redirecting the user to the authentication form as they
  38. # haven't verified this specific organization
  39. self._test_paths_with_status(401)
  40. def test_sso_superuser_required(self):
  41. with assume_test_silo_mode(SiloMode.CONTROL):
  42. # superuser should still require SSO as they're a member of the org
  43. self.user.update(is_superuser=True)
  44. self._test_paths_with_status(401)
  45. def test_sso_with_expiry_valid(self):
  46. sso_session = SsoSession.create(self.organization.id)
  47. self.session[sso_session.session_key] = sso_session.to_dict()
  48. self.save_session()
  49. self._test_paths_with_status(200)
  50. def test_sso_with_expiry_expired(self):
  51. sso_session_expired = SsoSession(
  52. self.organization.id,
  53. datetime.now(tz=timezone.utc) - SSO_EXPIRY_TIME - timedelta(hours=1),
  54. )
  55. self.session[sso_session_expired.session_key] = sso_session_expired.to_dict()
  56. self.save_session()
  57. self._test_paths_with_status(401)
  58. def test_sso_redirect_url_internal(self):
  59. sso_session_expired = SsoSession(
  60. self.organization.id,
  61. datetime.now(tz=timezone.utc) - SSO_EXPIRY_TIME - timedelta(hours=1),
  62. )
  63. self.session[sso_session_expired.session_key] = sso_session_expired.to_dict()
  64. self.save_session()
  65. resp = self.client.get(
  66. f"/api/0/teams/{self.organization.slug}/{self.team.slug}/",
  67. HTTP_REFERER=f"/organizations/{self.organization.slug}/teams",
  68. )
  69. assert (
  70. resp.json()["detail"]["extra"]["loginUrl"]
  71. == "/auth/login/foo/?next=%2Forganizations%2Ffoo%2Fteams"
  72. )
  73. def test_sso_redirect_url_internal_with_domain(self):
  74. sso_session_expired = SsoSession(
  75. self.organization.id,
  76. datetime.now(tz=timezone.utc) - SSO_EXPIRY_TIME - timedelta(hours=1),
  77. )
  78. self.session[sso_session_expired.session_key] = sso_session_expired.to_dict()
  79. self.save_session()
  80. resp = self.client.get(
  81. f"/api/0/teams/{self.organization.slug}/{self.team.slug}/",
  82. HTTP_REFERER=f"https://testdomain.com/organizations/{self.organization.slug}/teams",
  83. SERVER_NAME="testdomain.com",
  84. )
  85. assert (
  86. resp.json()["detail"]["extra"]["loginUrl"]
  87. == "/auth/login/foo/?next=https%3A%2F%2Ftestdomain.com%2Forganizations%2Ffoo%2Fteams"
  88. )
  89. def test_sso_redirect_url_external_removed(self):
  90. sso_session_expired = SsoSession(
  91. self.organization.id,
  92. datetime.now(tz=timezone.utc) - SSO_EXPIRY_TIME - timedelta(hours=1),
  93. )
  94. self.session[sso_session_expired.session_key] = sso_session_expired.to_dict()
  95. self.save_session()
  96. resp = self.client.get(
  97. f"/api/0/teams/{self.organization.slug}/{self.team.slug}/",
  98. HTTP_REFERER="http://example.com",
  99. )
  100. assert resp.json()["detail"]["extra"]["loginUrl"] == "/auth/login/foo/"
  101. def _test_paths_with_status(self, status):
  102. for path in self.paths:
  103. resp = self.client.get(path)
  104. assert resp.status_code == status, (resp.status_code, resp.content)