feat(admin): implement basic admin user roles

snuba/admin/auth.py

@@ -5,6 +5,7 @@
 from flask import request
 
 from snuba import settings
+from snuba.admin.auth_roles import DEFAULT_ROLES
 from snuba.admin.jwt import validate_assertion
 from snuba.admin.user import AdminUser
 
@@ -28,7 +29,15 @@ def authorize_request() -> AdminUser:
     provider = AUTH_PROVIDERS.get(provider_id)
     if provider is None:
         raise ValueError("Invalid authorization provider")
-    return provider()
+
+    return _set_roles(provider())
+
+
+def _set_roles(user: AdminUser) -> AdminUser:
+    # todo: depending on provider convert user email
+    # to subset of DEFAULT_ROLES based on IAM roles
+    user.roles = DEFAULT_ROLES
+    return user
 
 
 def passthrough_authorize() -> AdminUser:

snuba/admin/auth_roles.py

@@ -0,0 +1,101 @@
+from abc import ABC, abstractmethod, abstractproperty
+from dataclasses import dataclass
+from enum import Enum
+from typing import Generic, Sequence, Set, TypeVar, Union
+
+from snuba import settings
+
+
+class Category(Enum):
+    MIGRATIONS = "migrations"
+
+
+class Resource(ABC):
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+    @abstractproperty
+    def category(self) -> Category:
+        raise NotImplementedError
+
+
+class MigrationResource(Resource):
+    def category(self) -> Category:
+        return Category.MIGRATIONS
+
+
+TResource = TypeVar("TResource", bound=Resource)
+
+
+class Action(ABC, Generic[TResource]):
+    """
+    An action is used to describe the permissions a user has
+    on a specific set of resources.
+    """
+
+    @abstractmethod
+    def validated_resources(
+        self, resources: Sequence[TResource]
+    ) -> Sequence[TResource]:
+        """
+        A resource is considered valid if the action can be
+        taken on the resource.
+
+        e.g. a user can "execute" (action) a migration within
+        a "migration group" (resource)
+
+        Raise an error if any resources are invalid, otherwise
+        return the resources.
+        """
+        raise NotImplementedError
+
+
+class MigrationAction(Action[MigrationResource]):
+    def __init__(self, resources: Sequence[MigrationResource]) -> None:
+        self._resources = self.validated_resources(resources)
+
+    def validated_resources(
+        self, resources: Sequence[MigrationResource]
+    ) -> Sequence[MigrationResource]:
+        return resources
+
+
+class ExecuteAllAction(MigrationAction):
+    pass
+
+
+class ExecuteNonBlockingAction(MigrationAction):
+    pass
+
+
+class ExecuteNoneAction(MigrationAction):
+    pass
+
+
+# todo, shoudln't need .keys() once ADMIN_ALLOWED_MIGRATION_GROUPS is a set not dict
+MIGRATIONS_RESOURCES = {
+    group: MigrationResource(group)
+    for group in settings.ADMIN_ALLOWED_MIGRATION_GROUPS.keys()
+}
+
+
+@dataclass(frozen=True)
+class Role:
+    name: str
+    actions: Set[Union[MigrationAction]]
+
+
+DEFAULT_ROLES = [
+    Role(
+        name="MigrationsReader",
+        actions={ExecuteNoneAction(list(MIGRATIONS_RESOURCES.values()))},
+    ),
+    Role(
+        name="MigrationsLimitedExecutor",
+        actions={ExecuteNonBlockingAction(list(MIGRATIONS_RESOURCES.values()))},
+    ),
+    Role(
+        name="TestMigrationsExecutor",
+        actions={ExecuteAllAction([MIGRATIONS_RESOURCES["test_migration"]])},
+    ),
+]

snuba/admin/user.py

@@ -1,4 +1,7 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import Sequence
+
+from snuba.admin.auth_roles import Role
 
 
 @dataclass
@@ -10,3 +13,4 @@ class AdminUser:
 
     email: str
     id: str
+    roles: Sequence[Role] = field(default_factory=list)