Align authorization queryset filters with the object-level permission check

dojo/authorization/query_registrations.py

@@ -16,7 +16,7 @@
 from django.db.models import Q
 
 from dojo.authorization.query_filters import register_auth_filter
-from dojo.authorization.roles_permissions import permission_to_action
+from dojo.authorization.roles_permissions import Action, permission_to_action
 from dojo.location.models import Location, LocationFindingReference, LocationProductReference
 from dojo.models import (
     App_Analysis,
@@ -62,6 +62,15 @@ def _is_unrestricted(user, action):
     return bool(user.is_staff)
 
 
+def _requires_staff(action):
+    """
+    Actions ``user_has_permission`` gates on is_staff rather than on
+    membership. Only reached after the superuser/staff bypass above, so
+    membership must not scope a queryset for these.
+    """
+    return action in {Action.Delete, Action.StaffOnly, Action.SuperuserOnly}
+
+
 def _authorized_product_ids(user):
     """
     QuerySet of product ids the user can access via authorized_users.
@@ -126,6 +135,8 @@ def _filter_by_authorized_products(queryset, product_path, permission, user=None
     action = permission_to_action(permission)
     if _is_unrestricted(user, action):
         return queryset
+    if _requires_staff(action):
+        return queryset.none()
     return queryset.filter(**{f"{product_path}__id__in": _authorized_product_ids(user)})
 
 
@@ -138,8 +149,11 @@ def _get_authorized_products(permission, user=None):
     user = _resolve_user(user)
     if user is None or getattr(user, "is_anonymous", False):
         return Product.objects.none()
-    if _is_unrestricted(user, permission_to_action(permission)):
+    action = permission_to_action(permission)
+    if _is_unrestricted(user, action):
         return Product.objects.all().order_by("name")
+    if _requires_staff(action):
+        return Product.objects.none()
     return Product.objects.filter(
         Q(authorized_users=user) | Q(prod_type__authorized_users=user),
     ).distinct().order_by("name")
@@ -314,8 +328,11 @@ def _get_authorized_locations(permission, queryset=None, user=None):
     qs = queryset if queryset is not None else Location.objects.all()
     if user is None or getattr(user, "is_anonymous", False):
         return qs.none()
-    if _is_unrestricted(user, permission_to_action(permission)):
+    action = permission_to_action(permission)
+    if _is_unrestricted(user, action):
         return qs
+    if _requires_staff(action):
+        return qs.none()
     authorized_products = _authorized_product_ids(user)
     return qs.filter(products__product__id__in=authorized_products).distinct()
 
@@ -396,8 +413,11 @@ def _get_authorized_findings(permission, queryset=None, user=None):
     qs = queryset if queryset is not None else Finding.objects.all()
     if user is None or getattr(user, "is_anonymous", False):
         return qs.none()
-    if _is_unrestricted(user, permission_to_action(permission)):
+    action = permission_to_action(permission)
+    if _is_unrestricted(user, action):
         return qs
+    if _requires_staff(action):
+        return qs.none()
     return qs.filter(test__engagement__product__id__in=_authorized_product_ids(user))
 
 

unittests/test_bulk_finding_authorization.py

@@ -10,6 +10,7 @@
 
 from django.urls import reverse
 
+from dojo.authorization.authorization import user_has_permission
 from dojo.models import Dojo_User, Finding, Finding_Group, Test
 
 from .dojo_test_case import DojoTestCase, versioned_fixtures
@@ -84,3 +85,51 @@ def test_scoped_user_cannot_add_finding_to_other_products_group(self):
             list(other_group.findings.values_list("id", flat=True)),
             msg="scoped user added a finding to a group outside their authorized products",
         )
+
+
+@versioned_fixtures
+class TestBulkFindingDeleteRequiresStaff(DojoTestCase):
+
+    """
+    The bulk route must apply the same staff-only delete policy the single
+    delete view applies. Product membership alone authorizes viewing and
+    editing a finding, not deleting it.
+    """
+
+    fixtures = ["dojo_testdata.json"]
+
+    def setUp(self):
+        super().setUp()
+        self.product = Test.objects.get(id=3).engagement.product
+        self.finding = Finding.objects.filter(
+            test__engagement__product=self.product,
+        ).first()
+        self.assertIsNotNone(self.finding)
+
+    def _bulk_delete_as(self, user):
+        self.product.authorized_users.add(user)
+        self.client.force_login(user)
+        response = self.client.post(reverse("finding_bulk_update_all"), {
+            "finding_to_update": [self.finding.id],
+            "delete_bulk_findings": "1",
+        })
+        self.assertLess(response.status_code, 500)
+        return Finding.objects.filter(id=self.finding.id).exists()
+
+    def test_member_cannot_bulk_delete_findings_in_their_own_product(self):
+        member = Dojo_User.objects.create(
+            username="bulk_delete_member", is_active=True, is_staff=False,
+        )
+        self.product.authorized_users.add(member)
+        self.assertTrue(user_has_permission(member, self.finding, "edit"))
+        self.assertFalse(user_has_permission(member, self.finding, "delete"))
+        self.assertTrue(
+            self._bulk_delete_as(member),
+            msg="non-staff member deleted a finding the single delete view denies them",
+        )
+
+    def test_staff_can_still_bulk_delete_findings(self):
+        staff = Dojo_User.objects.create(
+            username="bulk_delete_staff", is_active=True, is_staff=True,
+        )
+        self.assertFalse(self._bulk_delete_as(staff))

unittests/test_location_reference_scoped_writes.py

@@ -137,17 +137,21 @@ def test_single_delete_route_stays_denied_without_the_delete_action(self):
         self.assertEqual(response.status_code, 400)
         self.assertEqual(self._refs(self.shared), {self.product_mine.name, self.product_outside.name})
 
-    def test_bulk_delete_removes_only_the_callers_reference(self):
+    def test_bulk_delete_stays_denied_without_the_delete_action(self):
+        # Same policy as the single delete route above: membership does not grant delete,
+        # so the bulk route removes nothing rather than removing the caller's reference.
         self.client.post(
             reverse("endpoints_bulk_all"),
             {"endpoints_to_update": [self.shared.id], "delete_bulk_endpoints": "1"},
         )
         self.assertTrue(Location.objects.filter(id=self.shared.id).exists())
-        self.assertEqual(self._refs(self.shared), {self.product_outside.name})
+        self.assertEqual(
+            self._refs(self.shared), {self.product_mine.name, self.product_outside.name},
+        )
 
-    def test_bulk_delete_removes_the_row_when_nothing_else_references_it(self):
+    def test_bulk_delete_stays_denied_for_a_row_only_the_caller_records(self):
         self.client.post(
             reverse("endpoints_bulk_all"),
             {"endpoints_to_update": [self.own.id], "delete_bulk_endpoints": "1"},
         )
-        self.assertFalse(Location.objects.filter(id=self.own.id).exists())
+        self.assertTrue(Location.objects.filter(id=self.own.id).exists())