ref(rbac): improve rbac implementation for views (#6226)

api/src/backend/api/base_views.py

@@ -11,11 +11,13 @@ from api.db_router import MainRouter
 from api.db_utils import POSTGRES_USER_VAR, rls_transaction
 from api.filters import CustomDjangoFilterBackend
 from api.models import Role, Tenant
+from api.rbac.permissions import HasPermissions
 
 
 class BaseViewSet(ModelViewSet):
     authentication_classes = [JWTAuthentication]
-    permission_classes = [permissions.IsAuthenticated]
+    required_permissions = []
+    permission_classes = [permissions.IsAuthenticated, HasPermissions]
     filter_backends = [
         filters.QueryParameterValidationFilter,
         filters.OrderingFilter,
@@ -29,6 +31,17 @@ class BaseViewSet(ModelViewSet):
     ordering_fields = "__all__"
     ordering = ["id"]
 
+    def initial(self, request, *args, **kwargs):
+        """
+        Sets required_permissions before permissions are checked.
+        """
+        self.set_required_permissions()
+        super().initial(request, *args, **kwargs)
+
+    def set_required_permissions(self):
+        """This is an abstract method that must be implemented by subclasses."""
+        NotImplemented
+
     def get_queryset(self):
         raise NotImplementedError
 

api/src/backend/api/rbac/permissions.py

@@ -1,7 +1,9 @@
 from enum import Enum
 from rest_framework.permissions import BasePermission
-from api.models import User
+from api.models import Provider, Role, User
 from api.db_router import MainRouter
+from typing import Optional
+from django.db.models import QuerySet
 
 
 class Permissions(Enum):
@@ -36,3 +38,33 @@ class HasPermissions(BasePermission):
                 return False
 
         return True
+
+
+def get_role(user: User) -> Optional[Role]:
+    """
+    Retrieve the first role assigned to the given user.
+
+    Returns:
+        The user's first Role instance if the user has any roles, otherwise None.
+    """
+    return user.roles.first()
+
+
+def get_providers(role: Role) -> QuerySet[Provider]:
+    """
+    Return a distinct queryset of Providers accessible by the given role.
+
+    If the role has no associated provider groups, an empty queryset is returned.
+
+    Args:
+        role: A Role instance.
+
+    Returns:
+        A QuerySet of Provider objects filtered by the role's provider groups.
+        If the role has no provider groups, returns an empty queryset.
+    """
+    provider_groups = role.provider_groups.all()
+    if not provider_groups.exists():
+        return Provider.objects.none()
+
+    return Provider.objects.filter(provider_groups__in=provider_groups).distinct()

api/src/backend/api/tests/test_views.py

@@ -543,6 +543,7 @@ class TestTenantViewSet:
         # Test user + 2 extra users for tenant 2
         assert len(response.json()["data"]) == 3
 
+    @patch("api.v1.views.TenantMembersViewSet.required_permissions", [])
     def test_tenants_list_memberships_as_member(
         self, authenticated_client, tenants_fixture, extra_users
     ):
@@ -3274,9 +3275,7 @@ class TestRoleViewSet:
         assert len(data) == 1
         assert data[0]["attributes"]["name"] == role.name
 
-    def test_role_list_sorting(
-        self, authenticated_client, set_user_admin_roles_fixture, roles_fixture
-    ):
+    def test_role_list_sorting(self, authenticated_client, roles_fixture):
         response = authenticated_client.get(reverse("role-list"), {"sort": "name"})
         assert response.status_code == status.HTTP_200_OK
         data = response.json()["data"]
@@ -3342,7 +3341,7 @@ class TestUserRoleRelationshipViewSet:
     ):
         data = {
             "data": [
-                {"type": "role", "id": str(roles_fixture[1].id)},
+                {"type": "role", "id": str(roles_fixture[2].id)},
             ]
         }
         response = authenticated_client.patch(
@@ -3353,7 +3352,7 @@ class TestUserRoleRelationshipViewSet:
         assert response.status_code == status.HTTP_204_NO_CONTENT
         relationships = UserRoleRelationship.objects.filter(user=create_test_user.id)
         assert relationships.count() == 1
-        assert {rel.role.id for rel in relationships} == {roles_fixture[1].id}
+        assert {rel.role.id for rel in relationships} == {roles_fixture[2].id}
 
         data = {
             "data": [

api/src/backend/api/v1/views.py

@@ -79,7 +79,7 @@ from api.models import (
     UserRoleRelationship,
 )
 from api.pagination import ComplianceOverviewPagination
-from api.rbac.permissions import HasPermissions, Permissions
+from api.rbac.permissions import Permissions, get_providers, get_role
 from api.rls import Tenant
 from api.utils import validate_invitation
 from api.uuid_utils import datetime_to_uuid7
@@ -290,26 +290,19 @@ class UserViewSet(BaseUserViewset):
     filterset_class = UserFilter
     ordering = ["-date_joined"]
     ordering_fields = ["name", "email", "company_name", "date_joined", "is_active"]
+    # RBAC required permissions
     required_permissions = [Permissions.MANAGE_USERS]
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
 
-    def initial(self, request, *args, **kwargs):
-        """
-        Sets required_permissions before permissions are checked.
-        """
-        self.required_permissions = self.get_required_permissions()
-        super().initial(request, *args, **kwargs)
-
-    def get_required_permissions(self):
+    def set_required_permissions(self):
         """
         Returns the required permissions based on the request method.
         """
         if self.action == "me":
             # No permissions required for me request
-            return []
+            self.required_permissions = []
         else:
             # Require permission for the rest of the requests
-            return [Permissions.MANAGE_USERS]
+            self.required_permissions = [Permissions.MANAGE_USERS]
 
     def get_queryset(self):
         # If called during schema generation, return an empty queryset
@@ -454,6 +447,8 @@ class UserRoleRelationshipView(RelationshipView, BaseRLSViewSet):
     resource_name = "roles"
     http_method_names = ["post", "patch", "delete"]
     schema = RelationshipViewSchema()
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_USERS]
 
     def get_queryset(self):
         return User.objects.all()
@@ -540,8 +535,8 @@ class TenantViewSet(BaseTenantViewset):
     search_fields = ["name"]
     ordering = ["-inserted_at"]
     ordering_fields = ["name", "inserted_at", "updated_at"]
+    # RBAC required permissions
     required_permissions = [Permissions.MANAGE_ACCOUNT]
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
 
     def get_queryset(self):
         return Tenant.objects.all()
@@ -599,6 +594,8 @@ class MembershipViewSet(BaseTenantViewset):
         "role",
         "date_joined",
     ]
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_ACCOUNT]
 
     def get_queryset(self):
         user = self.request.user
@@ -632,6 +629,8 @@ class TenantMembersViewSet(BaseTenantViewset):
     http_method_names = ["get", "delete"]
     serializer_class = MembershipSerializer
     queryset = Membership.objects.none()
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_ACCOUNT]
 
     def get_queryset(self):
         tenant = self.get_tenant()
