feat(scopes): Enforce scope hierarchy (#54510)

### Note
- User Token UI Changes should be merged beforehand
#54651
- See[ Notion
doc](https://www.notion.so/sentry/WIP-Permission-Scope-Hierarchy-67620f99783345bbb8bc2828b1addc64)
for detailed spec on this work

### Implementation
To model the hierarchy I used a simple dict of scope -> all granted
scopes. I chose this approach b/c it's very simple and easy to enforce
in the code.

We use a pre-save signal on the `ApiKey` and `ApiToken` models to
enforce scope hierarchy. The basic scope hierarchy for each resource is:
- read grants nothing
- write grants read
- admin grants read+write

When updating one of these models, there is no way to know if the scopes
are enforced without fetching it from the DB. To avoid this trip, we
always iterate through the hierarchy mapping.

src/sentry/conf/server.py

@@ -2245,15 +2245,41 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
  "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)