feat(scopes): Enforce scope hierarchy

src/sentry/conf/server.py

@@ -2235,15 +2235,41 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "event:read",
     "event:write",
     "event:admin",
-    "alerts:write",
     "alerts:read",
+    "alerts:write",
     # openid, profile, and email aren't prefixed to maintain compliance with the OIDC spec.
     # https://auth0.com/docs/get-started/apis/scopes/openid-connect-scopes.
     "openid",
     "profile",
     "email",
 }
 
+SENTRY_SCOPE_HIERARCHY_MAPPING = {
+    "org:read": {"org:read"},
+    "org:write": {"org:read", "org:write"},
+    "org:admin": {"org:read", "org:write", "org:admin", "org:integrations"},
+    "org:integrations": {"org:integrations"},
+    "org:ci": {"org:ci"},
+    "member:read": {"member:read"},
+    "member:write": {"member:read", "member:write"},
+    "member:admin": {"member:read", "member:write", "member:admin"},
+    "team:read": {"team:read"},
+    "team:write": {"team:read", "team:write"},
+    "team:admin": {"team:read", "team:write", "team:admin"},
+    "project:read": {"project:read"},
+    "project:write": {"project:read", "project:write"},
+    "project:admin": {"project:read", "project:write", "project:admin"},
+    "project:releases": {"project:releases"},
+    "event:read": {"event:read"},
+    "event:write": {"event:read", "event:write"},
+    "event:admin": {"event:read", "event:write", "event:admin"},
+    "alerts:read": {"alerts:read"},
+    "alerts:write": {"alerts:read", "alerts:write"},
+    "openid": {"openid"},
+    "profile": {"profile"},
+    "email": {"email"},
+}
+
 SENTRY_SCOPE_SETS = (
     (
         ("org:admin", "Read, write, and admin access to organization details."),

src/sentry/models/apiscopes.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from typing import TypedDict
+from typing import List, TypedDict
 
 from django.db import models
 
@@ -73,10 +73,16 @@ class Meta:
     # Human readable list of scopes
     scope_list = ArrayField(of=models.TextField)
 
-    def get_scopes(self):
+    def get_scopes(self) -> List[str]:
+        """
+        Returns a list of the token's scopes in alphabetical order.
+        """
         if self.scope_list:
-            return self.scope_list
-        return [k for k, v in self.scopes.items() if v]
+            return sorted(self.scope_list)
+        return sorted([k for k, v in self.scopes.items() if v])
 
-    def has_scope(self, scope):
+    def has_scope(self, scope: str) -> bool:
+        """
+        Checks whether the token has the given scope
+        """
         return scope in self.get_scopes()

src/sentry/receivers/__init__.py

@@ -14,5 +14,6 @@
 from .sentry_apps import *  # noqa: F401,F403
 from .stats import *  # noqa: F401,F403
 from .superuser import *  # noqa: F401,F403
+from .tokens import *  # noqa: F401,F403
 from .useremail import *  # noqa: F401,F403
 from .users import *  # noqa: F401,F403

src/sentry/receivers/tokens.py

@@ -0,0 +1,17 @@
+from django.db.models.signals import pre_save
+from django.dispatch import receiver
+
+from sentry.conf.server import SENTRY_SCOPE_HIERARCHY_MAPPING, SENTRY_SCOPES
+from sentry.models import ApiKey, ApiToken
+
+
+@receiver(pre_save, sender=ApiKey, dispatch_uid="enforce_scope_hierarchy_api_key")
+@receiver(pre_save, sender=ApiToken, dispatch_uid="enforce_scope_hierarchy_api_token")
+def enforce_scope_hierarchy(instance, **kwargs) -> None:
+    # It's impossible to know if the token scopes have been modified without
+    # fetching it from the DB, so we always enforce scope hierarchy
+    new_scopes = set(instance.get_scopes())
+    for scope in instance.scope_list:
+        if scope in SENTRY_SCOPES:
+            new_scopes = new_scopes.union(SENTRY_SCOPE_HIERARCHY_MAPPING[scope])
+    instance.scope_list = sorted(new_scopes)

tests/sentry/api/endpoints/test_sentry_apps.py

@@ -29,7 +29,10 @@
 EXPECTED = {
     "events": ["issue"],
     "name": "MyApp",
-    "scopes": ["project:read", "event:read"],
+    "scopes": [
+        "event:read",
+        "project:read",
+    ],
     "webhookUrl": "https://example.com",
 }
 

tests/sentry/conf/test_scopes.py

@@ -0,0 +1,25 @@
+from sentry.conf.server import SENTRY_SCOPE_HIERARCHY_MAPPING, SENTRY_SCOPES
+from sentry.testutils.cases import TestCase
+
+
+class ScopesTest(TestCase):
+    def test_scope_hierarchy_maintained(self):
+        for scope in SENTRY_SCOPES:
+            assert scope in SENTRY_SCOPE_HIERARCHY_MAPPING
+
+            # exclude special OAuth scopes
+            if ":" not in scope:
+                continue
+            resource, access_level = scope.split(":")
+
+            # check that scope is in its own mapping
+            assert scope in SENTRY_SCOPE_HIERARCHY_MAPPING[scope]
+
+            # check that write grants read
+            if access_level == "write":
+                assert resource + ":read" in SENTRY_SCOPE_HIERARCHY_MAPPING[scope]
+
+            # # check that admin grants read+write
+            if access_level == "admin":
+                assert resource + ":read" in SENTRY_SCOPE_HIERARCHY_MAPPING[scope]
+                assert resource + ":write" in SENTRY_SCOPE_HIERARCHY_MAPPING[scope]

tests/sentry/models/test_apikey.py

@@ -0,0 +1,13 @@
+from sentry.conf.server import SENTRY_SCOPE_HIERARCHY_MAPPING, SENTRY_SCOPES
+from sentry.testutils.cases import TestCase
+from sentry.testutils.silo import control_silo_test
+
+
+@control_silo_test(stable=True)
+class ApiTokenTest(TestCase):
+    def test_enforces_scope_hierarchy(self):
+        org = self.create_organization()
+        # Ensure hierarchy is enforced for all tokens
+        for scope in SENTRY_SCOPES:
+            token = self.create_api_key(org, scope_list=[scope])
+            assert token.get_scopes() == sorted(SENTRY_SCOPE_HIERARCHY_MAPPING[scope])

tests/sentry/models/test_apitoken.py

@@ -2,13 +2,14 @@
 
 from django.utils import timezone
 
+from sentry.conf.server import SENTRY_SCOPE_HIERARCHY_MAPPING, SENTRY_SCOPES
 from sentry.models import ApiToken
 from sentry.models.integrations.sentry_app_installation import SentryAppInstallation
 from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import region_silo_test
+from sentry.testutils.silo import control_silo_test
 
 
-@region_silo_test(stable=True)
+@control_silo_test(stable=True)
 class ApiTokenTest(TestCase):
     def test_is_expired(self):
         token = ApiToken(expires_at=None)
@@ -30,6 +31,17 @@ def test_get_scopes(self):
         token = ApiToken(scope_list=["project:read"])
         assert token.get_scopes() == ["project:read"]
 
+    def test_enforces_scope_hierarchy(self):
+        user = self.create_user()
+
+        # Ensure hierarchy is enforced for all tokens
+        for scope in SENTRY_SCOPES:
+            token = ApiToken.objects.create(
+                user_id=user.id,
+                scope_list=[scope],
+            )
+            assert set(token.get_scopes()) == SENTRY_SCOPE_HIERARCHY_MAPPING[scope]
+
 
 class ApiTokenInternalIntegrationTest(TestCase):
     def setUp(self):

tests/sentry/sentry_apps/test_sentry_app_installation_token_creator.py

@@ -64,7 +64,7 @@ def setUp(self):
         super().setUp()
 
         self.sentry_app = self.create_sentry_app(
-            name="external_app", organization=self.org, scopes=("org:write", "team:admin")
+            name="external_app", organization=self.org, scopes=("org:read", "team:read")
         )
 
         self.sentry_app_installation = SentryAppInstallationCreator(
@@ -79,4 +79,4 @@ def test_create_token(self):
 
         # verify token was created properly
         assert api_token.expires_at == today
-        assert api_token.scope_list == ["org:write", "team:admin"]
+        assert api_token.scope_list == ["org:read", "team:read"]

tests/sentry/web/frontend/test_oauth_token.py

@@ -325,7 +325,7 @@ def test_valid_params_id_token_additional_scopes(self):
             data = json.loads(resp.content)
             token = ApiToken.objects.get(token=data["access_token"])
 
-            assert token.get_scopes() == ["openid", "profile", "email"]
+            assert token.get_scopes() == ["email", "openid", "profile"]
             assert data["refresh_token"] == token.refresh_token
             assert data["access_token"] == token.token
             assert isinstance(data["expires_in"], int)