From 685f3ff67bd28212cb6bf818278d6e1c0a874a84 Mon Sep 17 00:00:00 2001 From: svader0 Date: Wed, 19 Aug 2026 17:43:25 -0500 Subject: [PATCH] fix(locations): scope endpoint delete and rename to the requesting product A Location row is deduplicated across every product that records the same value. Deleting the row removed it from products the caller has no rights over, and a rename rewrote the value those products see. Delete now removes the requesting product's own references and drops the row only once nothing references it, and a rename is refused while another product still records the value. --- dojo/location/queries.py | 42 +++++ dojo/url/ui/views.py | 29 +++- .../test_location_reference_scoped_writes.py | 153 ++++++++++++++++++ 3 files changed, 219 insertions(+), 5 deletions(-) create mode 100644 unittests/test_location_reference_scoped_writes.py diff --git a/dojo/location/queries.py b/dojo/location/queries.py index 583694a044..b6c3066025 100644 --- a/dojo/location/queries.py +++ b/dojo/location/queries.py @@ -1,9 +1,11 @@ import logging +from django.db import transaction from django.db.models import ( Case, CharField, Count, + Exists, F, IntegerField, OuterRef, @@ -70,6 +72,46 @@ def authorized_product_references(user=None): ) +def remove_location_references(locations, products): + """ + Drop ``products``' references to ``locations``, then delete any Location left + with none. + + A Location is deduplicated across every product that records the same value, so + deleting the row itself removes it from products the caller has no rights over. + The reference is the per-product object, so it is what a delete acts on. + """ + location_ids = list(locations.values_list("id", flat=True)) + if not location_ids: + return 0 + with transaction.atomic(): + LocationFindingReference.objects.filter( + location_id__in=location_ids, + finding__test__engagement__product__in=products, + ).delete() + removed = LocationProductReference.objects.filter( + location_id__in=location_ids, + product__in=products, + ).delete()[0] + Location.objects.filter( + id__in=location_ids, + products__isnull=True, + findings__isnull=True, + ).delete() + return removed + + +def locations_shared_outside(locations, products): + """Locations in ``locations`` that something outside ``products`` also references.""" + foreign_products = LocationProductReference.objects.filter( + location=OuterRef("pk"), + ).exclude(product__in=products) + foreign_findings = LocationFindingReference.objects.filter( + location=OuterRef("pk"), + ).exclude(finding__test__engagement__product__in=products) + return locations.filter(Exists(foreign_products) | Exists(foreign_findings)) + + def annotate_location_counts_and_status(locations, user=None): # Annotate the queryset with counts of findings # This aggregates the total and active findings by joining LocationFindingReference. diff --git a/dojo/url/ui/views.py b/dojo/url/ui/views.py index 70a27eb449..164ad9b4a7 100644 --- a/dojo/url/ui/views.py +++ b/dojo/url/ui/views.py @@ -23,7 +23,12 @@ ImportEndpointMetaForm, ) from dojo.location.models import Location, LocationFindingReference, LocationProductReference -from dojo.location.queries import annotate_location_counts_and_status, get_authorized_locations +from dojo.location.queries import ( + annotate_location_counts_and_status, + get_authorized_locations, + locations_shared_outside, + remove_location_references, +) from dojo.location.status import FindingLocationStatus, ProductLocationStatus from dojo.models import DojoMeta, Finding, Product from dojo.product.queries import get_authorized_products @@ -325,6 +330,15 @@ def edit_endpoint(request, location_id): # Handle form submission for editing an endpoint form = URLForm(request.POST, instance=location.url) if form.is_valid(): + editable_products = get_authorized_products(Permissions.Location_Edit, request.user) + if locations_shared_outside(Location.objects.filter(id=location.id), editable_products).exists(): + messages.add_message( + request, + messages.ERROR, + "This endpoint is also recorded by another product, so it cannot be renamed here.", + extra_tags="alert-danger", + ) + return HttpResponseRedirect(reverse("view_endpoint", args=(location.id,))) try: form.save(update_only=True) except ValidationError: @@ -401,8 +415,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() + remove_location_references( + 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 +544,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: + reference_products = Product.objects.filter(id=product_id) + else: + reference_products = get_authorized_products(Permissions.Location_Delete, request.user) + remove_location_references(locations, reference_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_reference_scoped_writes.py b/unittests/test_location_reference_scoped_writes.py new file mode 100644 index 0000000000..acbc06d39b --- /dev/null +++ b/unittests/test_location_reference_scoped_writes.py @@ -0,0 +1,153 @@ +""" +Regression tests for reference-scoped writes on a shared Location. + +A Location row is deduplicated globally, so any product that records the same URL +references the same row. Authorization for the row therefore follows from a reference +the caller can create for themselves. A rename or a delete must act on the caller's own +reference, never on the shared row, or one product silently rewrites or destroys +another product's endpoint. +""" +from datetime import UTC, date, datetime + +from django.urls import reverse + +from dojo.authorization.models import Product_Member, Role +from dojo.location.models import Location, LocationFindingReference, LocationProductReference +from dojo.location.queries import remove_location_references +from dojo.models import Dojo_User, Engagement, Finding, Product, Product_Type, Test, Test_Type +from dojo.url.models import URL + +from .dojo_test_case import DojoTestCase, skip_unless_v3 +from .test_permissions_audit import LegacyAuthMirrorMixin + +PASSWORD = "testTEST1234!@#$" +SHARED_HOST = "shared-writes.example.test" +OWN_HOST = "own-writes.example.test" + + +@skip_unless_v3 +class TestLocationReferenceScopedWrites(LegacyAuthMirrorMixin, DojoTestCase): + + @classmethod + def setUpTestData(cls): + cls.pt_mine = Product_Type.objects.create(name="Ref Scoped Mine PT") + cls.pt_outside = Product_Type.objects.create(name="Ref Scoped Outside PT") + cls.product_mine = Product.objects.create( + name="Ref Scoped Mine Product", description="mine", prod_type=cls.pt_mine, + ) + cls.product_outside = Product.objects.create( + name="Ref Scoped Outside Product", description="out", prod_type=cls.pt_outside, + ) + cls.user = Dojo_User.objects.create_user( + username="ref_scoped_user", password=PASSWORD, is_active=True, + ) + # Membership on the caller's product only. + Product_Member.objects.create( + product=cls.product_mine, user=cls.user, role=Role.objects.get(name="Owner"), + ) + + # One deduplicated Location that both products record. + cls.shared = URL.get_or_create_from_values( + protocol="https", host=SHARED_HOST, path="admin", + ).location + cls.shared.associate_with_product(cls.product_mine) + cls.shared.associate_with_product(cls.product_outside) + + # A Location only the caller's product records. + cls.own = URL.get_or_create_from_values( + protocol="https", host=OWN_HOST, path="admin", + ).location + cls.own.associate_with_product(cls.product_mine) + + # A finding on the other product, referencing the shared row. + test_type, _ = Test_Type.objects.get_or_create(name="Ref Scoped Scan") + engagement = Engagement.objects.create( + product=cls.product_outside, name="ref-scoped-eng", + target_start=date(2026, 1, 1), target_end=date(2026, 1, 2), + ) + outside_test = Test.objects.create( + engagement=engagement, test_type=test_type, + target_start=datetime(2026, 1, 1, tzinfo=UTC), + target_end=datetime(2026, 1, 2, tzinfo=UTC), + ) + cls.outside_finding = Finding.objects.create( + test=outside_test, title="Ref Scoped Outside Finding", severity="High", + numerical_severity="S1", description="outside", active=True, verified=False, + reporter=cls.user, + ) + cls.shared.associate_with_finding(cls.outside_finding) + + def setUp(self): + super().setUp() + self.client.force_login(self.user) + + def _refs(self, location): + return set( + LocationProductReference.objects.filter(location=location) + .values_list("product__name", flat=True), + ) + + def _rename(self, location, host): + return self.client.post( + reverse("edit_endpoint", args=(location.id,)), + {"protocol": "https", "host": host, "path": "admin"}, + ) + + def _delete(self, location): + return self.client.post( + reverse("delete_endpoint", args=(location.id,)), {"id": location.id}, + ) + + def test_rename_leaves_a_location_another_product_records_untouched(self): + self._rename(self.shared, "renamed-by-attacker.example.test") + self.shared.refresh_from_db() + self.shared.url.refresh_from_db() + self.assertEqual(self.shared.url.host, SHARED_HOST) + self.assertEqual(self.shared.location_value, f"https://{SHARED_HOST}/admin") + + def test_rename_still_works_when_no_other_product_records_it(self): + self._rename(self.own, "renamed-by-owner.example.test") + self.own.url.refresh_from_db() + self.assertEqual(self.own.url.host, "renamed-by-owner.example.test") + + def test_helper_removes_only_the_given_products_references(self): + remove_location_references( + Location.objects.filter(id=self.shared.id), + Product.objects.filter(id=self.product_mine.id), + ) + self.assertTrue(Location.objects.filter(id=self.shared.id).exists()) + self.assertEqual(self._refs(self.shared), {self.product_outside.name}) + self.assertTrue( + LocationFindingReference.objects.filter( + location_id=self.shared.id, finding=self.outside_finding, + ).exists(), + ) + + def test_helper_removes_the_row_when_nothing_else_references_it(self): + remove_location_references( + Location.objects.filter(id=self.own.id), + Product.objects.filter(id=self.product_mine.id), + ) + self.assertFalse(Location.objects.filter(id=self.own.id).exists()) + + def test_single_delete_route_stays_denied_without_the_delete_action(self): + # The object path treats Delete as staff only, so this route was already closed + # to an ordinary member. The bulk route is the one that was reachable. + response = self._delete(self.shared) + 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): + 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}) + + def test_bulk_delete_removes_the_row_when_nothing_else_references_it(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())