Scope endpoint delete and rename to the requesting product

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.

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(

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())