@@ -717,44 +716,29 @@ class ProviderGroupViewSet(BaseRLSViewSet):
     filterset_class = ProviderGroupFilter
     http_method_names = ["get", "post", "patch", "delete"]
     ordering = ["inserted_at"]
-    required_permissions = []
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_PROVIDERS]
 
-    def initial(self, request, *args, **kwargs):
-        """
-        Sets required_permissions before permissions are checked.
-        """
-        self.required_permissions = self.get_required_permissions()
-        super().initial(request, *args, **kwargs)
-
-    def get_required_permissions(self):
+    def set_required_permissions(self):
         """
         Returns the required permissions based on the request method.
         """
         if self.request.method in SAFE_METHODS:
             # No permissions required for GET requests
-            return []
+            self.required_permissions = []
         else:
             # Require permission for non-GET requests
-            return [Permissions.MANAGE_PROVIDERS]
+            self.required_permissions = [Permissions.MANAGE_PROVIDERS]
 
     def get_queryset(self):
-        user = self.request.user
-        user_roles = user.roles.all()
-
+        user_roles = get_role(self.request.user)
         # Check if any of the user's roles have UNLIMITED_VISIBILITY
-        if getattr(user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False):
+        if user_roles.unlimited_visibility:
             # User has unlimited visibility, return all provider groups
             return ProviderGroup.objects.prefetch_related("providers")
 
         # Collect provider groups associated with the user's roles
-        provider_groups = (
-            ProviderGroup.objects.filter(roles__in=user_roles)
-            .distinct()
-            .prefetch_related("providers")
-        )
-
-        return provider_groups
+        return user_roles.provider_groups.all()
 
     def get_serializer_class(self):
         if self.action == "partial_update":
