From 8c77d9a7a535bc0c9859a29d39067a3e9d215cf7 Mon Sep 17 00:00:00 2001 From: Sam Vader Date: Tue, 21 Jul 2026 02:26:52 -0500 Subject: [PATCH] Scope location and endpoint reference writes to authorized products Hardening to object-level authorization on several API and UI write paths that associate locations and endpoints with findings and products. These paths now restrict the selectable references to the objects the requesting user is authorized for, matching the scoping the UI forms and dedicated viewsets already apply. Adds regression tests. No functional change for correctly-permissioned users. --- dojo/api_v2/serializers.py | 10 +- dojo/finding/api/serializer.py | 14 +- dojo/url/ui/views.py | 19 ++- unittests/test_v3_location_reference_authz.py | 147 ++++++++++++++++++ 4 files changed, 186 insertions(+), 4 deletions(-) create mode 100644 unittests/test_v3_location_reference_authz.py diff --git a/dojo/api_v2/serializers.py b/dojo/api_v2/serializers.py index ff60bbf22a6..5565f9986ac 100644 --- a/dojo/api_v2/serializers.py +++ b/dojo/api_v2/serializers.py @@ -18,11 +18,13 @@ from rest_framework.exceptions import NotFound from rest_framework.exceptions import ValidationError as RestFrameworkValidationError +from dojo.endpoint.queries import get_authorized_endpoints from dojo.importers.auto_create_context import AutoCreateContextManager from dojo.importers.base_importer import BaseImporter from dojo.importers.default_importer import DefaultImporter from dojo.importers.default_reimporter import DefaultReImporter from dojo.location.models import Location +from dojo.location.queries import get_authorized_locations from dojo.models import ( DEDUPLICATION_EXECUTION_MODE_CHOICES, IMPORT_ACTIONS, @@ -514,14 +516,20 @@ class CommonImportScanSerializer(serializers.Serializer): # TODO: Delete this after the move to Locations def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) + # Scope endpoint_to_add to the locations/endpoints the requesting user is authorized for. + user = getattr(self.context.get("request"), "user", None) if not settings.V3_FEATURE_LOCATIONS: # TODO: why do we allow only existing endpoints? self.fields["endpoint_to_add"] = serializers.PrimaryKeyRelatedField( - queryset=Endpoint.objects.all(), + queryset=get_authorized_endpoints("view", user=user) if user else Endpoint.objects.none(), required=False, default=None, help_text="Enter the ID of an Endpoint that is associated with the target Product. New Findings will be added to that Endpoint.", ) + else: + self.fields["endpoint_to_add"].queryset = ( + get_authorized_locations("view", user=user) if user else Location.objects.none() + ) def get_importer( self, diff --git a/dojo/finding/api/serializer.py b/dojo/finding/api/serializer.py index 360f08b7926..b03f1e60930 100644 --- a/dojo/finding/api/serializer.py +++ b/dojo/finding/api/serializer.py @@ -15,6 +15,7 @@ import dojo.finding.helper as finding_helper from dojo.authorization.authorization import user_has_permission from dojo.celery_dispatch import dojo_dispatch_task +from dojo.endpoint.queries import get_authorized_endpoints from dojo.finding.helper import ( save_endpoints_template, save_vulnerability_ids, @@ -24,6 +25,7 @@ from dojo.jira import services as jira_services from dojo.jira.api.serializers import JIRAIssueSerializer from dojo.location.models import LocationFindingReference +from dojo.location.queries import get_authorized_location_finding_reference from dojo.models import ( SEVERITIES, Development_Environment, @@ -349,9 +351,19 @@ def get_fields(self): # TODO: Delete this after the move to Locations def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) + # Scope the endpoints field to the references the requesting user is authorized + # for, mirroring the scoping the finding UI form already applies. + user = getattr(self.context.get("request"), "user", None) if not settings.V3_FEATURE_LOCATIONS: self.fields["endpoints"] = serializers.PrimaryKeyRelatedField( - many=True, required=False, queryset=Endpoint.objects.all(), + many=True, required=False, + queryset=get_authorized_endpoints("view", user=user) if user else Endpoint.objects.none(), + ) + else: + self.fields["endpoints"] = serializers.PrimaryKeyRelatedField( + source="locations", many=True, required=False, + queryset=get_authorized_location_finding_reference("view", user=user) + if user else LocationFindingReference.objects.none(), ) def get_accepted_risks(self, obj): diff --git a/dojo/url/ui/views.py b/dojo/url/ui/views.py index 826c9329262..6c5f416a582 100644 --- a/dojo/url/ui/views.py +++ b/dojo/url/ui/views.py @@ -14,6 +14,7 @@ from django.utils import timezone from dojo.authorization.authorization import user_has_permission_or_403 +from dojo.authorization.roles_permissions import Permissions from dojo.endpoint.utils import endpoint_meta_import from dojo.forms import ( DeleteEndpointForm, @@ -24,6 +25,7 @@ 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 +from dojo.product.queries import get_authorized_products from dojo.reports.ui.views import generate_report from dojo.url.filters import URLFilter from dojo.url.models import URL @@ -536,13 +538,26 @@ def endpoint_bulk_update_all(request, product_id=None): f"Skipped mitigation of {skipped_location_count} locations because you are not authorized.", ) + # Scope the reference updates to the acting product (or, on the all-products + # route, the products the user may edit); get_authorized_locations above scopes + # only the Location rows, not their references. + if product_id is not None: + reference_products = Product.objects.filter(id=product_id) + else: + reference_products = get_authorized_products(Permissions.Product_Edit, request.user) # Bulk update the status of related FindingLocationStatus and ProductLocationStatus objects to 'Mitigated' - finding_update_counts = LocationFindingReference.objects.filter(location__in=locations).update( + finding_update_counts = LocationFindingReference.objects.filter( + location__in=locations, + finding__test__engagement__product__in=reference_products, + ).update( status=FindingLocationStatus.Mitigated, auditor=request.user, audit_time=timezone.now(), ) - product_update_counts = LocationProductReference.objects.filter(location__in=locations).update( + product_update_counts = LocationProductReference.objects.filter( + location__in=locations, + product__in=reference_products, + ).update( status=ProductLocationStatus.Mitigated, ) # Total number of updated statuses for reporting diff --git a/unittests/test_v3_location_reference_authz.py b/unittests/test_v3_location_reference_authz.py new file mode 100644 index 00000000000..fc9dc7d6a37 --- /dev/null +++ b/unittests/test_v3_location_reference_authz.py @@ -0,0 +1,147 @@ +""" +Regression tests: the V3 location/endpoint reference write paths limit the +references a request may select to the objects the requesting user is authorized +for. Each test asserts an authorized reference is accepted and an unauthorized +one is rejected, for the Finding endpoints field, the import endpoint_to_add +field, and bulk-mitigate reference updates. +""" +from django.core.files.uploadedfile import SimpleUploadedFile +from django.urls import reverse +from django.utils.timezone import now +from rest_framework.exceptions import ValidationError as DRFValidationError +from rest_framework.test import APIRequestFactory + +from dojo.api_v2.serializers import ImportScanSerializer +from dojo.authorization.roles_permissions import Roles +from dojo.location.models import ( + LocationFindingReference, + LocationProductReference, +) +from dojo.location.status import FindingLocationStatus +from dojo.models import ( + Dojo_User, + Engagement, + Finding, + Product, + Product_Member, + Product_Type, + Role, + Test, + Test_Type, +) +from dojo.url.models import URL +from unittests.dojo_test_case import DojoAPITestCase, skip_unless_v3 + +GENERIC_FINDINGS = b'{"findings": [{"title": "probe", "severity": "Info", "description": "probe"}]}' + + +@skip_unless_v3 +class V3LocationReferenceAuthzTests(DojoAPITestCase): + + """Two products, two users; each user is authorized only for their own product.""" + + @classmethod + def setUpTestData(cls): + cls.pt = Product_Type.objects.create(name="v3_locref_pt") + cls.product_a = Product.objects.create(name="v3_locref_a", description="a", prod_type=cls.pt) + cls.product_b = Product.objects.create(name="v3_locref_b", description="b", prod_type=cls.pt) + + cls.user_a = Dojo_User.objects.create(username="v3_locref_user_a", is_active=True) + cls.user_b = Dojo_User.objects.create(username="v3_locref_user_b", is_active=True) + # Grant membership two ways so the test is authorization-backend agnostic: + # legacy authorized_users (honored by the OS backend) and a Product_Member + # Owner role (honored by the Pro backend). + owner = Role.objects.get(id=Roles.Owner) + for product, user in ((cls.product_a, cls.user_a), (cls.product_b, cls.user_b)): + product.authorized_users.add(user) + Product_Member.objects.create(product=product, user=user, role=owner) + + cls.test_type, _ = Test_Type.objects.get_or_create(name="v3_locref_scan") + + cls.finding_a = cls._make_finding(cls.product_a, cls.user_a, "finding_a") + cls.engagement_a = cls.finding_a.test.engagement + cls.finding_b = cls._make_finding(cls.product_b, cls.user_b, "finding_b") + + # A location on product_b (authorized only for user_b). + url_b = URL.get_or_create_from_object(URL.from_value("https://b.example.test/x")) + cls.lfr_b = url_b.location.associate_with_finding(cls.finding_b) + cls.location_b = url_b.location + + # A location on product_a (authorized for user_a) for the accepted-case controls. + cls.finding_a_extra = cls._make_finding(cls.product_a, cls.user_a, "finding_a_extra") + url_a = URL.get_or_create_from_object(URL.from_value("https://a.example.test/ok")) + cls.lfr_a = url_a.location.associate_with_finding(cls.finding_a_extra) + cls.location_a = url_a.location + + # A location associated with both products, with an active reference from product_b. + url_shared = URL.get_or_create_from_object(URL.from_value("https://shared.example.test/s")) + cls.location_shared = url_shared.location + cls.location_shared.associate_with_product(cls.product_a) + cls.lfr_shared_b = LocationFindingReference.objects.create( + location=cls.location_shared, finding=cls.finding_b, status=FindingLocationStatus.Active, + ) + + @classmethod + def _make_finding(cls, product, user, title): + engagement = Engagement.objects.create(name=title + "_eng", product=product, target_start=now(), target_end=now()) + test = Test.objects.create(engagement=engagement, test_type=cls.test_type, target_start=now(), target_end=now()) + return Finding.objects.create( + test=test, title=title, description="x", severity="High", + numerical_severity="S0", active=True, verified=True, reporter=user, + ) + + # Finding endpoints field + def test_finding_endpoints_rejects_unauthorized_reference(self): + self.client.force_authenticate(user=self.user_a) + response = self.client.patch( + reverse("finding-list") + f"{self.finding_a.id}/", + {"endpoints": [self.lfr_b.id], "push_to_jira": False}, format="json", secure=True, + ) + self.assertEqual(response.status_code, 400) + self.assertFalse( + LocationProductReference.objects.filter(location=self.location_b, product=self.product_a).exists(), + ) + + def test_finding_endpoints_allows_authorized_reference(self): + self.client.force_authenticate(user=self.user_a) + response = self.client.patch( + reverse("finding-list") + f"{self.finding_a.id}/", + {"endpoints": [self.lfr_a.id], "push_to_jira": False}, format="json", secure=True, + ) + self.assertEqual(response.status_code, 200) + + # import endpoint_to_add + def test_import_endpoint_to_add_rejects_unauthorized_reference(self): + self.client.force_authenticate(user=self.user_a) + response = self.client.post(reverse("importscan-list"), { + "engagement": self.engagement_a.id, + "scan_type": "Generic Findings Import", + "endpoint_to_add": self.location_b.id, + "file": SimpleUploadedFile("s.json", GENERIC_FINDINGS, content_type="application/json"), + }, secure=True) + self.assertEqual(response.status_code, 400) + self.assertFalse( + LocationProductReference.objects.filter(location=self.location_b, product=self.product_a).exists(), + ) + + def test_import_endpoint_to_add_allows_authorized_reference(self): + # Checked at the field level so the assertion is about the queryset scoping, + # not the downstream import pipeline. + request = APIRequestFactory().post(reverse("importscan-list")) + request.user = self.user_a + field = ImportScanSerializer(context={"request": request}).fields["endpoint_to_add"] + self.assertEqual(field.to_internal_value(self.location_a.id), self.location_a) + with self.assertRaises(DRFValidationError): + field.to_internal_value(self.location_b.id) + + # bulk mitigate reference updates + def test_bulk_mitigate_scopes_reference_updates(self): + self.client.force_login(self.user_a) + response = self.client.post( + reverse("endpoints_bulk_update_all_product", args=(self.product_a.id,)), + data={"endpoints_to_update": self.location_shared.id}, secure=True, + ) + self.assertIn(response.status_code, (200, 302)) + self.lfr_shared_b.refresh_from_db() + # A reference outside the acting product is left unchanged. + self.assertEqual(self.lfr_shared_b.status, FindingLocationStatus.Active)