fix(api): require operation permission to revoke tasks (#12893)

api/changelog.d/task-revoke-permissions.security.md

@@ -0,0 +1 @@
+`DELETE /api/v1/tasks/{id}` requires the permission of the operation that queued the task and rejects provider deletions, and `GET /api/v1/tasks` hides tasks of providers outside the visibility of the role

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

@@ -1,7 +1,7 @@
 from enum import Enum
 
 from api.db_router import MainRouter
-from api.models import Integration, Provider, Role, User
+from api.models import Integration, Provider, Role, Task, User
 from django.db.models import Q, QuerySet
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.permissions import BasePermission
@@ -17,6 +17,50 @@ class Permissions(Enum):
     UNLIMITED_VISIBILITY = "unlimited_visibility"
 
 
+# Revoking a task needs the permission of the operation that queued it.
+# None and unmapped names are not revocable; a revoked provider deletion
+# would leave the provider soft-deleted with nothing re-queuing the cleanup.
+TASK_REVOKE_PERMISSIONS: dict[str, list[Permissions] | None] = {
+    "provider-connection-check": [Permissions.MANAGE_PROVIDERS],
+    "provider-deletion": None,
+    "integration-connection-check": [Permissions.MANAGE_INTEGRATIONS],
+    "integration-s3": [Permissions.MANAGE_INTEGRATIONS],
+    "integration-security-hub": [Permissions.MANAGE_INTEGRATIONS],
+    "integration-jira": [Permissions.MANAGE_INTEGRATIONS],
+    "scan-perform": [Permissions.MANAGE_SCANS],
+    "scan-perform-scheduled": [Permissions.MANAGE_SCANS],
+    "scan-compliance-overviews": [Permissions.MANAGE_SCANS],
+    "scan-compliance-reports": [Permissions.MANAGE_SCANS],
+    "scan-finding-group-summaries": [Permissions.MANAGE_SCANS],
+    "scan-report": [Permissions.MANAGE_SCANS],
+    "attack-paths-scan-perform": [Permissions.MANAGE_SCANS],
+    "findings-mute-latest-scans": [Permissions.MANAGE_SCANS],
+    "lighthouse-connection-check": [],
+    "lighthouse-provider-connection-check": [],
+    "lighthouse-provider-models-refresh": [],
+}
+
+
+def get_user_roles(user: User, tenant_id: str) -> list[Role]:
+    """Return every role assigned to the user in the tenant."""
+    return list(
+        User.objects.using(MainRouter.admin_db)
+        .get(id=user.id)
+        .roles.using(MainRouter.admin_db)
+        .filter(tenant_id=tenant_id)
+    )
+
+
+def roles_have_permissions(
+    roles: list[Role], required_permissions: list[Permissions]
+) -> bool:
+    """Return True when every required permission is granted by at least one role."""
+    return all(
+        any(getattr(role, permission.value, False) for role in roles)
+        for permission in required_permissions
+    )
+
+
 class HasPermissions(BasePermission):
     """
     Custom permission to check if the user's role has the required permissions.
@@ -34,19 +78,11 @@ class HasPermissions(BasePermission):
         if not tenant_id:
             return False
 
-        user_roles = list(
-            User.objects.using(MainRouter.admin_db)
-            .get(id=request.user.id)
-            .roles.using(MainRouter.admin_db)
-            .filter(tenant_id=tenant_id)
-        )
+        user_roles = get_user_roles(request.user, tenant_id)
         if not user_roles:
             return False
 
-        return all(
-            any(getattr(role, permission.value, False) for role in user_roles)
-            for permission in required_permissions
-        )
+        return roles_have_permissions(user_roles, required_permissions)
 
 
 def get_role(user: User, tenant_id: str) -> Role:
@@ -85,6 +121,25 @@ def get_providers(role: Role) -> QuerySet[Provider]:
     ).distinct()
 
 
+def get_tasks(role: Role) -> QuerySet[Task]:
+    """Return the tasks visible to the role: tenant-wide ones and those of its providers."""
+    queryset = Task.objects.filter(tenant_id=role.tenant_id)
+    if role.unlimited_visibility:
+        return queryset
+
+    # Task has no provider FK, so match provider ids inside the stored kwargs.
+    # all_objects keeps a soft-deleted provider visible to its own groups, so the
+    # role that queued its deletion can still follow the task.
+    hidden = Q()
+    for provider_id in (
+        Provider.all_objects.filter(tenant_id=role.tenant_id)
+        .exclude(provider_groups__in=role.provider_groups.all())
+        .values_list("id", flat=True)
+    ):
+        hidden |= Q(task_runner_task__task_kwargs__contains=str(provider_id))
+    return queryset.exclude(hidden) if hidden else queryset
+
+
 def get_integrations(
     role: Role, providers: QuerySet[Provider] | None = None
 ) -> QuerySet[Integration]:

api/src/backend/api/specs/v1.yaml

@@ -14823,7 +14823,9 @@ paths:
     get:
       operationId: api_v1_tasks_list
       description: Retrieve a list of all tasks with options for filtering by name,
-        state, and other criteria.
+        state, and other criteria. Tasks that reference a provider are only returned
+        when the role can access it; tasks without a provider reference are returned
+        for every role.
       summary: List all tasks
       parameters:
       - in: query
@@ -14922,7 +14924,8 @@ paths:
   /api/v1/tasks/{id}:
     get:
       operationId: api_v1_tasks_retrieve
-      description: Fetch detailed information about a specific task by its ID.
+      description: Fetch detailed information about a specific task by its ID. Tasks
+        tied to a provider outside the visibility of the role are not found.
       summary: Retrieve data from a specific task
       parameters:
       - in: query
@@ -14963,7 +14966,9 @@ paths:
     delete:
       operationId: api_v1_tasks_destroy
       description: Try to revoke a task using its ID. Only tasks that are not yet
-        in progress can be revoked.
+        in progress can be revoked, and the caller needs the same permission as the
+        operation that queued the task (for example MANAGE_SCANS for a scan). Provider
+        deletions cannot be revoked.
       summary: Revoke a task
       parameters:
       - in: path

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

@@ -60,6 +60,7 @@ from api.models import (
     User,
     UserRoleRelationship,
 )
+from api.rbac.permissions import TASK_REVOKE_PERMISSIONS
 from api.rls import Tenant
 from api.uuid_utils import datetime_to_uuid7
 from api.v1.views import (
@@ -5239,6 +5240,7 @@ class TestTaskViewSet:
     @patch("api.v1.views.AsyncResult", return_value=Mock())
     def test_tasks_revoke(self, mock_async_result, authenticated_client, tasks_fixture):
         _, task2 = tasks_fixture
+        self._set_task_name(task2, "scan-perform")
         response = authenticated_client.delete(
             reverse("task-detail", kwargs={"pk": task2.id})
         )
@@ -5254,12 +5256,311 @@ class TestTaskViewSet:
 
     def test_tasks_revoke_invalid_status(self, authenticated_client, tasks_fixture):
         task1, _ = tasks_fixture
+        self._set_task_name(task1, "scan-perform")
         response = authenticated_client.delete(
             reverse("task-detail", kwargs={"pk": task1.id})
         )
         # Task status is SUCCESS
         assert response.status_code == status.HTTP_400_BAD_REQUEST
 
+    @staticmethod
+    def _set_task_name(task, name):
+        task.task_runner_task.task_name = name
+        task.task_runner_task.save(update_fields=["task_name"])
+
+    @staticmethod
+    def _set_task_kwargs(task, kwargs):
+        task.task_runner_task.task_kwargs = json.dumps(repr(kwargs))
+        task.task_runner_task.save(update_fields=["task_kwargs"])
+
+    @staticmethod
+    def _client_with_role(tenant, factory, **permissions):
+        user = User.objects.create_user(
+            name=f"revoker-{uuid4()}",
+            email=f"revoker-{uuid4()}@prowler.com",
+            password=TEST_PASSWORD,
+        )
+        Membership.objects.create(
+            user=user, tenant=tenant, role=Membership.RoleChoices.MEMBER
+        )
+        flags = {
+            "manage_users": False,
+            "manage_account": False,
+            "manage_billing": False,
+            "manage_providers": False,
+            "manage_integrations": False,
+            "manage_scans": False,
+            "unlimited_visibility": True,
+            **permissions,
+        }
+        role = Role.objects.create(
+            name=f"revoker-{uuid4()}", tenant_id=tenant.id, **flags
+        )
+        UserRoleRelationship.objects.create(user=user, role=role, tenant_id=tenant.id)
+        return factory(user, tenant)
+
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_revoke_without_permission_is_forbidden(
+        self, mock_async_result, authenticated_client_no_permissions_rbac, tasks_fixture
+    ):
+        _, pending_task = tasks_fixture
+        self._set_task_name(pending_task, "provider-connection-check")
+
+        response = authenticated_client_no_permissions_rbac.delete(
+            reverse("task-detail", kwargs={"pk": pending_task.id})
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+        mock_async_result.return_value.revoke.assert_not_called()
+
+    @pytest.mark.parametrize(
+        "task_name, permissions, expected_status",
+        [
+            (
+                "provider-connection-check",
+                {"manage_providers": True},
+                status.HTTP_202_ACCEPTED,
+            ),
+            (
+                "provider-connection-check",
+                {"manage_scans": True},
+                status.HTTP_403_FORBIDDEN,
+            ),
+            ("scan-perform", {"manage_scans": True}, status.HTTP_202_ACCEPTED),
+            (
+                "scan-perform-scheduled",
+                {"manage_providers": True},
+                status.HTTP_403_FORBIDDEN,
+            ),
+            (
+                "integration-jira",
+                {"manage_integrations": True},
+                status.HTTP_202_ACCEPTED,
+            ),
+            ("integration-jira", {"manage_providers": True}, status.HTTP_403_FORBIDDEN),
+            ("lighthouse-connection-check", {}, status.HTTP_202_ACCEPTED),
+        ],
+    )
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_revoke_requires_originating_operation_permission(
+        self,
+        mock_async_result,
+        authenticated_client_for_tenant_factory,
+        tenants_fixture,
+        tasks_fixture,
+        task_name,
+        permissions,
+        expected_status,
+    ):
+        tenant, *_ = tenants_fixture
+        _, pending_task = tasks_fixture
+        self._set_task_name(pending_task, task_name)
+        client = self._client_with_role(
+            tenant, authenticated_client_for_tenant_factory, **permissions
+        )
+
+        response = client.delete(reverse("task-detail", kwargs={"pk": pending_task.id}))
+
+        assert response.status_code == expected_status
+        if expected_status == status.HTTP_202_ACCEPTED:
+            mock_async_result.return_value.revoke.assert_called_once()
+        else:
+            mock_async_result.return_value.revoke.assert_not_called()
+
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_revoke_provider_deletion_is_forbidden_even_for_admin(
+        self, mock_async_result, authenticated_client, tasks_fixture
+    ):
+        _, pending_task = tasks_fixture
+        self._set_task_name(pending_task, "provider-deletion")
+
+        response = authenticated_client.delete(
+            reverse("task-detail", kwargs={"pk": pending_task.id})
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+        mock_async_result.return_value.revoke.assert_not_called()
+
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_revoke_unmapped_task_is_forbidden(
+        self, mock_async_result, authenticated_client, tasks_fixture
+    ):
+        _, pending_task = tasks_fixture
+        assert pending_task.task_runner_task.task_name not in TASK_REVOKE_PERMISSIONS
+
+        response = authenticated_client.delete(
+            reverse("task-detail", kwargs={"pk": pending_task.id})
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+        mock_async_result.return_value.revoke.assert_not_called()
+
+    def test_every_rls_task_has_revoke_permissions(self):
+        from config.celery import RLSTask, celery_app
+
+        rls_task_names = {
+            name for name, task in celery_app.tasks.items() if isinstance(task, RLSTask)
+        }
+        assert rls_task_names
+        assert rls_task_names <= set(TASK_REVOKE_PERMISSIONS)
+
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_hidden_for_providers_outside_role_visibility(
+        self,
+        mock_async_result,
+        authenticated_client_no_permissions_rbac,
+        tasks_fixture,
+        aws_provider_pair,
+    ):
+        client = authenticated_client_no_permissions_rbac
+        limited_user = client.user
+        tenant = Membership.objects.filter(user=limited_user).first().tenant
+        allowed_provider, denied_provider = aws_provider_pair
+        allowed_task, denied_task = tasks_fixture
+        self._set_task_kwargs(
+            allowed_task,
+            {"tenant_id": str(tenant.id), "provider_id": str(allowed_provider.id)},
+        )
+        self._set_task_name(denied_task, "provider-deletion")
+        self._set_task_kwargs(
+            denied_task,
+            {"tenant_id": str(tenant.id), "provider_id": str(denied_provider.id)},
+        )
+        provider_group = ProviderGroup.objects.create(
+            name="limited-task-group", tenant_id=tenant.id
+        )
+        ProviderGroupMembership.objects.create(
+            tenant_id=tenant.id,
+            provider_group=provider_group,
+            provider=allowed_provider,
+        )
+        RoleProviderGroupRelationship.objects.create(
+            tenant_id=tenant.id,
+            role=limited_user.roles.first(),
+            provider_group=provider_group,
+        )
+
+        response = client.get(reverse("task-list"))
+        assert response.status_code == status.HTTP_200_OK
+        assert [item["id"] for item in response.json()["data"]] == [
+            str(allowed_task.id)
+        ]
+
+        response = client.get(reverse("task-detail", kwargs={"pk": denied_task.id}))
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+        response = client.delete(reverse("task-detail", kwargs={"pk": denied_task.id}))
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+        mock_async_result.return_value.revoke.assert_not_called()
+
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_of_soft_deleted_provider_stay_visible_to_its_groups(
+        self,
+        mock_async_result,
+        authenticated_client_for_tenant_factory,
+        tenants_fixture,
+        tasks_fixture,
+        aws_provider_pair,
+    ):
+        tenant, *_ = tenants_fixture
+        provider, _ = aws_provider_pair
+        finished_task, pending_task = tasks_fixture
+        client = self._client_with_role(
+            tenant,
+            authenticated_client_for_tenant_factory,
+            manage_providers=True,
+            unlimited_visibility=False,
+        )
+        provider_group = ProviderGroup.objects.create(
+            name="own-group", tenant_id=tenant.id
+        )
+        ProviderGroupMembership.objects.create(
+            tenant_id=tenant.id, provider_group=provider_group, provider=provider
+        )
+        RoleProviderGroupRelationship.objects.create(
+            tenant_id=tenant.id,
+            role=client.user.roles.first(),
+            provider_group=provider_group,
+        )
+        for task, name in (
+            (finished_task, "provider-deletion"),
+            (pending_task, "provider-connection-check"),
+        ):
+            self._set_task_name(task, name)
+            self._set_task_kwargs(
+                task, {"tenant_id": str(tenant.id), "provider_id": str(provider.id)}
+            )
+        provider.is_deleted = True
+        provider.save()
+
+        response = client.get(reverse("task-detail", kwargs={"pk": finished_task.id}))
+        assert response.status_code == status.HTTP_200_OK
+
+        response = client.delete(reverse("task-detail", kwargs={"pk": pending_task.id}))
+        assert response.status_code == status.HTTP_202_ACCEPTED
+        mock_async_result.return_value.revoke.assert_called_once()
+
+    def test_tasks_without_provider_stay_visible_for_limited_roles(
+        self, authenticated_client_no_permissions_rbac, tasks_fixture, aws_provider_pair
+    ):
+        response = authenticated_client_no_permissions_rbac.get(reverse("task-list"))
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()["data"]) == len(tasks_fixture)
+
+    def test_tasks_list_without_role_is_forbidden(
+        self, authenticated_client_rbac_noroles, tasks_fixture
+    ):
+        response = authenticated_client_rbac_noroles.get(reverse("task-list"))
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_tasks_revoke_without_permission_hides_task_status(
+        self, authenticated_client_no_permissions_rbac, tasks_fixture
+    ):
+        finished_task, _ = tasks_fixture
+        self._set_task_name(finished_task, "provider-connection-check")
+
+        response = authenticated_client_no_permissions_rbac.delete(
+            reverse("task-detail", kwargs={"pk": finished_task.id})
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_revoke_unauthenticated_returns_401(
+        self, mock_async_result, tasks_fixture
+    ):
+        from rest_framework.test import APIClient
+
+        _, pending_task = tasks_fixture
+        self._set_task_name(pending_task, "scan-perform")
+
+        response = APIClient().delete(
+            reverse("task-detail", kwargs={"pk": pending_task.id})
+        )
+
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+        mock_async_result.return_value.revoke.assert_not_called()
+
+    @patch("api.v1.views.AsyncResult")
+    def test_tasks_revoke_foreign_tenant_task_returns_404(
+        self,
+        mock_async_result,
+        authenticated_client_for_tenant_factory,
+        tenants_fixture,
+        tasks_fixture,
+    ):
+        _, foreign_tenant, *_ = tenants_fixture
+        _, pending_task = tasks_fixture
+        self._set_task_name(pending_task, "scan-perform")
+        client = self._client_with_role(
+            foreign_tenant, authenticated_client_for_tenant_factory, manage_scans=True
+        )
+
+        response = client.delete(reverse("task-detail", kwargs={"pk": pending_task.id}))
+
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+        mock_async_result.return_value.revoke.assert_not_called()
+
 
 @pytest.mark.django_db
 class TestAttackPathsScanViewSet:

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

@@ -125,10 +125,14 @@ from api.models import (
 )
 from api.pagination import ComplianceOverviewPagination
 from api.rbac.permissions import (
+    TASK_REVOKE_PERMISSIONS,
     Permissions,
     get_integrations,
     get_providers,
     get_role,
+    get_tasks,
+    get_user_roles,
+    roles_have_permissions,
 )
 from api.renderers import APIJSONRenderer, PlainTextRenderer
 from api.rls import Tenant
@@ -2858,17 +2862,29 @@ class ScanViewSet(ProviderVisibilityMixin, BaseRLSViewSet):
     list=extend_schema(
         tags=["Task"],
         summary="List all tasks",
-        description="Retrieve a list of all tasks with options for filtering by name, state, and other criteria.",
+        description=(
+            "Retrieve a list of all tasks with options for filtering by name, state, and other "
+            "criteria. Tasks that reference a provider are only returned when the role can "
+            "access it; tasks without a provider reference are returned for every role."
+        ),
     ),
     retrieve=extend_schema(
         tags=["Task"],
         summary="Retrieve data from a specific task",
-        description="Fetch detailed information about a specific task by its ID.",
+        description=(
+            "Fetch detailed information about a specific task by its ID. Tasks tied to a provider "
+            "outside the visibility of the role are not found."
+        ),
     ),
     destroy=extend_schema(
         tags=["Task"],
         summary="Revoke a task",
-        description="Try to revoke a task using its ID. Only tasks that are not yet in progress can be revoked.",
+        description=(
+            "Try to revoke a task using its ID. Only tasks that are not yet in progress can be "
+            "revoked, and the caller needs the same permission as the operation that queued "
+            "the task (for example MANAGE_SCANS for a scan). Provider deletions cannot be "
+            "revoked."
+        ),
         responses={202: OpenApiResponse(response=TaskSerializer)},
     ),
 )
@@ -2884,13 +2900,26 @@ class TaskViewSet(BaseRLSViewSet):
     required_permissions = []
 
     def get_queryset(self):
-        return Task.objects.annotate(
-            name=F("task_runner_task__task_name"),
-            state=F("task_runner_task__status"),
-        ).select_related("task_runner_task")
+        return (
+            get_tasks(self.user_role)
+            .annotate(
+                name=F("task_runner_task__task_name"),
+                state=F("task_runner_task__status"),
+            )
+            .select_related("task_runner_task")
+        )
 
     def destroy(self, request, *args, pk=None, **kwargs):
-        task = get_object_or_404(Task, pk=pk)
+        task = self.get_object()
+        required_permissions = TASK_REVOKE_PERMISSIONS.get(
+            task.task_runner_task.task_name
+        )
+        # Same multi-role semantics as HasPermissions.
+        if required_permissions is None or not roles_have_permissions(
+            get_user_roles(request.user, request.tenant_id), required_permissions
+        ):
+            raise PermissionDenied("You do not have permission to revoke this task.")
+
         if task.task_runner_task.status not in ["PENDING", "RECEIVED"]:
             serializer = TaskSerializer(task)
             return Response(