close finding: sync api and ui behaviour (#13230)

* close finding: sync api and ui behaviour

* deduplicate code into common method

* use Finding_Edit permission

* restore schema

* fix test

Author: valentijnscholten

dojo/api_v2/serializers.py

@@ -22,6 +22,7 @@
 from rest_framework.exceptions import ValidationError as RestFrameworkValidationError
 from rest_framework.fields import DictField, MultipleChoiceField
 
+import dojo.finding.helper as finding_helper
 import dojo.jira_link.helper as jira_helper
 import dojo.risk_acceptance.helper as ra_helper
 from dojo.authorization.authorization import user_has_permission
@@ -122,6 +123,7 @@
     requires_file,
     requires_tool_type,
 )
+from dojo.user.queries import get_authorized_users
 from dojo.user.utils import get_configuration_permissions_codenames
 from dojo.utils import is_scan_file_too_large
 from dojo.validators import ImporterFileExtensionValidator, tag_validator
@@ -2697,6 +2699,9 @@ class FindingCloseSerializer(serializers.ModelSerializer):
     false_p = serializers.BooleanField(required=False)
     out_of_scope = serializers.BooleanField(required=False)
     duplicate = serializers.BooleanField(required=False)
+    mitigated_by = serializers.PrimaryKeyRelatedField(required=False, allow_null=True, queryset=Dojo_User.objects.all())
+    note = serializers.CharField(required=False, allow_blank=True)
+    note_type = serializers.PrimaryKeyRelatedField(required=False, allow_null=True, queryset=Note_Type.objects.all())
 
     class Meta:
         model = Finding
@@ -2706,8 +2711,34 @@ class Meta:
             "false_p",
             "out_of_scope",
             "duplicate",
+            "mitigated_by",
+            "note",
+            "note_type",
         )
 
+    def validate(self, data):
+        request = self.context.get("request")
+        request_user = getattr(request, "user", None)
+
+        mitigated_by_user = data.get("mitigated_by")
+        if mitigated_by_user is not None:
+            # Require permission to edit mitigated metadata
+            if not (request_user and finding_helper.can_edit_mitigated_data(request_user)):
+                raise serializers.ValidationError({
+                    "mitigated_by": ["Not allowed to set mitigated_by."],
+                })
+
+            # Ensure selected user is authorized (Finding_Edit)
+            authorized_users = get_authorized_users(Permissions.Finding_Edit, user=request_user)
+            if not authorized_users.filter(id=mitigated_by_user.id).exists():
+                raise serializers.ValidationError({
+                    "mitigated_by": [
+                        "Selected user is not authorized to be set as mitigated_by.",
+                    ],
+                })
+
+        return data
+
 
 class ReportGenerateOptionSerializer(serializers.Serializer):
     include_finding_notes = serializers.BooleanField(default=False)

dojo/api_v2/views.py

@@ -31,6 +31,7 @@
 from rest_framework.permissions import DjangoModelPermissions, IsAuthenticated
 from rest_framework.response import Response
 