@@ -799,6 +783,8 @@ class ProviderGroupProvidersRelationshipView(RelationshipView, BaseRLSViewSet):
     resource_name = "providers"
     http_method_names = ["post", "patch", "delete"]
     schema = RelationshipViewSchema()
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_PROVIDERS]
 
     def get_queryset(self):
         return ProviderGroup.objects.all()
@@ -897,41 +883,28 @@ class ProviderViewSet(BaseRLSViewSet):
         "inserted_at",
         "updated_at",
     ]
-    required_permissions = []
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_PROVIDERS]
 
-    def initial(self, request, *args, **kwargs):
-        """
-        Sets required_permissions before permissions are checked.
-        """
-        self.required_permissions = self.get_required_permissions()
-        super().initial(request, *args, **kwargs)
-
-    def get_required_permissions(self):
+    def set_required_permissions(self):
         """
         Returns the required permissions based on the request method.
         """
         if self.request.method in SAFE_METHODS:
             # No permissions required for GET requests
-            return []
+            self.required_permissions = []
         else:
             # Require permission for non-GET requests
-            return [Permissions.MANAGE_PROVIDERS]
+            self.required_permissions = [Permissions.MANAGE_PROVIDERS]
 
     def get_queryset(self):
-        user = self.request.user
-        user_roles = user.roles.all()
-        if getattr(user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False):
+        user_roles = get_role(self.request.user)
+        if user_roles.unlimited_visibility:
             # User has unlimited visibility, return all providers
             return Provider.objects.all()
 
         # User lacks permission, filter providers based on provider groups associated with the role
-        provider_groups = user_roles[0].provider_groups.all()
-        providers = Provider.objects.filter(
-            provider_groups__in=provider_groups
-        ).distinct()
-
-        return providers
+        return get_providers(user_roles)
 
     def get_serializer_class(self):
         if self.action == "create":
@@ -1051,40 +1024,28 @@ class ScanViewSet(BaseRLSViewSet):
         "inserted_at",
         "updated_at",
     ]
+    # RBAC required permissions
     required_permissions = [Permissions.MANAGE_SCANS]
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
 
-    def initial(self, request, *args, **kwargs):
-        """
-        Sets required_permissions before permissions are checked.
-        """
-        self.required_permissions = self.get_required_permissions()
-        super().initial(request, *args, **kwargs)
-
-    def get_required_permissions(self):
+    def set_required_permissions(self):
         """
         Returns the required permissions based on the request method.
         """
         if self.request.method in SAFE_METHODS:
             # No permissions required for GET requests
-            return []
+            self.required_permissions = [Permissions.MANAGE_PROVIDERS]
         else:
             # Require permission for non-GET requests
-            return [Permissions.MANAGE_SCANS]
+            self.required_permissions = [Permissions.MANAGE_SCANS]
 
     def get_queryset(self):
-        user = self.request.user
-        user_roles = user.roles.all()
-        if getattr(user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False):
+        user_roles = get_role(self.request.user)
+        if user_roles.unlimited_visibility:
             # User has unlimited visibility, return all scans
             return Scan.objects.all()
 
         # User lacks permission, filter providers based on provider groups associated with the role
-        provider_groups = user_roles[0].provider_groups.all()
-        providers = Provider.objects.filter(
-            provider_groups__in=provider_groups
-        ).distinct()
-        return Scan.objects.filter(provider__in=providers).distinct()
+        return Scan.objects.filter(provider__in=get_providers(user_roles))
 
     def get_serializer_class(self):
         if self.action == "create":
@@ -1174,26 +1135,14 @@ class TaskViewSet(BaseRLSViewSet):
     search_fields = ["name"]
     ordering = ["-inserted_at"]
     ordering_fields = ["inserted_at", "completed_at", "name", "state"]
+    # RBAC required permissions
     required_permissions = []
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
 
     def get_queryset(self):
-        user = self.request.user
-        user_roles = user.roles.all()
-        if getattr(user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False):
-            # User has unlimited visibility, return all tasks
-            return Task.objects.annotate(
-                name=F("task_runner_task__task_name"),
-                state=F("task_runner_task__status"),
-            )
-
-        # User lacks permission, filter tasks based on provider groups associated with the role
-        provider_groups = user_roles[0].provider_groups.all()
-        providers = Provider.objects.filter(
-            provider_groups__in=provider_groups
-        ).distinct()
-        scans = Scan.objects.filter(provider__in=providers).distinct()
-        return Task.objects.filter(scan__in=scans).distinct()
+        return Task.objects.annotate(
+            name=F("task_runner_task__task_name"),
+            state=F("task_runner_task__status"),
+        )
 
     def destroy(self, request, *args, pk=None, **kwargs):
         task = get_object_or_404(Task, pk=pk)
