feat(hc): Adds organization region endpoint (#67496)

src/sentry/api/endpoints/organization_region.py

@@ -0,0 +1,76 @@
+from typing import Any
+
+from rest_framework.request import Request
+from rest_framework.response import Response
+from sentry_sdk import capture_message
+
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import Endpoint, control_silo_endpoint
+from sentry.api.exceptions import ResourceDoesNotExist
+from sentry.api.permissions import SentryPermission
+from sentry.models.organizationmapping import OrganizationMapping
+from sentry.models.organizationmembermapping import OrganizationMemberMapping
+from sentry.types.region import get_region_by_name
+
+
+class OrganizationRegionEndpointPermissions(SentryPermission):
+    # Although this permission set is a bit weird, we need to have
+    # project:read for integration auth tokens, org:ci for org auth tokens
+    # and org:read for user auth tokens.
+    scope_map = {"GET": ["project:read", "org:ci", "org:read"]}
+
+    def has_object_permission(self, request, view, org_mapping: OrganizationMapping):
+        if request.auth is None or request.auth.organization_id is None:
+            if request.user.is_anonymous:
+                capture_message("Anonymous user missing auth found in object permission check")
+                return False
+
+            try:
+                OrganizationMemberMapping.objects.get(
+                    user_id=request.user.id, organization_id=org_mapping.organization_id
+                )
+                return True
+            except OrganizationMemberMapping.DoesNotExist:
+                return False
+
+        is_org_or_api_token = (
+            request.auth.kind == "org_auth_token" or request.auth.kind == "api_token"
+        )
+
+        if is_org_or_api_token and request.auth.organization_id == org_mapping.organization_id:
+            return True
+
+        return False
+
+
+@control_silo_endpoint
+class OrganizationRegionEndpoint(Endpoint):
+    owner = ApiOwner.HYBRID_CLOUD
+    publish_status = {
+        "GET": ApiPublishStatus.PRIVATE,
+    }
+    permission_classes = (OrganizationRegionEndpointPermissions,)
+
+    def convert_args(
+        self, request: Request, organization_slug: str | None = None, *args: Any, **kwargs: Any
+    ) -> tuple[tuple[Any, ...], dict[str, Any]]:
+        if not organization_slug:
+            raise ResourceDoesNotExist
+
+        try:
+            org_mapping: OrganizationMapping = OrganizationMapping.objects.get(
+                slug=organization_slug
+            )
+        except OrganizationMapping.DoesNotExist:
+            raise ResourceDoesNotExist
+
+        self.check_object_permissions(request, org_mapping)
+
+        kwargs["organization_mapping"] = org_mapping
+        return (args, kwargs)
+
+    def get(self, request: Request, organization_mapping: OrganizationMapping) -> Response:
+        region_data = get_region_by_name(organization_mapping.region_name)
+
+        return self.respond(region_data.api_serialize())

src/sentry/api/urls.py

@@ -434,6 +434,7 @@ from .endpoints.organization_projects_sent_first_event import (
     OrganizationProjectsSentFirstEventEndpoint,
 )
 from .endpoints.organization_recent_searches import OrganizationRecentSearchesEndpoint
+from .endpoints.organization_region import OrganizationRegionEndpoint
 from .endpoints.organization_relay_usage import OrganizationRelayUsage
 from .endpoints.organization_release_assemble import OrganizationReleaseAssembleEndpoint
 from .endpoints.organization_release_commits import OrganizationReleaseCommitsEndpoint
@@ -2091,6 +2092,11 @@ ORGANIZATION_URLS = [
         PromptsActivityEndpoint.as_view(),
         name="sentry-api-0-organization-prompts-activity",
     ),
+    re_path(
+        r"^(?P<organization_slug>[^\/]+)/region/$",
+        OrganizationRegionEndpoint.as_view(),
+        name="sentry-api-0-organization-region",
+    ),
 ]
 
 PROJECT_URLS: list[URLPattern | URLResolver] = [

static/app/data/controlsiloUrlPatterns.ts

@@ -54,6 +54,7 @@ const patterns: RegExp[] = [
   new RegExp('^api/0/organizations/[^/]+/org-auth-tokens/$'),
   new RegExp('^api/0/organizations/[^/]+/org-auth-tokens/[^/]+/$'),
   new RegExp('^api/0/organizations/[^/]+/broadcasts/$'),
+  new RegExp('^api/0/organizations/[^/]+/region/$'),
   new RegExp('^api/0/users/$'),
   new RegExp('^api/0/users/[^/]+/$'),
   new RegExp('^api/0/users/[^/]+/regions/$'),

tests/sentry/api/endpoints/test_organization_region.py

@@ -0,0 +1,165 @@
+from sentry.models.organization import Organization
+from sentry.models.organizationmember import OrganizationMember
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.silo import assume_test_silo_mode_of, control_silo_test, create_test_regions
+from sentry.types.region import Region, get_region_by_name
+from sentry.utils.security.orgauthtoken_token import generate_token, hash_token
+
+
+@control_silo_test(regions=create_test_regions("us", "de"))
+class OrganizationRegionTest(APITestCase):
+    endpoint = "sentry-api-0-organization-region"
+
+    def setUp(self):
+        super().setUp()
+        self.org_owner = self.create_user()
+        us_region = get_region_by_name("us")
+        self.org = self.create_organization(owner=self.org_owner, region=us_region)
+        self.test_project = self.create_project(organization=self.org, name="test_project")
+
+    def create_internal_integration_for_org(self, org, user, scopes: list[str]):
+        internal_integration = self.create_internal_integration(organization=org, scopes=scopes)
+        integration_token = self.create_internal_integration_token(
+            internal_integration=internal_integration,
+            user=user,
+        )
+
+        return (internal_integration, integration_token)
+
+    def create_auth_token_for_org(self, org: Organization, region: Region, scopes: list[str]):
+        org_auth_token_str = generate_token(org.slug, region.to_url(""))
+        self.create_org_auth_token(
+            organization_id=org.id,
+            scope_list=scopes,
+            name="test_token",
+            token_hashed=hash_token(org_auth_token_str),
+            date_last_used=None,
+        )
+
+        return org_auth_token_str
+
+    def send_get_request_with_auth(self, org_slug: str, auth_token: str):
+        return self.get_response(
+            org_slug,
+            extra_headers={
+                "HTTP_AUTHORIZATION": f"Bearer {auth_token}",
+            },
+        )
+
+    def test_org_member_has_access(self):
+        self.login_as(self.org_owner)
+        response = self.get_response(self.org.slug)
+
+        assert response.status_code == 200
+        us_region = get_region_by_name("us")
+        assert response.data == {"url": us_region.to_url(""), "name": us_region.name}
+
+    def test_non_org_member_has_no_access(self):
+        non_member_user = self.create_user()
+        self.login_as(non_member_user)
+        response = self.get_response(self.org.slug)
+        assert response.status_code == 403
+
+    def test_org_auth_token_access_with_org_read(self):
+        us_region = get_region_by_name("us")
+        org_auth_token_str = self.create_auth_token_for_org(
+            region=us_region, org=self.org, scopes=["org:ci"]
+        )
+        response = self.send_get_request_with_auth(self.org.slug, org_auth_token_str)
+
+        us_region = get_region_by_name("us")
+        assert response.data == {"url": us_region.to_url(""), "name": us_region.name}
+        assert response.status_code == 200
+
+    def test_org_auth_token_access_with_incorrect_scopes(self):
+        us_region = get_region_by_name("us")
+        org_auth_token_str = self.create_auth_token_for_org(
+            region=us_region, org=self.org, scopes=[]
+        )
+        response = self.send_get_request_with_auth(self.org.slug, org_auth_token_str)
+
+        assert response.status_code == 403
+
+    def test_org_auth_token_access_for_different_organization(self):
+        us_region = get_region_by_name("us")
+
+        other_user = self.create_user()
+        org_auth_token_str = self.create_auth_token_for_org(
+            region=us_region, org=self.create_organization(owner=other_user), scopes=["org:ci"]
+        )
+        response = self.send_get_request_with_auth(self.org.slug, org_auth_token_str)
+
+        assert response.status_code == 403
+
+    def test_integration_token_access(self):
+        integration, token = self.create_internal_integration_for_org(
+            self.org, self.org_owner, ["project:read"]
+        )
+
+        response = self.send_get_request_with_auth(self.org.slug, token)
+
+        assert response.status_code == 200
+        us_region = get_region_by_name("us")
+        assert response.data == {"name": us_region.name, "url": us_region.to_url("")}
+
+    def test_integration_token_with_invalid_scopes(self):
+        integration, token = self.create_internal_integration_for_org(self.org, self.org_owner, [])
+
+        response = self.get_response(
+            self.org.slug,
+            extra_headers={
+                "HTTP_AUTHORIZATION": f"Bearer {token}",
+            },
+        )
+        assert response.status_code == 403
+
+    def test_integration_for_different_organization(self):
+        other_user = self.create_user()
+        integration, token = self.create_internal_integration_for_org(
+            self.create_organization(owner=other_user), other_user, ["project:read"]
+        )
+
+        response = self.send_get_request_with_auth(self.org.slug, token)
+        assert response.status_code == 403
+
+    def test_user_auth_token_for_owner(self):
+        user_auth_token = self.create_user_auth_token(user=self.org_owner, scope_list=["org:read"])
+        response = self.send_get_request_with_auth(self.org.slug, user_auth_token)
+
+        assert response.status_code == 200
+        us_region = get_region_by_name("us")
+        assert response.data == {"url": us_region.to_url(""), "name": us_region.name}
+
+    def test_user_auth_token_for_member(self):
+        org_user = self.create_user()
+        with assume_test_silo_mode_of(OrganizationMember):
+            OrganizationMember.objects.create(
+                user_id=org_user.id, organization_id=self.org.id, role="member"
+            )
+
+        user_auth_token = self.create_user_auth_token(user=org_user, scope_list=["org:read"])
+        response = self.send_get_request_with_auth(self.org.slug, user_auth_token)
+
+        assert response.status_code == 200
+        us_region = get_region_by_name("us")
+        assert response.data == {"url": us_region.to_url(""), "name": us_region.name}
+
+    def test_user_auth_token_for_non_member(self):
+        user_auth_token = self.create_user_auth_token(
+            user=self.create_user(), scope_list=["org:read"]
+        )
+        response = self.send_get_request_with_auth(self.org.slug, user_auth_token)
+        assert response.status_code == 403
+
+    def test_user_auth_token_with_invalid_scopes(self):
+        user_auth_token = self.create_user_auth_token(user=self.org_owner, scope_list=[])
+        response = self.send_get_request_with_auth(self.org.slug, user_auth_token)
+
+        assert response.status_code == 403
+
+        user_auth_token = self.create_user_auth_token(
+            user=self.org_owner, scope_list=["event:read"]
+        )
+        response = self.send_get_request_with_auth(self.org.slug, user_auth_token)
+
+        assert response.status_code == 403