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)