@@ -1254,29 +1203,17 @@ class ResourceViewSet(BaseRLSViewSet):
         "inserted_at",
         "updated_at",
     ]
+    # RBAC required permissions (implicit -> MANAGE_PROVIDERS enable unlimited visibility or check the visibility of the provider through the provider group)
     required_permissions = []
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
-
-    def initial(self, request, *args, **kwargs):
-        """
-        Sets required_permissions before permissions are checked.
-        """
-        self.required_permissions = ResourceViewSet.required_permissions
-        super().initial(request, *args, **kwargs)
 
     def get_queryset(self):
-        user = self.request.user
-        user_roles = user.roles.all()
-        if getattr(user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False):
+        user_roles = get_role(self.request.user)
+        if user_roles.unlimited_visibility:
             # User has unlimited visibility, return all scans
             queryset = Resource.objects.all()
         else:
             # User lacks permission, filter providers based on provider groups associated with the role
-            provider_groups = user_roles[0].provider_groups.all()
-            providers = Provider.objects.filter(
-                provider_groups__in=provider_groups
-            ).distinct()
-            queryset = Resource.objects.filter(provider__in=providers).distinct()
+            queryset = Resource.objects.filter(provider__in=get_providers(user_roles))
 
         search_value = self.request.query_params.get("filter[search]", None)
         if search_value:
@@ -1349,15 +1286,8 @@ class FindingViewSet(BaseRLSViewSet):
         "inserted_at",
         "updated_at",
     ]
+    # RBAC required permissions (implicit -> MANAGE_PROVIDERS enable unlimited visibility or check the visibility of the provider through the provider group)
     required_permissions = []
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
-
-    def initial(self, request, *args, **kwargs):
-        """
-        Sets required_permissions before permissions are checked.
-        """
-        self.required_permissions = ResourceViewSet.required_permissions
-        super().initial(request, *args, **kwargs)
 
     def get_serializer_class(self):
         if self.action == "findings_services_regions":
@@ -1366,19 +1296,15 @@ class FindingViewSet(BaseRLSViewSet):
         return super().get_serializer_class()
 
     def get_queryset(self):
-        user = self.request.user
-        user_roles = user.roles.all()
-        if getattr(user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False):
+        user_roles = get_role(self.request.user)
+        if user_roles.unlimited_visibility:
             # User has unlimited visibility, return all scans
             queryset = Finding.objects.all()
         else:
             # User lacks permission, filter providers based on provider groups associated with the role
-            provider_groups = user_roles[0].provider_groups.all()
-            providers = Provider.objects.filter(
-                provider_groups__in=provider_groups
-            ).distinct()
-            scans = Scan.objects.filter(provider__in=providers).distinct()
-            queryset = Finding.objects.filter(scan__in=scans).distinct()
+            queryset = Finding.objects.filter(
+                scan__provider__in=get_providers(user_roles)
+            )
 
         search_value = self.request.query_params.get("filter[search]", None)
         if search_value:
@@ -1476,6 +1402,8 @@ class ProviderSecretViewSet(BaseRLSViewSet):
         "inserted_at",
         "updated_at",
     ]
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_PROVIDERS]
 
     def get_queryset(self):
         return ProviderSecret.objects.all()
@@ -1533,8 +1461,8 @@ class InvitationViewSet(BaseRLSViewSet):
         "state",
         "inviter",
     ]
+    # RBAC required permissions
     required_permissions = [Permissions.MANAGE_ACCOUNT]
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
 
     def get_queryset(self):
         return Invitation.objects.all()
@@ -1672,8 +1600,8 @@ class RoleViewSet(BaseRLSViewSet):
     filterset_class = RoleFilter
     http_method_names = ["get", "post", "patch", "delete"]
     ordering = ["inserted_at"]
+    # RBAC required permissions
     required_permissions = [Permissions.MANAGE_ACCOUNT]
-    permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
 
     def get_queryset(self):
         return Role.objects.all()
