feat(api): add bulk finding update endpoint

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

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": <int>, ...}, ...]}`` 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},

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",
+        )