reimport: close_old_findings must respect service field

Author: valentijnscholten

dojo/importers/default_reimporter.py

@@ -162,7 +162,15 @@ def process_findings(
         at import time
         """
         self.deduplication_algorithm = self.determine_deduplication_algorithm()
-        self.original_items = list(self.test.finding_set.all())
+        # Only process findings with the same service value (or None)
+        # Even though the service values is used in the hash_code calculation,
+        # we need to make sure there are no side effects such as closing findings
+        # for findings with a different service value
+        # https://github.com/DefectDojo/django-DefectDojo/issues/12754
+        original_findings = self.test.finding_set.all().filter(service=self.service)
+        logger.debug(f"original_findings_qyer: {original_findings.query}")
+        self.original_items = list(original_findings)
+        logger.debug(f"original_items: {[(item.id, item.hash_code) for item in self.original_items]}")
         self.new_items = []
         self.reactivated_items = []
         self.unchanged_items = []

dojo/importers/endpoint_manager.py

@@ -52,7 +52,7 @@ def add_endpoints_to_unsaved_finding(
                 finding=finding,
                 endpoint=ep,
                 defaults={"date": finding.date})
-        logger.debug(f"IMPORT_SCAN: {len(endpoints)} imported")
+        logger.debug(f"IMPORT_SCAN: {len(endpoints)} endpoints imported")
         return
 
     @dojo_async_task

unittests/dojo_test_case.py

@@ -594,7 +594,7 @@ def import_scan_with_params(self, filename, scan_type="ZAP Scan", engagement=1,
             return self.import_scan(payload, expected_http_status_code)
 
     def reimport_scan_with_params(self, test_id, filename, scan_type="ZAP Scan", engagement=1, minimum_severity="Low", *, active=True, verified=False, push_to_jira=None,
-                                  tags=None, close_old_findings=True, group_by=None, engagement_name=None, scan_date=None,
+                                  tags=None, close_old_findings=True, group_by=None, engagement_name=None, scan_date=None, service=None,
                                   product_name=None, product_type_name=None, auto_create_context=None, expected_http_status_code=201, test_title=None):
         with Path(filename).open(encoding="utf-8") as testfile:
             payload = {
@@ -640,6 +640,9 @@ def reimport_scan_with_params(self, test_id, filename, scan_type="ZAP Scan", eng
             if scan_date is not None:
                 payload["scan_date"] = scan_date
 
+            if service is not None:
+                payload["service"] = service
+
             return self.reimport_scan(payload, expected_http_status_code=expected_http_status_code)
 
     def endpoint_meta_import_scan_with_params(self, filename, product=1, product_name=None, *,

unittests/test_import_reimport.py

@@ -1335,6 +1335,53 @@ def test_import_param_close_old_findings_with_and_without_service_2(self):
         engagement_findings_count = Finding.objects.filter(test__engagement_id=1, test__test_type=test.test_type, active=True, is_mitigated=False).count()
         self.assertEqual(engagement_findings_count, 4)
 
+    # close_old_findings functionality: second import to different engagement with different service should not close findings from the first engagement
+    def test_import_param_close_old_findings_different_engagements_different_services(self):
+        logger.debug("importing clair report with service A into engagement 1")
+        with assertTestImportModelsCreated(self, imports=1, affected_findings=4, created=4):
+            import1 = self.import_scan_with_params(self.clair_few_findings, scan_type=self.scan_type_clair, engagement=1, close_old_findings=True, service="service_A")
+
+        test_id = import1["test"]
+        test = self.get_test(test_id)
+        findings = self.get_test_findings_api(test_id)
+        self.log_finding_summary_json_api(findings)
+        # imported count must match count in the report
+        self.assert_finding_count_json(4, findings)
+
+        # imported findings should be active in engagement 1
+        engagement1_findings = Finding.objects.filter(test__engagement_id=1, test__test_type=test.test_type, active=True, is_mitigated=False)
+        self.assertEqual(engagement1_findings.count(), 4)
+
+        # reimporting the same report into the same test with a different service should not close any findings and create 4 new findings
+        self.reimport_scan_with_params(test_id, self.clair_few_findings, scan_type=self.scan_type_clair, close_old_findings=True, service="service_B")
+
+        engagement1_active_finding_count = Finding.objects.filter(test__engagement_id=1, test__test_type=test.test_type, active=True, is_mitigated=False)
+        self.assertEqual(engagement1_active_finding_count.count(), 8)
+        engagement1_mitigated_finding_count = Finding.objects.filter(test__engagement_id=1, test__test_type=test.test_type, active=False, is_mitigated=True)
+        self.assertEqual(engagement1_mitigated_finding_count.count(), 0)
+        # verify findings from engagement 1 are still the same (not mitigated/closed)
+        for finding in engagement1_active_finding_count:
+            self.assertTrue(finding.active)
+            self.assertFalse(finding.is_mitigated)
+
+        # reimporting an empty report with service A should close all findings from the first import, but not the reimported ones with service B
+        self.reimport_scan_with_params(test_id, self.clair_empty, scan_type=self.scan_type_clair, close_old_findings=True, service="service_A")
+
+        engagement1_active_finding_count = Finding.objects.filter(test__engagement_id=1, test__test_type=test.test_type, active=True, is_mitigated=False)
+        self.assertEqual(engagement1_active_finding_count.count(), 4)
+        engagement1_mitigated_finding_count = Finding.objects.filter(test__engagement_id=1, test__test_type=test.test_type, active=False, is_mitigated=True)
+        self.assertEqual(engagement1_mitigated_finding_count.count(), 4)
+
+        for finding in engagement1_active_finding_count:
+            self.assertTrue(finding.active)
+            self.assertFalse(finding.is_mitigated)
+            self.assertEqual(finding.service, "service_B")
+
+        for finding in engagement1_mitigated_finding_count:
+            self.assertFalse(finding.active)
+            self.assertTrue(finding.is_mitigated)
+            self.assertEqual(finding.service, "service_A")
+
     def test_import_reimport_generic(self):
         """
         This test do a basic import and re-import of a generic JSON report
@@ -1873,7 +1920,7 @@ def import_scan_with_params_ui(self, filename, scan_type="ZAP Scan", engagement=
 
             return self.import_scan_ui(engagement, payload)
 
-    def reimport_scan_with_params_ui(self, test_id, filename, scan_type="ZAP Scan", minimum_severity="Low", *, active=True, verified=False, push_to_jira=None, tags=None, close_old_findings=True, scan_date=None):
+    def reimport_scan_with_params_ui(self, test_id, filename, scan_type="ZAP Scan", minimum_severity="Low", *, active=True, verified=False, push_to_jira=None, tags=None, close_old_findings=True, scan_date=None, service=None):
         # Mimic old functionality for active/verified to avoid breaking tests
         activePayload = "force_to_true"
         if not active:
@@ -1902,6 +1949,9 @@ def reimport_scan_with_params_ui(self, test_id, filename, scan_type="ZAP Scan",
             if scan_date is not None:
                 payload["scan_date"] = scan_date
 
+            if service is not None:
+                payload["service"] = service
+
             return self.reimport_scan_ui(test_id, payload)
 
 # Observations: