diff --git a/dojo/finding/api/serializer.py b/dojo/finding/api/serializer.py index f29c023c129..bc5e5b33535 100644 --- a/dojo/finding/api/serializer.py +++ b/dojo/finding/api/serializer.py @@ -744,6 +744,92 @@ def validate_severity(self, value: str) -> str: return value +# Fields that may be changed through the bulk-update endpoint. Kept to an +# explicit allowlist so the endpoint stays a narrow enrichment tool (EPSS / KEV +# threat-intelligence metadata) and can never be used to mass-edit sensitive +# finding attributes such as severity, status or ownership. +BULK_UPDATE_ALLOWED_FIELDS = ( + "epss_score", + "epss_percentile", + "known_exploited", + "ransomware_used", + "kev_date", +) + +# Upper bound on the number of findings a single bulk request may touch. The +# whole batch is applied in one transaction, so this caps the amount of work +# (and the database lock footprint) of a single request. +BULK_UPDATE_MAX_FINDINGS = 200 + + +class FindingBulkUpdateFieldsSerializer(serializers.ModelSerializer): + + """ + A single item in a bulk finding update: the target finding ``id`` plus any + subset of the allowlisted fields. + + Unknown fields are rejected with a 400 so that typos, or attempts to update + non-allowlisted attributes, fail loudly instead of being silently ignored + (which is DRF's default behaviour). + """ + + id = serializers.PrimaryKeyRelatedField(queryset=Finding.objects.all()) + + class Meta: + model = Finding + fields = ("id", *BULK_UPDATE_ALLOWED_FIELDS) + # Every allowlisted field is optional; each item only sends what it changes. + extra_kwargs = {field: {"required": False} for field in BULK_UPDATE_ALLOWED_FIELDS} + + def to_internal_value(self, data): + # Reject unknown keys before DRF drops them, so callers learn about + # mistakes instead of having part of their payload silently ignored. + if isinstance(data, dict): + unknown_fields = set(data) - set(self.fields) + if unknown_fields: + raise serializers.ValidationError({ + field: ["This field cannot be updated through the bulk endpoint."] + for field in sorted(unknown_fields) + }) + return super().to_internal_value(data) + + def validate(self, data): + # Reject no-op items so callers get clear feedback instead of a write + # that silently changes nothing. + if not any(field in data for field in BULK_UPDATE_ALLOWED_FIELDS): + msg = f"Provide at least one field to update: {', '.join(BULK_UPDATE_ALLOWED_FIELDS)}." + raise serializers.ValidationError(msg) + return data + + +class FindingBulkUpdateSerializer(serializers.Serializer): + + """ + Request body for ``PATCH /api/v2/findings/bulk/``. + + Shape: ``{"findings": [{"id": 123, "epss_score": 0.42}, ...]}``. + """ + + findings = FindingBulkUpdateFieldsSerializer(many=True, allow_empty=False) + + def validate_findings(self, value): + if len(value) > BULK_UPDATE_MAX_FINDINGS: + msg = ( + f"A bulk update is limited to {BULK_UPDATE_MAX_FINDINGS} findings " + f"per request, but {len(value)} were provided." + ) + raise serializers.ValidationError(msg) + # A finding listed twice would receive two conflicting updates in the + # same transaction; reject duplicates up front instead of silently + # letting the last item win. + ids = [item["id"].pk for item in value] + duplicates = sorted({finding_id for finding_id in ids if ids.count(finding_id) > 1}) + if duplicates: + msg = f"Each finding may only appear once per bulk update. Duplicate ids: {duplicates}." + raise serializers.ValidationError(msg) + return value + + class FindingTemplateSerializer(serializers.ModelSerializer): vulnerability_ids = serializers.SerializerMethodField() endpoints = serializers.SerializerMethodField() diff --git a/dojo/finding/api/views.py b/dojo/finding/api/views.py index 2655b7ddc35..06eb7b04c16 100644 --- a/dojo/finding/api/views.py +++ b/dojo/finding/api/views.py @@ -4,7 +4,7 @@ import tagulous from django.conf import settings from django.core.exceptions import ValidationError -from django.db import IntegrityError +from django.db import IntegrityError, transaction from django.shortcuts import get_object_or_404 from django.urls import reverse from django.utils import timezone @@ -43,6 +43,7 @@ from dojo.finding.api.serializer import ( BurpRawRequestResponseMultiSerializer, BurpRawRequestResponseSerializer, + FindingBulkUpdateSerializer, FindingCloseSerializer, FindingCreateSerializer, FindingMetaSerializer, @@ -675,6 +676,84 @@ def set_finding_as_original(self, request, pk, new_fid): return Response(status=status.HTTP_400_BAD_REQUEST) return Response(status=status.HTTP_204_NO_CONTENT) + @extend_schema( + methods=["PATCH"], + request=FindingBulkUpdateSerializer, + responses={ + status.HTTP_200_OK: FindingSerializer(many=True), + status.HTTP_400_BAD_REQUEST: OpenApiResponse( + description="Validation failed: unknown field, invalid value, unknown finding id, " + "duplicate id, or more findings than the per-request limit.", + ), + status.HTTP_403_FORBIDDEN: OpenApiResponse( + description="The user lacks edit permission on at least one referenced finding; " + "the entire batch is rejected and rolled back.", + ), + }, + ) + @action( + detail=False, + methods=["patch"], + url_path="bulk", + filter_backends=[], + pagination_class=None, + ) + def bulk_update(self, request): + """ + Update an allowlisted set of fields on many findings in a single atomic request. + + The request body is ``{"findings": [{"id": , ...}, ...]}`` where each item + carries the target finding id plus any subset of the allowlisted fields + (epss_score, epss_percentile, known_exploited, ransomware_used, kev_date). The + user must have edit permission on every referenced finding; if any check fails, + the entire batch is rolled back and a 403 is returned. Findings are never pushed + to JIRA from this endpoint. + """ + serializer = FindingBulkUpdateSerializer( + data=request.data, context={"request": request}, + ) + serializer.is_valid(raise_exception=True) + updated_findings = self._perform_bulk_update( + request, serializer.validated_data["findings"], + ) + # Re-fetch through the viewset queryset so the response reuses the same + # prefetching and authorization scoping as a normal finding list. + response_findings = self.get_queryset().filter( + id__in=[finding.id for finding in updated_findings], + ) + response_serializer = FindingSerializer( + response_findings, many=True, context={"request": request}, + ) + return Response(response_serializer.data, status=status.HTTP_200_OK) + + @transaction.atomic + def _perform_bulk_update(self, request, items): + updated_findings = [] + for item in items: + finding = item["id"] + # Per-item authorization mirrors a normal PATCH: require edit permission + # on each finding. check_object_permissions raises a 403 on the first + # failure and, because this whole method runs in one transaction, that + # 403 rolls back any updates already applied earlier in the batch. + self.check_object_permissions(request, finding) + for field, value in item.items(): + if field == "id": + continue + setattr(finding, field, value) + # The allowlisted fields never affect the dedupe hash or JIRA sync, so + # the expensive post-save processing is skipped for performance. The row + # is still UPDATEd, which fires the pghistory trigger, so audit history + # is recorded exactly as it is for a normal PATCH. + finding.save( + dedupe_option=False, + rules_option=False, + product_grading_option=False, + issue_updater_option=False, + push_to_jira=False, + ) + updated_findings.append(finding) + return updated_findings + @extend_schema( request=api_v2_serializers.ReportGenerateOptionSerializer, responses={status.HTTP_200_OK: api_v2_serializers.ReportGenerateSerializer}, diff --git a/unittests/test_finding_bulk_update_api.py b/unittests/test_finding_bulk_update_api.py new file mode 100644 index 00000000000..4e85e0fa5dd --- /dev/null +++ b/unittests/test_finding_bulk_update_api.py @@ -0,0 +1,193 @@ +""" +API tests for the bulk finding-update endpoint: ``PATCH /api/v2/findings/bulk/``. + +The endpoint updates an allowlisted set of fields (EPSS / KEV threat-intelligence +metadata) on many findings in a single atomic transaction, with a per-item edit +permission check. These tests cover the happy path, input validation, the +per-request limit, atomic rollback on a permission failure, and that audit +history is recorded exactly as for a normal PATCH. +""" +import datetime + +from django.apps import apps +from rest_framework import status +from rest_framework.authtoken.models import Token +from rest_framework.test import APIClient + +from dojo.finding.api.serializer import BULK_UPDATE_MAX_FINDINGS +from dojo.models import ( + Dojo_User, + Engagement, + Finding, + Product, + Product_Type, + Test, + Test_Type, +) +from unittests.dojo_test_case import DojoAPITestCase + +BULK_URL = "/api/v2/findings/bulk/" + + +class TestFindingBulkUpdateApi(DojoAPITestCase): + + @classmethod + def setUpTestData(cls): + start = datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC) + end = datetime.datetime(2020, 2, 1, tzinfo=datetime.UTC) + + # A non-staff user authorized on product_a only. Staff/superusers bypass + # all object permissions, so a non-staff, scoped user is required to + # exercise the per-item authorization path. + cls.user = Dojo_User.objects.create(username="bulk_tester", is_staff=False) + cls.token = Token.objects.create(user=cls.user) + + cls.test_type = Test_Type.objects.create(name="Bulk Update Mock Scan", static_tool=True) + + # Product the user is authorized to edit. + cls.product_type_a = Product_Type.objects.create(name="Owned") + cls.product_a = Product.objects.create(prod_type=cls.product_type_a, name="Owned Product", description="Owned") + cls.product_a.authorized_users.add(cls.user) + cls.engagement_a = Engagement.objects.create(product=cls.product_a, target_start=start, target_end=end) + cls.test_a = Test.objects.create(engagement=cls.engagement_a, test_type=cls.test_type, target_start=start, target_end=end) + + # Product the user has no membership in (and therefore cannot edit). + cls.product_type_b = Product_Type.objects.create(name="Foreign") + cls.product_b = Product.objects.create(prod_type=cls.product_type_b, name="Foreign Product", description="Foreign") + cls.engagement_b = Engagement.objects.create(product=cls.product_b, target_start=start, target_end=end) + cls.test_b = Test.objects.create(engagement=cls.engagement_b, test_type=cls.test_type, target_start=start, target_end=end) + + cls.finding_1 = cls._create_finding(cls.test_a, "Finding One") + cls.finding_2 = cls._create_finding(cls.test_a, "Finding Two") + cls.foreign_finding = cls._create_finding(cls.test_b, "Foreign Finding") + + @classmethod + def _create_finding(cls, test, title): + return Finding.objects.create( + test=test, + title=title, + severity="High", + numerical_severity="S1", + verified=False, + active=True, + description="desc", + reporter=cls.user, + ) + + def setUp(self): + self.client = APIClient() + self.client.credentials(HTTP_AUTHORIZATION="Token " + self.token.key) + + def _patch(self, findings): + return self.client.patch(BULK_URL, data={"findings": findings}, format="json") + + # --- happy path ------------------------------------------------------- + + def test_bulk_update_success(self): + response = self._patch([ + {"id": self.finding_1.id, "epss_score": 0.42, "known_exploited": True}, + {"id": self.finding_2.id, "epss_percentile": 0.9, "kev_date": "2024-01-15"}, + ]) + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content[:500]) + + self.finding_1.refresh_from_db() + self.finding_2.refresh_from_db() + self.assertEqual(self.finding_1.epss_score, 0.42) + self.assertTrue(self.finding_1.known_exploited) + self.assertEqual(self.finding_2.epss_percentile, 0.9) + self.assertEqual(self.finding_2.kev_date, datetime.date(2024, 1, 15)) + + def test_bulk_update_response_lists_updated_findings(self): + response = self._patch([ + {"id": self.finding_1.id, "epss_score": 0.1}, + {"id": self.finding_2.id, "epss_score": 0.2}, + ]) + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content[:500]) + returned_ids = {finding["id"] for finding in response.json()} + self.assertEqual(returned_ids, {self.finding_1.id, self.finding_2.id}) + + def test_bulk_update_leaves_unlisted_fields_untouched(self): + original_severity = self.finding_1.severity + response = self._patch([{"id": self.finding_1.id, "epss_score": 0.33}]) + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content[:500]) + self.finding_1.refresh_from_db() + self.assertEqual(self.finding_1.severity, original_severity) + + # --- validation ------------------------------------------------------- + + def test_unknown_field_rejected(self): + response = self._patch([{"id": self.finding_1.id, "severity": "Critical"}]) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content[:500]) + self.finding_1.refresh_from_db() + self.assertEqual(self.finding_1.severity, "High") + + def test_unknown_finding_id_rejected(self): + response = self._patch([{"id": 999999, "epss_score": 0.1}]) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content[:500]) + + def test_duplicate_id_rejected(self): + response = self._patch([ + {"id": self.finding_1.id, "epss_score": 0.1}, + {"id": self.finding_1.id, "epss_score": 0.2}, + ]) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content[:500]) + + def test_item_without_updatable_fields_rejected(self): + response = self._patch([{"id": self.finding_1.id}]) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content[:500]) + + def test_invalid_epss_score_rejected(self): + response = self._patch([{"id": self.finding_1.id, "epss_score": 5}]) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content[:500]) + self.finding_1.refresh_from_db() + self.assertIsNone(self.finding_1.epss_score) + + def test_empty_findings_list_rejected(self): + response = self._patch([]) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content[:500]) + + def test_over_limit_rejected(self): + payload = [{"id": self.finding_1.id, "epss_score": 0.1} for _ in range(BULK_UPDATE_MAX_FINDINGS + 1)] + response = self._patch(payload) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content[:500]) + + # --- authorization ---------------------------------------------------- + + def test_forbidden_finding_fails_entire_batch(self): + # finding_1 is editable, foreign_finding is not: the whole batch must be + # rejected and finding_1 must remain unchanged (atomic rollback). + response = self._patch([ + {"id": self.finding_1.id, "epss_score": 0.5}, + {"id": self.foreign_finding.id, "epss_score": 0.5}, + ]) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN, response.content[:500]) + self.finding_1.refresh_from_db() + self.foreign_finding.refresh_from_db() + self.assertIsNone(self.finding_1.epss_score) + self.assertIsNone(self.foreign_finding.epss_score) + + def test_requires_authentication(self): + anonymous = APIClient() + response = anonymous.patch(BULK_URL, data={"findings": [{"id": self.finding_1.id, "epss_score": 0.1}]}, format="json") + self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN), response.content[:500]) + + # --- audit history ---------------------------------------------------- + + def test_bulk_update_records_audit_history(self): + # A bulk update is a normal row UPDATE, so it fires the pghistory trigger + # and produces an audit event exactly as a normal PATCH would. pghistory + # triggers are enabled by default (ENABLE_AUDITLOG defaults to True), so + # this only verifies the event is recorded. + finding_event_model = apps.get_model("dojo", "FindingEvent") + events_before = finding_event_model.objects.filter(pgh_obj_id=self.finding_1.id).count() + + response = self._patch([{"id": self.finding_1.id, "epss_score": 0.77}]) + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content[:500]) + + events_after = finding_event_model.objects.filter(pgh_obj_id=self.finding_1.id).count() + self.assertGreater(events_after, events_before, "Expected a pghistory event after the bulk update") + # The audit trail must capture the new value, exactly as for a normal PATCH. + self.assertTrue( + finding_event_model.objects.filter(pgh_obj_id=self.finding_1.id, epss_score=0.77).exists(), + "Expected an audit event recording the updated epss_score", + )