diff --git a/api/src/backend/api/base_views.py b/api/src/backend/api/base_views.py index 11f1d4bf5f..3e965482df 100644 --- a/api/src/backend/api/base_views.py +++ b/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 diff --git a/api/src/backend/api/rbac/permissions.py b/api/src/backend/api/rbac/permissions.py index f1a9b184ae..abc435846b 100644 --- a/api/src/backend/api/rbac/permissions.py +++ b/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() diff --git a/api/src/backend/api/tests/test_views.py b/api/src/backend/api/tests/test_views.py index 794dae7137..9b06381d8f 100644 --- a/api/src/backend/api/tests/test_views.py +++ b/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": [ diff --git a/api/src/backend/api/v1/views.py b/api/src/backend/api/v1/views.py index 600d313ade..948439b88e 100644 --- a/api/src/backend/api/v1/views.py +++ b/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()