From 474b602d46913ca30dacb1a1acc101c2005e6d21 Mon Sep 17 00:00:00 2001 From: Sam Vader Date: Wed, 19 Aug 2026 17:16:51 -0500 Subject: [PATCH 1/2] Scope endpoint deletion to the caller's authorized products Hardening to the endpoint (Location) delete paths. A Location row is shared by every product that records the same value, so both delete paths now route through one helper that removes only the references the caller's products own. Adds regression tests. No functional change for correctly-permissioned users. --- dojo/location/models.py | 40 +++++- dojo/url/ui/views.py | 20 ++- .../test_location_cross_product_authz.py | 124 +++++++++++++++++- 3 files changed, 177 insertions(+), 7 deletions(-) diff --git a/dojo/location/models.py b/dojo/location/models.py index c20b6eaa34a..2805f8530d4 100644 --- a/dojo/location/models.py +++ b/dojo/location/models.py @@ -12,11 +12,13 @@ RESTRICT, CharField, DateTimeField, + Exists, ForeignKey, Index, JSONField, Model, OneToOneField, + OuterRef, Q, QuerySet, TextChoices, @@ -36,7 +38,7 @@ LocationQueryset, ) from dojo.location.status import FindingLocationStatus, ProductLocationStatus -from dojo.models import Dojo_User, Finding, Product, copy_model_util +from dojo.models import Dojo_User, DojoMeta, Finding, Product, copy_model_util from dojo.tools.locations import LocationAssociationData if TYPE_CHECKING: @@ -587,3 +589,39 @@ class Meta: def __str__(self) -> str: """Return the string representation of a LocationProductReference.""" return f"{self.location} - Product: {self.product} ({self.status})" + + +def delete_locations_for_products(locations: QuerySet[Location], products) -> None: + """ + Remove ``locations`` from ``products``, keeping rows that anything else still references. + + A Location is deduplicated on its value alone, so several products share one row and + deleting the row would cascade away the other products' references. Drop only the + references owned by ``products``, then delete the rows that nothing points at any more. + """ + # Materialize before deleting anything: callers pass a queryset filtered through the + # reference rows below, so it would re-evaluate to empty part-way through. + location_ids = list(locations.values_list("id", flat=True)) + if not location_ids: + return + + with transaction.atomic(): + LocationFindingReference.objects.filter( + location_id__in=location_ids, + finding__test__engagement__product__in=products, + ).delete() + LocationProductReference.objects.filter( + location_id__in=location_ids, + product__in=products, + ).delete() + DojoMeta.objects.filter( + location_id__in=location_ids, + location_product__in=products, + ).delete() + + unreferenced = Location.objects.filter(id__in=location_ids).exclude( + Exists(LocationProductReference.objects.filter(location=OuterRef("pk"))), + ).exclude( + Exists(LocationFindingReference.objects.filter(location=OuterRef("pk"))), + ) + Location.objects.filter(id__in=list(unreferenced.values_list("id", flat=True))).delete() diff --git a/dojo/url/ui/views.py b/dojo/url/ui/views.py index 70a27eb449d..ee9e4787c72 100644 --- a/dojo/url/ui/views.py +++ b/dojo/url/ui/views.py @@ -22,7 +22,12 @@ DojoMetaFormSet, ImportEndpointMetaForm, ) -from dojo.location.models import Location, LocationFindingReference, LocationProductReference +from dojo.location.models import ( + Location, + LocationFindingReference, + LocationProductReference, + delete_locations_for_products, +) from dojo.location.queries import annotate_location_counts_and_status, get_authorized_locations from dojo.location.status import FindingLocationStatus, ProductLocationStatus from dojo.models import DojoMeta, Finding, Product @@ -401,8 +406,10 @@ def delete_endpoint(request, location_id): if request.method == "POST": form = DeleteEndpointForm(request.POST, instance=location) if form.is_valid(): - # Delete the location, which will also cascade delete related findings and product references - location.delete() + delete_locations_for_products( + Location.objects.filter(id=location.id), + get_authorized_products(Permissions.Location_Delete, request.user), + ) messages.add_message( request, messages.SUCCESS, "Endpoint and relationships removed.", extra_tags="alert-success", ) @@ -528,8 +535,11 @@ def endpoint_bulk_update_all(request, product_id=None): locations = get_authorized_locations("delete", locations, request.user) skipped_location_count = total_location_count - locations.count() deleted_location_count = locations.count() - # This will also delete related finding and product location references via cascade - locations.delete() + if product_id is not None: + delete_products = Product.objects.filter(id=product_id) + else: + delete_products = get_authorized_products(Permissions.Location_Delete, request.user) + delete_locations_for_products(locations, delete_products) # Notify user if any locations were skipped due to lack of authorization if skipped_location_count > 0: add_error_message_to_response( diff --git a/unittests/test_location_cross_product_authz.py b/unittests/test_location_cross_product_authz.py index 0eaddd18b5c..602cb7ee89b 100644 --- a/unittests/test_location_cross_product_authz.py +++ b/unittests/test_location_cross_product_authz.py @@ -1,14 +1,19 @@ from django.urls import reverse +from django.utils import timezone from dojo.authorization.roles_permissions import Roles -from dojo.location.models import Location, LocationProductReference +from dojo.location.models import Location, LocationFindingReference, LocationProductReference from dojo.location.status import ProductLocationStatus from dojo.models import ( Dojo_User, + Engagement, + Finding, Product, Product_Member, Product_Type, Role, + Test, + Test_Type, User, ) from dojo.url.models import URL @@ -89,3 +94,120 @@ def test_delete_endpoint_cross_product_is_denied_and_persists(self): ) self.assertEqual(self.DENIED_STATUS, response.status_code) self.assertTrue(Location.objects.filter(pk=self.location_b.id).exists()) + + +@skip_unless_v3 +class SharedLocationDeleteScopingTest(DojoTestCase): + + """ + A Location row is deduplicated on its value, so several products share one row. + + Deleting the row takes every product's references with it. Recording a URL another + product already recorded attaches the caller's product to that existing row, which + is enough to pass the row-level authorization check. Removing an endpoint must + therefore drop only the caller's own references and keep a row anything else uses. + """ + + @classmethod + def setUpTestData(cls): + prod_type, _ = Product_Type.objects.get_or_create(name="LOC-Del PT") + writer_role = Role.objects.get(id=Roles.Writer) + + cls.product_a = Product.objects.create(name="LOC-Del Product A", description="A", prod_type=prod_type) + cls.product_b = Product.objects.create(name="LOC-Del Product B", description="B", prod_type=prod_type) + + cls.alice = User.objects.create_user( + username="loc_del_alice", + password="not-a-real-secret", # noqa: S106 - test fixture user + ) + Product_Member.objects.create(user=cls.alice, product=cls.product_a, role=writer_role) + cls.product_a.authorized_users.add(Dojo_User.objects.get(pk=cls.alice.pk)) + + engagement = Engagement.objects.create( + product=cls.product_b, name="LOC-Del eng", + target_start=timezone.now().date(), target_end=timezone.now().date(), + ) + test_type, _ = Test_Type.objects.get_or_create(name="LOC-Del scan") + test = Test.objects.create( + engagement=engagement, test_type=test_type, + target_start=timezone.now(), target_end=timezone.now(), + ) + cls.finding_b = Finding.objects.create( + test=test, title="LOC-Del Product B finding", severity="High", + numerical_severity="S1", active=True, verified=False, + ) + + def setUp(self): + super().setUp() + self.client.force_login(self.alice) + # Product B records the URL first, with a finding on it. + self.shared = URL.create_location_from_value("https://shared.example.test/secret").location + LocationProductReference.objects.create( + location=self.shared, product=self.product_b, status=ProductLocationStatus.Active, + ) + self.shared.associate_with_finding(self.finding_b, audit_time=timezone.now()) + # A row only Product A uses, to prove a legitimate delete still works. + self.own = URL.create_location_from_value("https://own.example.test/ok").location + LocationProductReference.objects.create( + location=self.own, product=self.product_a, status=ProductLocationStatus.Active, + ) + + def _graft(self): + """Record Product B's URL against Product A, which reuses Product B's row.""" + response = self.client.post( + reverse("add_endpoint_to_product", kwargs={"product_id": self.product_a.id}), + {"protocol": "https", "host": "shared.example.test", "path": "secret"}, + ) + self.assertIn(response.status_code, (200, 302)) + self.assertTrue( + LocationProductReference.objects.filter(location=self.shared, product=self.product_a).exists(), + ) + + def _assert_product_b_intact(self): + self.assertTrue(Location.objects.filter(pk=self.shared.id).exists()) + self.assertTrue( + LocationProductReference.objects.filter(location=self.shared, product=self.product_b).exists(), + ) + self.assertTrue( + LocationFindingReference.objects.filter(location=self.shared, finding=self.finding_b).exists(), + ) + + def test_bulk_delete_keeps_the_other_products_shared_row(self): + self._graft() + response = self.client.post( + reverse("endpoints_bulk_all"), + {"endpoints_to_update": [self.shared.id], "delete_bulk_endpoints": "1"}, + ) + self.assertIn(response.status_code, (200, 302)) + self._assert_product_b_intact() + self.assertFalse( + LocationProductReference.objects.filter(location=self.shared, product=self.product_a).exists(), + ) + + def test_single_delete_keeps_the_other_products_shared_row(self): + self._graft() + response = self.client.post( + reverse("delete_endpoint", kwargs={"location_id": self.shared.id}), + {"id": self.shared.id}, + ) + self.assertIn(response.status_code, (200, 302)) + self._assert_product_b_intact() + self.assertFalse( + LocationProductReference.objects.filter(location=self.shared, product=self.product_a).exists(), + ) + + def test_bulk_delete_still_removes_a_row_only_the_caller_uses(self): + response = self.client.post( + reverse("endpoints_bulk_all"), + {"endpoints_to_update": [self.own.id], "delete_bulk_endpoints": "1"}, + ) + self.assertIn(response.status_code, (200, 302)) + self.assertFalse(Location.objects.filter(pk=self.own.id).exists()) + + def test_single_delete_still_removes_a_row_only_the_caller_uses(self): + response = self.client.post( + reverse("delete_endpoint", kwargs={"location_id": self.own.id}), + {"id": self.own.id}, + ) + self.assertIn(response.status_code, (200, 302)) + self.assertFalse(Location.objects.filter(pk=self.own.id).exists()) From c38c721d58752cd1ebb1b25edeb649f1c2bfad4b Mon Sep 17 00:00:00 2001 From: svader0 Date: Thu, 20 Aug 2026 16:18:02 -0500 Subject: [PATCH 2/2] test(location): fix the shared-location delete scoping tests Two problems made the v3 rest-framework job red. Finding.reporter defaults to pk 1, and this test class creates no such user. Postgres defers the FK, so the four tests only blew up in _fixture_teardown. Pass the test user as the reporter. The two single-delete tests posted to delete_endpoint and got 400. Under legacy authorization user_has_permission maps Location_Delete to Action.Delete, which is staff-only, so a non-staff product member never reaches that view. Making the user staff does not help either, because a staff user is unrestricted and every product lands in scope, which is the opposite of what the test checks. Call the helper both delete views share instead, with the product scope the views pass it. --- .../test_location_cross_product_authz.py | 31 ++++++++++++------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/unittests/test_location_cross_product_authz.py b/unittests/test_location_cross_product_authz.py index 602cb7ee89b..bf5bc7bc628 100644 --- a/unittests/test_location_cross_product_authz.py +++ b/unittests/test_location_cross_product_authz.py @@ -2,7 +2,12 @@ from django.utils import timezone from dojo.authorization.roles_permissions import Roles -from dojo.location.models import Location, LocationFindingReference, LocationProductReference +from dojo.location.models import ( + Location, + LocationFindingReference, + LocationProductReference, + delete_locations_for_products, +) from dojo.location.status import ProductLocationStatus from dojo.models import ( Dojo_User, @@ -135,6 +140,7 @@ def setUpTestData(cls): cls.finding_b = Finding.objects.create( test=test, title="LOC-Del Product B finding", severity="High", numerical_severity="S1", active=True, verified=False, + reporter=cls.alice, ) def setUp(self): @@ -184,13 +190,17 @@ def test_bulk_delete_keeps_the_other_products_shared_row(self): LocationProductReference.objects.filter(location=self.shared, product=self.product_a).exists(), ) - def test_single_delete_keeps_the_other_products_shared_row(self): + # The single-endpoint delete view is not reachable with a scoped user under legacy + # authorization: user_has_permission maps Location_Delete to Action.Delete, which is + # staff-only, and a staff user is unrestricted so every product is in scope. The two + # tests below therefore call the helper both views share, with the product scope the + # views hand it. + def test_scoped_delete_keeps_the_other_products_shared_row(self): self._graft() - response = self.client.post( - reverse("delete_endpoint", kwargs={"location_id": self.shared.id}), - {"id": self.shared.id}, + delete_locations_for_products( + Location.objects.filter(id=self.shared.id), + Product.objects.filter(id=self.product_a.id), ) - self.assertIn(response.status_code, (200, 302)) self._assert_product_b_intact() self.assertFalse( LocationProductReference.objects.filter(location=self.shared, product=self.product_a).exists(), @@ -204,10 +214,9 @@ def test_bulk_delete_still_removes_a_row_only_the_caller_uses(self): self.assertIn(response.status_code, (200, 302)) self.assertFalse(Location.objects.filter(pk=self.own.id).exists()) - def test_single_delete_still_removes_a_row_only_the_caller_uses(self): - response = self.client.post( - reverse("delete_endpoint", kwargs={"location_id": self.own.id}), - {"id": self.own.id}, + def test_scoped_delete_still_removes_a_row_only_the_caller_uses(self): + delete_locations_for_products( + Location.objects.filter(id=self.own.id), + Product.objects.filter(id=self.product_a.id), ) - self.assertIn(response.status_code, (200, 302)) self.assertFalse(Location.objects.filter(pk=self.own.id).exists())