@@ -1686,8 +1614,7 @@ class RoleViewSet(BaseRLSViewSet):
         return super().get_serializer_class()
 
     def partial_update(self, request, *args, **kwargs):
-        user = request.user
-        user_role = user.roles.all().first()
+        user_role = get_role(request.user)
         # If the user is the owner of the role, the manage_account field is not editable
         if user_role and kwargs["pk"] == str(user_role.id):
             request.data["manage_account"] = str(user_role.manage_account).lower()
@@ -1733,6 +1660,8 @@ class RoleProviderGroupRelationshipView(RelationshipView, BaseRLSViewSet):
     resource_name = "provider_groups"
     http_method_names = ["post", "patch", "delete"]
     schema = RelationshipViewSchema()
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_ACCOUNT]
 
     def get_queryset(self):
         return Role.objects.all()
@@ -1818,12 +1747,32 @@ class ComplianceOverviewViewSet(BaseRLSViewSet):
     search_fields = ["compliance_id"]
     ordering = ["compliance_id"]
     ordering_fields = ["inserted_at", "compliance_id", "framework", "region"]
+    # RBAC required permissions (implicit -> MANAGE_PROVIDERS enable unlimited visibility or check the visibility of the provider through the provider group)
+    required_permissions = []
 
     def get_queryset(self):
-        if self.action == "retrieve":
-            return ComplianceOverview.objects.all()
+        role = get_role(self.request.user)
+        unlimited_visibility = getattr(
+            role, Permissions.UNLIMITED_VISIBILITY.value, False
+        )
 
-        base_queryset = self.filter_queryset(ComplianceOverview.objects.all())
+        if self.action == "retrieve":
+            if unlimited_visibility:
+                # User has unlimited visibility, return all compliance compliances
+                return ComplianceOverview.objects.all()
+
+            providers = get_providers(role)
+            return ComplianceOverview.objects.filter(scan__provider__in=providers)
+
+        if unlimited_visibility:
+            base_queryset = self.filter_queryset(ComplianceOverview.objects.all())
+        else:
+            providers = Provider.objects.filter(
+                provider_groups__in=role.provider_groups.all()
+            ).distinct()
+            base_queryset = self.filter_queryset(
+                ComplianceOverview.objects.filter(scan__provider__in=providers)
+            )
 
         max_failed_ids = (
             base_queryset.filter(compliance_id=OuterRef("compliance_id"))
@@ -1831,12 +1780,10 @@ class ComplianceOverviewViewSet(BaseRLSViewSet):
             .values("id")[:1]
         )
 
-        queryset = base_queryset.filter(id__in=Subquery(max_failed_ids)).order_by(
+        return base_queryset.filter(id__in=Subquery(max_failed_ids)).order_by(
             "compliance_id"
         )
 
-        return queryset
-
     def get_serializer_class(self):
         if self.action == "retrieve":
             return ComplianceOverviewFullSerializer
@@ -1903,16 +1850,22 @@ class OverviewViewSet(BaseRLSViewSet):
     queryset = ComplianceOverview.objects.all()
     http_method_names = ["get"]
     ordering = ["-id"]
+    # RBAC required permissions (implicit -> MANAGE_PROVIDERS enable unlimited visibility or check the visibility of the provider through the provider group)
+    required_permissions = []
 
     def get_queryset(self):
+        role = get_role(self.request.user)
+        providers = get_providers(role)
+
+        def _get_filtered_queryset(model):
+            if role.unlimited_visibility:
+                return model.objects.all()
+            return model.objects.filter(scan__provider__in=providers)
+
         if self.action == "providers":
-            return Finding.objects.all()
-        elif self.action == "findings":
-            return ScanSummary.objects.all()
-        elif self.action == "findings_severity":
-            return ScanSummary.objects.all()
-        elif self.action == "services":
-            return ScanSummary.objects.all()
+            return _get_filtered_queryset(Finding)
+        elif self.action in ("findings", "findings_severity", "services"):
+            return _get_filtered_queryset(ScanSummary)
         else:
             return super().get_queryset()
 
@@ -2126,6 +2079,8 @@ class ScheduleViewSet(BaseRLSViewSet):
     # TODO: change to Schedule when implemented
     queryset = Task.objects.none()
     http_method_names = ["post"]
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_SCANS]
 
     def get_queryset(self):
         return super().get_queryset()