Scoped permissions

NOTICE.md

@@ -1,4 +1,4 @@
-This product includes BSD licensed software developed by
+This product includes [BSD licensed](./sentry/LICENSE) software developed by
 Sentry (https://github.com/getsentry/sentry).
 
-See the [sentry](./sentry) directory for more information.
+See the [sentry](./sentry) directory for more information.

api_tokens/models.py

@@ -1,5 +1,6 @@
 import binascii
 import os
+from typing import List
 
 from django.conf import settings
 from django.db import models
@@ -45,3 +46,21 @@ class APIToken(models.Model):
 
     def __str__(self):
         return self.token
+
+    def get_scopes(self):
+        """
+        Return array of set scope flags.
+        Example: ["project:read"]
+        """
+        return [i[0] for i in self.scopes.items() if i[1] is True]
+
+    def add_permission(self, permission: str):
+        """ Add permission flag to scopes and save """
+        setattr(self.scopes, permission, True)
+        self.save(update_fields=["scopes"])
+
+    def add_permissions(self, permissions: List[str]):
+        """ Add permission flags to scopes and save """
+        for permission in permissions:
+            setattr(self.scopes, permission, True)
+        self.save(update_fields=["scopes"])

glitchtip/permissions.py

@@ -0,0 +1,27 @@
+from rest_framework.permissions import BasePermission
+
+
+class ScopedPermission(BasePermission):
+    """
+    Check if view has scope_map and compare it with request's auth scope map
+
+    Fall back to checking for user authentication
+    """
+
+    def get_allowed_scopes(self, request, view):
+        return self.scope_map[request.method]
+
+    def has_permission(self, request, view):
+        if request.auth:
+            allowed_scopes = self.get_allowed_scopes(request, view)
+            current_scopes = request.auth.get_scopes()
+            return any(s in allowed_scopes for s in current_scopes)
+        return bool(request.user and request.user.is_authenticated)
+
+    def get_user_scopes(self, obj, user):
+        pass
+
+    def has_object_permission(self, request, view, obj):
+        allowed_scopes = self.get_allowed_scopes(request, view)
+        current_scopes = self.get_user_scopes(obj, request.user)
+        return any(s in allowed_scopes for s in current_scopes)

glitchtip/settings.py

@@ -366,7 +366,7 @@ if DEBUG:
     )
 
 REST_FRAMEWORK = {
-    "DEFAULT_PERMISSION_CLASSES": ["rest_framework.permissions.IsAuthenticated",],
+    "DEFAULT_PERMISSION_CLASSES": ["rest_framework.permissions.IsAuthenticated"],
     "DEFAULT_PAGINATION_CLASS": "glitchtip.pagination.LinkHeaderPagination",
     "PAGE_SIZE": 50,
     "DEFAULT_FILTER_BACKENDS": ("django_filters.rest_framework.DjangoFilterBackend",),

glitchtip/test_utils/test_case.py

@@ -15,3 +15,40 @@ class GlitchTipTestCase(APITestCase):
         self.project = baker.make("projects.Project", organization=self.organization)
         self.project.team_set.add(self.team)
         self.client.force_login(self.user)
+
+
+class APIPermissionTestCase(APITestCase):
+    """ Base class for testing viewsets with permissions """
+
+    def create_user_org(self):
+        self.user = baker.make("users.user")
+        self.organization = baker.make("organizations_ext.Organization")
+        self.org_user = self.organization.add_user(self.user)
+        self.auth_token = baker.make("api_tokens.APIToken", user=self.user)
+
+    def set_client_credentials(self, token: str):
+        self.client.credentials(HTTP_AUTHORIZATION="Bearer " + token)
+
+    def set_user_role(self, role: OrganizationUserRole):
+        self.org_user.role = role
+        self.org_user.save(update_fields=["role"])
+
+    def assertGetReqStatusCode(self, url: str, status_code: int, msg=None):
+        """ Make GET request to url and check status code """
+        res = self.client.get(url)
+        self.assertEqual(res.status_code, status_code, msg)
+
+    def assertPostReqStatusCode(self, url: str, data, status_code: int, msg=None):
+        """ Make POST request to url and check status code """
+        res = self.client.post(url, data)
+        self.assertEqual(res.status_code, status_code, msg)
+
+    def assertPutReqStatusCode(self, url: str, data, status_code: int, msg=None):
+        """ Make PUT request to url and check status code """
+        res = self.client.put(url, data)
+        self.assertEqual(res.status_code, status_code, msg)
+
+    def assertDeleteReqStatusCode(self, url: str, status_code: int, msg=None):
+        """ Make DELETE request to url and check status code """
+        res = self.client.delete(url)
+        self.assertEqual(res.status_code, status_code, msg)

organizations_ext/models.py

@@ -10,6 +10,104 @@ from organizations.fields import SlugField
 from organizations.signals import user_added
 
 
+# Defines which scopes belong to which role
+# Credit to sentry/conf/server.py
+ROLES = (
+    {
+        "id": "member",
+        "name": "Member",
+        "desc": "Members can view and act on events, as well as view most other data within the organization.",
+        "scopes": set(
+            [
+                "event:read",
+                "event:write",
+                "event:admin",
+                "project:releases",
+                "project:read",
+                "org:read",
+                "member:read",
+                "team:read",
+            ]
+        ),
+    },
+    {
+        "id": "admin",
+        "name": "Admin",
+        "desc": "Admin privileges on any teams of which they're a member. They can create new teams and projects, as well as remove teams and projects which they already hold membership on (or all teams, if open membership is on). Additionally, they can manage memberships of teams that they are members of.",
+        "scopes": set(
+            [
+                "event:read",
+                "event:write",
+                "event:admin",
+                "org:read",
+                "member:read",
+                "project:read",
+                "project:write",
+                "project:admin",
+                "project:releases",
+                "team:read",
+                "team:write",
+                "team:admin",
+                "org:integrations",
+            ]
+        ),
+    },
+    {
+        "id": "manager",
+        "name": "Manager",
+        "desc": "Gains admin access on all teams as well as the ability to add and remove members.",
+        "is_global": True,
+        "scopes": set(
+            [
+                "event:read",
+                "event:write",
+                "event:admin",
+                "member:read",
+                "member:write",
+                "member:admin",
+                "project:read",
+                "project:write",
+                "project:admin",
+                "project:releases",
+                "team:read",
+                "team:write",
+                "team:admin",
+                "org:read",
+                "org:write",
+                "org:integrations",
+            ]
+        ),
+    },
+    {
+        "id": "owner",
+        "name": "Organization Owner",
+        "desc": "Unrestricted access to the organization, its data, and its settings. Can add, modify, and delete projects and members, as well as make billing and plan changes.",
+        "is_global": True,
+        "scopes": set(
+            [
+                "org:read",
+                "org:write",
+                "org:admin",
+                "org:integrations",
+                "member:read",
+                "member:write",
+                "member:admin",
+                "team:read",
+                "team:write",
+                "team:admin",
+                "project:read",
+                "project:write",
+                "project:admin",
+                "project:releases",
+                "event:read",
+                "event:write",
+                "event:admin",
+            ]
+        ),
+    },
+)
+
+
 class OrganizationUserRole(models.IntegerChoices):
     MEMBER = 0, "Member"
     ADMIN = 1, "Admin"
@@ -22,6 +120,10 @@ class OrganizationUserRole(models.IntegerChoices):
             if status.label.lower() == string.lower():
                 return status
 
+    @classmethod
+    def get_role(cls, role: int):
+        return ROLES[role]
+
 
 class Organization(SharedBaseModel, OrganizationBase):
     slug = SlugField(
@@ -71,6 +173,10 @@ class Organization(SharedBaseModel, OrganizationBase):
         billing_contact = self.owner.organization_user.user
         return billing_contact.email
 
+    def get_user_scopes(self, user):
+        org_user = self.organization_users.get(user=user)
+        return org_user.get_scopes()
+
 
 class OrganizationUser(SharedBaseModel, OrganizationUserBase):
     user = models.ForeignKey(
@@ -101,6 +207,10 @@ class OrganizationUser(SharedBaseModel, OrganizationUserBase):
     def get_role(self):
         return self.get_role_display().lower()
 
+    def get_scopes(self):
+        role = OrganizationUserRole.get_role(self.role)
+        return role["scopes"]
+
     def accept_invite(self, user):
         self.user = user
         self.email = None

organizations_ext/permissions.py

@@ -0,0 +1,51 @@
+from glitchtip.permissions import ScopedPermission
+
+
+class OrganizationPermission(ScopedPermission):
+    scope_map = {
+        "GET": ["org:read", "org:write", "org:admin"],
+        "POST": ["org:write", "org:admin"],
+        "PUT": ["org:write", "org:admin"],
+        "DELETE": ["org:admin"],
+    }
+
+    def get_user_scopes(self, obj, user):
+        return obj.get_user_scopes(user)
+
+
+class OrganizationMemberPermission(ScopedPermission):
+    scope_map = {
+        "GET": ["member:read", "member:write", "member:admin"],
+        "POST": ["member:write", "member:admin"],
+        "PUT": ["member:write", "member:admin"],
+        "DELETE": ["member:admin"],
+    }
+
+    def has_permission(self, request, view):
+        # teams action has entirely different permissions
+        if view.action == "teams":
+            permission = OrganizationMemberTeamsPermission()
+            if request.auth:
+                allowed_scopes = permission.get_allowed_scopes(request, view)
+                current_scopes = request.auth.get_scopes()
+                return any(s in allowed_scopes for s in current_scopes)
+            return bool(request.user and request.user.is_authenticated)
+        return super().has_permission(request, view)
+
+    def get_user_scopes(self, obj, user):
+        return obj.organization.get_user_scopes(user)
+
+
+class OrganizationMemberTeamsPermission(OrganizationMemberPermission):
+    _allowed_scopes = [
+        "org:read",
+        "org:write",
+        "org:admin",
+        "member:read",
+        "member:write",
+        "member:admin",
+    ]
+    scope_map = {
+        "POST": _allowed_scopes,
+        "DELETE": _allowed_scopes,
+    }

organizations_ext/serializers/serializers.py

@@ -5,7 +5,7 @@ from users.serializers import UserSerializer
 from teams.serializers import TeamSerializer
 from teams.models import Team
 from .base_serializers import OrganizationReferenceSerializer
-from ..models import OrganizationUser, OrganizationUserRole
+from ..models import OrganizationUser, OrganizationUserRole, ROLES
 
 
 class OrganizationSerializer(OrganizationReferenceSerializer):
@@ -106,6 +106,13 @@ class OrganizationUserDetailSerializer(OrganizationUserSerializer):
     teams = serializers.SlugRelatedField(
         source="team_set", slug_field="slug", read_only=True, many=True
     )
+    roles = serializers.SerializerMethodField()
+
+    class Meta(OrganizationUserSerializer.Meta):
+        fields = OrganizationUserSerializer.Meta.fields + ("roles",)
+
+    def get_roles(self, obj):
+        return ROLES
 
 
 class OrganizationUserProjectsSerializer(OrganizationUserSerializer):

organizations_ext/tests/test_api_permissions.py

@@ -0,0 +1,127 @@
+from django.urls import reverse
+from model_bakery import baker
+from organizations_ext.models import OrganizationUserRole
+from glitchtip.test_utils.test_case import APIPermissionTestCase
+
+
+class OrganizationAPIPermissionTests(APIPermissionTestCase):
+    def setUp(self):
+        self.create_user_org()
+        self.set_client_credentials(self.auth_token.token)
+        self.list_url = reverse("organization-list")
+        self.detail_url = reverse("organization-detail", args=[self.organization.slug])
+
+    def test_list(self):
+        self.assertGetReqStatusCode(self.list_url, 403)
+        self.auth_token.add_permission("org:read")
+        self.assertGetReqStatusCode(self.list_url, 200)
+
+    def test_retrieve(self):
+        self.assertGetReqStatusCode(self.detail_url, 403)
+        self.auth_token.add_permission("org:read")
+        self.assertGetReqStatusCode(self.detail_url, 200)
+
+    def test_create(self):
+        self.auth_token.add_permission("org:read")
+        data = {"name": "new org"}
+        self.assertPostReqStatusCode(self.list_url, data, 403)
+        self.auth_token.add_permission("org:write")
+        self.assertPostReqStatusCode(self.list_url, data, 201)
+
+    def test_destroy(self):
+        self.auth_token.add_permissions(["org:read", "org:write"])
+        self.assertDeleteReqStatusCode(self.detail_url, 403)
+        self.auth_token.add_permission("org:admin")
+        self.assertDeleteReqStatusCode(self.detail_url, 204)
+
+    def test_user_destroy(self):
+        self.client.force_login(self.user)
+        self.set_user_role(OrganizationUserRole.MEMBER)
+        self.assertDeleteReqStatusCode(self.detail_url, 403)
+        self.set_user_role(OrganizationUserRole.OWNER)
+        self.assertDeleteReqStatusCode(self.detail_url, 204)
+
+    def test_update(self):
+        self.auth_token.add_permission("org:read")
+        data = {"name": "new name"}
+        self.assertPutReqStatusCode(self.detail_url, data, 403)
+        self.auth_token.add_permission("org:write")
+        self.assertPutReqStatusCode(self.detail_url, data, 200)
+
+    def test_user_update(self):
+        self.client.force_login(self.user)
+        self.set_user_role(OrganizationUserRole.MEMBER)
+        data = {"name": "new name"}
+        self.assertPutReqStatusCode(self.detail_url, data, 403)
+        self.set_user_role(OrganizationUserRole.MANAGER)
+        self.assertPutReqStatusCode(self.detail_url, data, 200)
+
+
+class OrganizationMemberAPIPermissionTests(APIPermissionTestCase):
+    def setUp(self):
+        self.create_user_org()
+        self.set_client_credentials(self.auth_token.token)
+        self.list_url = reverse(
+            "organization-members-list",
+            kwargs={"organization_slug": self.organization.slug},
+        )
+        self.detail_url = reverse(
+            "organization-members-detail",
+            kwargs={
+                "organization_slug": self.organization.slug,
+                "pk": self.org_user.pk,
+            },
+        )
+
+    def test_list(self):
+        self.assertGetReqStatusCode(self.list_url, 403)
+        self.auth_token.add_permission("member:read")
+        self.assertGetReqStatusCode(self.list_url, 200)
+
+    def test_retrieve(self):
+        self.assertGetReqStatusCode(self.detail_url, 403)
+        self.auth_token.add_permission("member:read")
+        self.assertGetReqStatusCode(self.detail_url, 200)
+
+    def test_create(self):
+        self.auth_token.add_permission("member:read")
+        data = {"email": "lol@example.com", "role": "member"}
+        self.assertPostReqStatusCode(self.list_url, data, 403)
+        self.auth_token.add_permission("member:write")
+        self.assertPostReqStatusCode(self.list_url, data, 201)
+
+    def test_destroy(self):
+        self.auth_token.add_permissions(["member:read", "member:write"])
+        self.assertDeleteReqStatusCode(self.detail_url, 403)
+        self.auth_token.add_permission("member:admin")
+        self.assertDeleteReqStatusCode(self.detail_url, 204)
+
+    def test_user_destroy(self):
+        self.client.force_login(self.user)
+        self.set_user_role(OrganizationUserRole.MEMBER)
+        self.assertDeleteReqStatusCode(self.detail_url, 403)
+        self.set_user_role(OrganizationUserRole.OWNER)
+        self.assertDeleteReqStatusCode(self.detail_url, 204)
+
+    def test_update(self):
+        self.auth_token.add_permission("member:read")
+        data = {"email": "lol@example.com", "role": "member"}
+        self.assertPutReqStatusCode(self.detail_url, data, 403)
+        self.auth_token.add_permission("member:write")
+        self.assertPutReqStatusCode(self.detail_url, data, 200)
+
+    def test_teams_add(self):
+        self.team = baker.make("teams.Team", organization=self.organization)
+        url = self.detail_url + "teams/" + self.team.slug + "/"
+        data = {}
+        self.assertPostReqStatusCode(url, data, 403)
+        self.auth_token.add_permissions(["org:read", "org:write"])
+        self.assertPostReqStatusCode(url, data, 201)
+
+    def test_teams_remove(self):
+        self.team = baker.make("teams.Team", organization=self.organization)
+        url = self.detail_url + "teams/" + self.team.slug + "/"
+        self.assertDeleteReqStatusCode(url, 403)
+        self.auth_token.add_permissions(["org:read", "org:write"])
+        self.assertDeleteReqStatusCode(url, 200)
+

organizations_ext/views.py

@@ -10,6 +10,11 @@ from organizations.backends import invitation_backend
 from teams.serializers import TeamSerializer
 from users.utils import is_user_registration_open
 from projects.views import NestedProjectViewSet
+from .permissions import (
+    OrganizationPermission,
+    OrganizationMemberPermission,
+    OrganizationMemberTeamsPermission,
+)
 from .invitation_backend import InvitationTokenGenerator
 from .models import Organization, OrganizationUserRole, OrganizationUser
 from .serializers.serializers import (
@@ -27,6 +32,7 @@ class OrganizationViewSet(viewsets.ModelViewSet):
     queryset = Organization.objects.all()
     serializer_class = OrganizationSerializer
     lookup_field = "slug"
+    permission_classes = [OrganizationPermission]
 
     def get_serializer_class(self):
         if self.action in ["retrieve"]:
@@ -55,6 +61,7 @@ class OrganizationMemberViewSet(viewsets.ModelViewSet):
 
     queryset = OrganizationUser.objects.all()
     serializer_class = OrganizationUserSerializer
+    permission_classes = [OrganizationMemberPermission]
 
     def get_serializer_class(self):
         if self.action in ["retrieve"]:
@@ -169,9 +176,23 @@ class OrganizationMemberViewSet(viewsets.ModelViewSet):
         if not pk or not organization_slug or not members_team_slug:
             raise exceptions.MethodNotAllowed(request.method)
 
-        org_user = self.get_object()
+        pk = self.kwargs.get("pk")
+        if pk == "me":
+            org_user = get_object_or_404(self.get_queryset(), user=self.request.user)
+        else:
+            queryset = self.filter_queryset(self.get_queryset())
+            lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field
+            filter_kwargs = {self.lookup_field: self.kwargs[lookup_url_kwarg]}
+            org_user = get_object_or_404(queryset, **filter_kwargs)
+
         team = org_user.organization.teams.filter(slug=members_team_slug).first()
 
+        # Instead of check_object_permissions
+        permission = OrganizationMemberTeamsPermission()
+        if not permission.has_permission(request, self):
+            self.permission_denied(
+                request, message=getattr(permission, "message", None)
+            )
         self.check_team_member_permission(org_user, self.request.user, team)
 
         if not team: