From 681141ef31f0fba2c30a62f526f9c671b20e7c9f Mon Sep 17 00:00:00 2001 From: Sam Vader Date: Mon, 24 Aug 2026 12:24:56 -0500 Subject: [PATCH 1/2] Align authorization queryset filters with the object-level permission check The queryset filters scoped results by product membership for every action intent. The object-level check gates some intents on is_staff instead, so a route that relied on the queryset filter alone applied a weaker rule than the object-level check applied on the same operation. Return an empty queryset for those intents instead, so both layers agree. --- dojo/authorization/query_registrations.py | 28 +++++++++-- unittests/test_bulk_finding_authorization.py | 49 ++++++++++++++++++++ 2 files changed, 73 insertions(+), 4 deletions(-) diff --git a/dojo/authorization/query_registrations.py b/dojo/authorization/query_registrations.py index 7f2bf455008..c4927e83cbb 100644 --- a/dojo/authorization/query_registrations.py +++ b/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)) diff --git a/unittests/test_bulk_finding_authorization.py b/unittests/test_bulk_finding_authorization.py index 2dcd3e70a57..874e12bb085 100644 --- a/unittests/test_bulk_finding_authorization.py +++ b/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)) From de049f2775a5d73718c0d2ade1ffc6c42b60a95d Mon Sep 17 00:00:00 2001 From: Sam Vader Date: Mon, 24 Aug 2026 14:55:13 -0500 Subject: [PATCH 2/2] Update the location bulk delete regression tests for the aligned policy These two tests asserted that a non-staff member reaches the bulk delete route and removes their own reference. The object-level check treats that intent as staff only, which the test one row above already asserts for the single delete route, so the bulk route now removes nothing for that caller. The direct tests of the reference helper are unchanged. --- unittests/test_location_reference_scoped_writes.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/unittests/test_location_reference_scoped_writes.py b/unittests/test_location_reference_scoped_writes.py index acbc06d39b7..9c8c6bfa2c8 100644 --- a/unittests/test_location_reference_scoped_writes.py +++ b/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())