diff --git a/dojo/location/models.py b/dojo/location/models.py index c20b6eaa34..2805f8530d 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 70a27eb449..ee9e4787c7 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 0eaddd18b5..bf5bc7bc62 100644 --- a/unittests/test_location_cross_product_authz.py +++ b/unittests/test_location_cross_product_authz.py @@ -1,14 +1,24 @@ 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, + delete_locations_for_products, +) 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 +99,124 @@ 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, + reporter=cls.alice, + ) + + 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(), + ) + + # 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() + delete_locations_for_products( + Location.objects.filter(id=self.shared.id), + Product.objects.filter(id=self.product_a.id), + ) + 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_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.assertFalse(Location.objects.filter(pk=self.own.id).exists())