+import dojo.finding.helper as finding_helper
 import dojo.jira_link.helper as jira_helper
 from dojo.api_v2 import (
     mixins as dojo_mixins,
@@ -924,49 +925,27 @@ def close(self, request, pk=None):
         if request.method == "POST":
             finding_close = serializers.FindingCloseSerializer(
                 data=request.data,
+                context={"request": request},
             )
             if finding_close.is_valid():
-                finding.is_mitigated = finding_close.validated_data[
-                    "is_mitigated"
-                ]
-                if settings.EDITABLE_MITIGATED_DATA:
-                    finding.mitigated = (
-                        finding_close.validated_data["mitigated"]
-                        or timezone.now()
-                    )
-                else:
-                    finding.mitigated = timezone.now()
-                finding.mitigated_by = request.user
-                finding.active = False
-                finding.false_p = finding_close.validated_data.get(
-                    "false_p", False,
-                )
-                finding.duplicate = finding_close.validated_data.get(
-                    "duplicate", False,
-                )
-                finding.out_of_scope = finding_close.validated_data.get(
-                    "out_of_scope", False,
+                # Use shared helper to perform close operations
+                finding_helper.close_finding(
+                    finding=finding,
+                    user=request.user,
+                    is_mitigated=finding_close.validated_data["is_mitigated"],
+                    mitigated=(finding_close.validated_data.get("mitigated") if settings.EDITABLE_MITIGATED_DATA else timezone.now()),
+                    mitigated_by=finding_close.validated_data.get("mitigated_by") or (request.user if not finding_helper.can_edit_mitigated_data(request.user) else None),
+                    false_p=finding_close.validated_data.get("false_p", False),
+                    out_of_scope=finding_close.validated_data.get("out_of_scope", False),
+                    duplicate=finding_close.validated_data.get("duplicate", False),
+                    note_entry=finding_close.validated_data.get("note"),
+                    note_type=finding_close.validated_data.get("note_type"),
                 )
-
-                endpoints_status = finding.status_finding.all()
-                for e_status in endpoints_status:
-                    e_status.mitigated_by = request.user
-                    if settings.EDITABLE_MITIGATED_DATA:
-                        e_status.mitigated_time = (
-                            finding_close.validated_data["mitigated"]
-                            or timezone.now()
-                        )
-                    else:
-                        e_status.mitigated_time = timezone.now()
-                    e_status.mitigated = True
-                    e_status.last_modified = timezone.now()
-                    e_status.save()
-                finding.save()
             else:
                 return Response(
                     finding_close.errors, status=status.HTTP_400_BAD_REQUEST,
                 )
-        serialized_finding = serializers.FindingCloseSerializer(finding)
+        serialized_finding = serializers.FindingCloseSerializer(finding, context={"request": request})
         return Response(serialized_finding.data)
 
     @extend_schema(

dojo/finding/helper.py

@@ -7,10 +7,12 @@
 from django.db.models.signals import post_delete, pre_delete
 from django.db.utils import IntegrityError
 from django.dispatch.dispatcher import receiver
+from django.urls import reverse
 from django.utils import timezone
 from fieldsignals import pre_save_changed
 
 import dojo.jira_link.helper as jira_helper
+import dojo.risk_acceptance.helper as ra_helper
 from dojo.celery import app
 from dojo.decorators import dojo_async_task, dojo_model_from_id, dojo_model_to_id
 from dojo.endpoint.utils import save_endpoints_to_add
@@ -21,15 +23,18 @@
     Engagement,
     Finding,
     Finding_Group,
+    Notes,
     System_Settings,
     Test,
     Vulnerability_Id,
     Vulnerability_Id_Template,
 )
 from dojo.notes.helper import delete_related_notes
+from dojo.notifications.helper import create_notification
 from dojo.tools import tool_issue_updater
 from dojo.utils import (
     calculate_grade,
+    close_external_issue,
     do_dedupe_finding,
     do_false_positive_history,
     get_current_user,
@@ -271,7 +276,6 @@ def get_group_by_group_name(finding, finding_group_by_option):
     else:
         msg = f"Invalid group_by option {finding_group_by_option}"
         raise ValueError(msg)
-
     if group_name:
         return f"Findings in: {group_name}"
 
@@ -689,3 +693,91 @@ def save_vulnerability_ids_template(finding_template, vulnerability_ids):
         finding_template.cve = vulnerability_ids[0]
     else:
         finding_template.cve = None
+
+
+def close_finding(
+    *,
+    finding,
+    user,
+    is_mitigated,
+    mitigated,
+    mitigated_by,
+    false_p,
+    out_of_scope,
+    duplicate,
+    note_entry=None,
+    note_type=None,
+) -> None:
+    """
+    Shared close logic used by UI and API.
+
+    Handles status updates, endpoint statuses, risk acceptance, external issues,
+    JIRA sync, and notification.
+    """
+    # Core status updates
+    finding.is_mitigated = is_mitigated
+    now = timezone.now()
+    finding.mitigated = mitigated or now
+    finding.mitigated_by = mitigated_by or user
+    finding.active = False
+    finding.false_p = bool(false_p)
+    finding.out_of_scope = bool(out_of_scope)
+    finding.duplicate = bool(duplicate)
+    finding.under_review = False
+    finding.last_reviewed = finding.mitigated
+    finding.last_reviewed_by = user
+
+    # Create note if provided
+    new_note = None
+    if note_entry:
+        new_note = Notes.objects.create(
+            entry=note_entry,
+            author=user,
+            note_type=note_type,
+            date=finding.mitigated,
+        )
+        finding.notes.add(new_note)
+
+    # Endpoint statuses
+    for status in finding.status_finding.all():
+        status.mitigated_by = finding.mitigated_by
+        status.mitigated_time = finding.mitigated
+        status.mitigated = True
+        status.last_modified = timezone.now()
+        status.save()
+
+    # Risk acceptance
+    ra_helper.risk_unaccept(user, finding, perform_save=False)
+
+    # External issues (best effort)
+    close_external_issue(finding, "Closed by defectdojo", "github")
+
+    # JIRA sync
+    push_to_jira = False
+    finding_in_group = finding.has_finding_group
+    jira_issue_exists = finding.has_jira_issue or (
+        finding.finding_group and finding.finding_group.has_jira_issue
+    )
+    jira_instance = jira_helper.get_jira_instance(finding)
+    jira_project = jira_helper.get_jira_project(finding)
+    if jira_issue_exists:
+        push_to_jira = (
+            jira_helper.is_push_all_issues(finding)
+            or (jira_instance and jira_instance.finding_jira_sync)
+        )
+        if new_note and (getattr(jira_project, "push_notes", False) or push_to_jira) and not finding_in_group:
+            jira_helper.add_comment(finding, new_note, force_push=True)
+
+    # Persist and push JIRA if applicable
+    finding.save(push_to_jira=(push_to_jira and not finding_in_group))
+    if push_to_jira and finding_in_group:
+        jira_helper.push_to_jira(finding.finding_group)
+
+    # Notification
+    create_notification(
+        event="finding_closed",
+        title=f"Closing of {finding.title}",
+        finding=finding,
+        description=f'The finding "{finding.title}" was closed by {user}',
+        url=reverse("view_finding", args=(finding.id,)),
+    )

dojo/finding/views.py

@@ -112,7 +112,6 @@
     add_success_message_to_response,
     apply_cwe_to_template,
     calculate_grade,
-    close_external_issue,
     do_false_positive_history,
     get_page_items,
     get_page_items_and_count,
@@ -1140,68 +1139,22 @@ def close_finding(request, fid):
     if request.method == "POST":
         form = CloseFindingForm(request.POST, missing_note_types=missing_note_types)
 
-        close_external_issue(finding, "Closed by defectdojo", "github")
-
         if form.is_valid():
-            now = timezone.now()
-            new_note = form.save(commit=False)
-            new_note.author = request.user
-            new_note.date = form.cleaned_data.get("mitigated") or now
-            new_note.save()
-            finding.notes.add(new_note)
-
-            messages.add_message(
-                request, messages.SUCCESS, "Note Saved.", extra_tags="alert-success",
-            )
+            messages.add_message(request, messages.SUCCESS, "Note Saved.", extra_tags="alert-success")
 
             if len(missing_note_types) <= 1:
-                finding.active = False
-                now = timezone.now()
-                finding.mitigated = form.cleaned_data.get("mitigated") or now
-                finding.mitigated_by = (
-                    form.cleaned_data.get("mitigated_by") or request.user
+                finding_helper.close_finding(
+                    finding=finding,
+                    user=request.user,
+                    is_mitigated=True,
+                    mitigated=form.cleaned_data.get("mitigated"),
+                    mitigated_by=form.cleaned_data.get("mitigated_by") or request.user,
+                    false_p=form.cleaned_data.get("false_p", False),
+                    out_of_scope=form.cleaned_data.get("out_of_scope", False),
+                    duplicate=form.cleaned_data.get("duplicate", False),
+                    note_entry=form.cleaned_data.get("entry"),
+                    note_type=form.cleaned_data.get("note_type"),
                 )
-                finding.is_mitigated = True
-                finding.under_review = False
-                finding.last_reviewed = finding.mitigated
-                finding.last_reviewed_by = request.user
-                finding.false_p = form.cleaned_data.get("false_p", False)
-                finding.out_of_scope = form.cleaned_data.get("out_of_scope", False)
-                finding.duplicate = form.cleaned_data.get("duplicate", False)
-                endpoint_status = finding.status_finding.all()
-                for status in endpoint_status:
-                    status.mitigated_by = (
-                        form.cleaned_data.get("mitigated_by") or request.user
-                    )
-                    status.mitigated_time = form.cleaned_data.get("mitigated") or now
-                    status.mitigated = True
-                    status.last_modified = timezone.now()
-                    status.save()
-                # Clear the risk acceptance, if present
-                ra_helper.risk_unaccept(request.user, finding)
-
-                # Manage the jira status changes
-                push_to_jira = False
-                # Determine if the finding is in a group. if so, not push to jira
-                finding_in_group = finding.has_finding_group
-                # Check if there is a jira issue that needs to be updated
-                jira_issue_exists = finding.has_jira_issue or (finding.finding_group and finding.finding_group.has_jira_issue)
-                # fetch the project
-                jira_instance = jira_helper.get_jira_instance(finding)
-                jira_project = jira_helper.get_jira_project(finding)
-                # Only push if the finding is not in a group
-                if jira_issue_exists:
-                    # Determine if any automatic sync should occur
-                    push_to_jira = jira_helper.is_push_all_issues(finding) or jira_instance.finding_jira_sync
-                    # Add the closing note
-                    if (jira_project.push_notes or push_to_jira) and not finding_in_group:
-                        jira_helper.add_comment(finding, new_note, force_push=True)
-                # Save the finding
-                finding.save(push_to_jira=(push_to_jira and not finding_in_group))
-                # we only push the group after saving the finding to make sure
-                # the updated data of the finding is pushed as part of the group
-                if push_to_jira and finding_in_group:
-                    jira_helper.push_to_jira(finding.finding_group)
 
                 messages.add_message(
                     request,
@@ -1210,17 +1163,7 @@ def close_finding(request, fid):
                     extra_tags="alert-success",
                 )
 
-                # Note: this notification has not be moved to "@receiver(pre_save, sender=Finding)" method as many other notifications
-                # Because it could generate too much noise, we keep it here only for findings created by hand in WebUI
-                # TODO: but same should be implemented for API endpoint
-
-                create_notification(
-                    event="finding_closed",
-                    title=_("Closing of %s") % finding.title,
-                    finding=finding,
-                    description=f'The finding "{finding.title}" was closed by {request.user}',
-                    url=reverse("view_finding", args=(finding.id,)),
-                )
+                # Notification sent by helper
                 return HttpResponseRedirect(
                     reverse("view_test", args=(finding.test.id,)),
                 )

dojo/forms.py

@@ -1470,7 +1470,7 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
         self.fields["endpoints"].queryset = Endpoint.objects.filter(product=self.instance.test.engagement.product)
-        self.fields["mitigated_by"].queryset = get_authorized_users(Permissions.Test_Edit)
+        self.fields["mitigated_by"].queryset = get_authorized_users(Permissions.Finding_Edit)
 
         # do not show checkbox if finding is not accepted and simple risk acceptance is disabled
         # if checked, always show to allow unaccept also with full risk acceptance enabled
@@ -1908,7 +1908,7 @@ def __init__(self, *args, **kwargs):
             else False
 
         if self.can_edit_mitigated_data:
-            self.fields["mitigated_by"].queryset = get_authorized_users(Permissions.Test_Edit)
+            self.fields["mitigated_by"].queryset = get_authorized_users(Permissions.Finding_Edit)
             self.fields["mitigated"].initial = self.instance.mitigated
             self.fields["mitigated_by"].initial = self.instance.mitigated_by
         if disclaimer := get_system_setting("disclaimer_notes"):