dedupe reopen: continue to try all match candidates (#14011)

valentijnscholten · web-flow · commit 3627081bc14f · 2026-01-04T15:26:17.000-07:00
* dedupe reopen: add test cases that prove the bug

* remove obsolete method

* dedupe reopen: proceed with next candidate if candidate is mitigated

* rename methods
diff --git a/dojo/finding/deduplication.py b/dojo/finding/deduplication.py
@@ -1,4 +1,5 @@
 import logging
+from collections.abc import Iterator
 from operator import attrgetter
 
 import hyperlink
@@ -127,7 +128,7 @@ def set_duplicate(new_finding, existing_finding):
         msg = "Can not add duplicate to itself"
         raise Exception(msg)
     if is_duplicate_reopen(new_finding, existing_finding):
-        msg = "Found a regression. Ignore this so that a new duplicate chain can be made"
+        msg = "Found a regression where a duplicate of a mitigated finding is found. We do not reopen this, but create a new duplicate chain as per @PR 9558: Deduplication: Do not reopen original finding"
         raise Exception(msg)
     if new_finding.duplicate and finding_mitigated(existing_finding):
         msg = "Skip this finding as we do not want to attach a new duplicate to a mitigated finding"
@@ -174,17 +175,6 @@ def finding_not_human_set_status(finding: Finding) -> bool:
     return finding.out_of_scope is False and finding.false_p is False
 
 
-def set_duplicate_reopen(new_finding, existing_finding):
-    logger.debug("duplicate reopen existing finding")
-    existing_finding.mitigated = new_finding.mitigated
-    existing_finding.is_mitigated = new_finding.is_mitigated
-    existing_finding.active = new_finding.active
-    existing_finding.verified = new_finding.verified
-    existing_finding.notes.create(author=existing_finding.reporter,
-                                    entry="This finding has been automatically re-opened as it was found in recent scans.")
-    existing_finding.save()
-
-
 def is_deduplication_on_engagement_mismatch(new_finding, to_duplicate_finding):
     if new_finding.test.engagement != to_duplicate_finding.test.engagement:
         deduplication_mismatch = new_finding.test.engagement.deduplication_on_engagement \
@@ -355,9 +345,9 @@ def _is_candidate_older(new_finding, candidate):
     return is_older
 
 
-def match_hash_candidate(new_finding, candidates_by_hash):
+def get_matches_from_hash_candidates(new_finding, candidates_by_hash) -> Iterator[Finding]:
     if new_finding.hash_code is None:
-        return None
+        return
     possible_matches = candidates_by_hash.get(new_finding.hash_code, [])
     deduplicationLogger.debug(f"Finding {new_finding.id}: Found {len(possible_matches)} findings with same hash_code, ids={[(c.id, c.hash_code) for c in possible_matches]}")
 
@@ -368,13 +358,12 @@ def match_hash_candidate(new_finding, candidates_by_hash):
             deduplicationLogger.debug("deduplication_on_engagement_mismatch, skipping dedupe.")
             continue
         if are_endpoints_duplicates(new_finding, candidate):
-            return candidate
-    return None
+            yield candidate
 
 
-def match_unique_id_candidate(new_finding, candidates_by_uid):
+def get_matches_from_unique_id_candidates(new_finding, candidates_by_uid) -> Iterator[Finding]:
     if new_finding.unique_id_from_tool is None:
-        return None
+        return
 
     possible_matches = candidates_by_uid.get(new_finding.unique_id_from_tool, [])
     deduplicationLogger.debug(f"Finding {new_finding.id}: Found {len(possible_matches)} findings with same unique_id_from_tool, ids={[(c.id, c.unique_id_from_tool) for c in possible_matches]}")
@@ -385,11 +374,10 @@ def match_unique_id_candidate(new_finding, candidates_by_uid):
         if is_deduplication_on_engagement_mismatch(new_finding, candidate):
             deduplicationLogger.debug("deduplication_on_engagement_mismatch, skipping dedupe.")
             continue
-        return candidate
-    return None
+        yield candidate
 
 
-def match_uid_or_hash_candidate(new_finding, candidates_by_uid, candidates_by_hash):
+def get_matches_from_uid_or_hash_candidates(new_finding, candidates_by_uid, candidates_by_hash) -> Iterator[Finding]:
     # Combine UID and hash candidates and walk oldest-first
     uid_list = candidates_by_uid.get(new_finding.unique_id_from_tool, []) if new_finding.unique_id_from_tool is not None else []
     hash_list = candidates_by_hash.get(new_finding.hash_code, []) if new_finding.hash_code is not None else []
@@ -404,15 +392,15 @@ def match_uid_or_hash_candidate(new_finding, candidates_by_uid, candidates_by_ha
             continue
         if is_deduplication_on_engagement_mismatch(new_finding, candidate):
             deduplicationLogger.debug("deduplication_on_engagement_mismatch, skipping dedupe.")
-            return None
+            continue
         if are_endpoints_duplicates(new_finding, candidate):
             deduplicationLogger.debug("UID_OR_HASH: endpoints match, returning candidate %s with test_type %s unique_id_from_tool %s hash_code %s", candidate.id, candidate.test.test_type, candidate.unique_id_from_tool, candidate.hash_code)
-            return candidate
-        deduplicationLogger.debug("UID_OR_HASH: endpoints mismatch, skipping candidate %s", candidate.id)
-    return None
+            yield candidate
+        else:
+            deduplicationLogger.debug("UID_OR_HASH: endpoints mismatch, skipping candidate %s", candidate.id)
 
 
-def match_legacy_candidate(new_finding, candidates_by_title, candidates_by_cwe):
+def get_matches_from_legacy_candidates(new_finding, candidates_by_title, candidates_by_cwe) -> Iterator[Finding]:
     # ---------------------------------------------------------
     # 1) Collects all the findings that have the same:
     #      (title  and static_finding and dynamic_finding)
@@ -469,8 +457,7 @@ def match_legacy_candidate(new_finding, candidates_by_title, candidates_by_cwe):
             + " flag_endpoints: " + str(flag_endpoints) + " flag_line_path:" + str(flag_line_path) + " flag_hash:" + str(flag_hash))
 
         if (flag_endpoints or flag_line_path) and flag_hash:
-            return candidate
-    return None
+            yield candidate
 
 
 def _dedupe_batch_hash_code(findings):
@@ -482,10 +469,10 @@ def _dedupe_batch_hash_code(findings):
         return
     for new_finding in findings:
         deduplicationLogger.debug(f"deduplication start for finding {new_finding.id} with DEDUPE_ALGO_HASH_CODE")
-        match = match_hash_candidate(new_finding, candidates_by_hash)
-        if match:
+        for match in get_matches_from_hash_candidates(new_finding, candidates_by_hash):
             try:
                 set_duplicate(new_finding, match)
+                break
             except Exception as e:
                 deduplicationLogger.debug(str(e))
 
@@ -499,12 +486,14 @@ def _dedupe_batch_unique_id(findings):
         return
     for new_finding in findings:
         deduplicationLogger.debug(f"deduplication start for finding {new_finding.id} with DEDUPE_ALGO_UNIQUE_ID_FROM_TOOL")
-        match = match_unique_id_candidate(new_finding, candidates_by_uid)
-        if match:
+        for match in get_matches_from_unique_id_candidates(new_finding, candidates_by_uid):
+            deduplicationLogger.debug(f"Trying to deduplicate finding {new_finding.id} against candidate {match.id}")
             try:
                 set_duplicate(new_finding, match)
+                deduplicationLogger.debug(f"Successfully deduplicated finding {new_finding.id} against candidate {match.id}")
+                break
             except Exception as e:
-                deduplicationLogger.debug(str(e))
+                deduplicationLogger.debug(f"Exception when deduplicating finding {new_finding.id} against candidate {match.id}: {e!s}")
 
 
 def _dedupe_batch_uid_or_hash(findings):
@@ -520,13 +509,12 @@ def _dedupe_batch_uid_or_hash(findings):
         if new_finding.duplicate:
             continue
 
-        match = match_uid_or_hash_candidate(new_finding, candidates_by_uid, existing_by_hash)
-        if match:
+        for match in get_matches_from_uid_or_hash_candidates(new_finding, candidates_by_uid, existing_by_hash):
             try:
                 set_duplicate(new_finding, match)
+                break
             except Exception as e:
                 deduplicationLogger.debug(str(e))
-                continue
 
 
 def _dedupe_batch_legacy(findings):
@@ -538,10 +526,10 @@ def _dedupe_batch_legacy(findings):
         return
     for new_finding in findings:
         deduplicationLogger.debug(f"deduplication start for finding {new_finding.id} with DEDUPE_ALGO_LEGACY")
-        match = match_legacy_candidate(new_finding, candidates_by_title, candidates_by_cwe)
-        if match:
+        for match in get_matches_from_legacy_candidates(new_finding, candidates_by_title, candidates_by_cwe):
             try:
                 set_duplicate(new_finding, match)
+                break
             except Exception as e:
                 deduplicationLogger.debug(str(e))
 
diff --git a/unittests/test_deduplication_logic.py b/unittests/test_deduplication_logic.py
@@ -9,6 +9,7 @@
 from django.core import serializers
 from django.utils import timezone
 
+from dojo.finding.deduplication import set_duplicate
 from dojo.importers.default_importer import DefaultImporter
 from dojo.models import (
     Development_Environment,
@@ -879,6 +880,136 @@ def test_extra_endpoints_unique_id(self):
         # expect duplicate: unique_id match regardless of extra endpoints
         self.assert_finding(finding_new, not_pk=124, duplicate=True, duplicate_finding_id=124, hash_code=finding_124.hash_code)
 
+    def test_regression_duplicate_reopen_unique_id(self):
+        """
+        Test the is_duplicate_reopen exception in set_duplicate.
+        When trying to attach a new active finding to a mitigated finding,
+        an exception should be raised.
+
+        Related to: https://github.com/DefectDojo/django-DefectDojo/issues/14010
+        """
+        # Get an existing finding with unique_id (finding 124)
+        finding_124 = Finding.objects.get(id=124)
+        original_unique_id = finding_124.unique_id_from_tool
+
+        # Mitigate the existing finding
+        finding_124.active = False
+        finding_124.is_mitigated = True
+        finding_124.mitigated = timezone.now()
+        finding_124.out_of_scope = False
+        finding_124.false_p = False
+        finding_124.save(dedupe_option=False)
+
+        # Create a new finding with the same unique_id that is active (not mitigated)
+        finding_new, _ = self.copy_and_reset_finding(find_id=124)
+        finding_new.active = True
+        finding_new.is_mitigated = False
+        finding_new.mitigated = None
+        finding_new.unique_id_from_tool = original_unique_id
+        finding_new.save(dedupe_option=False)
+
+        # Try to deduplicate - this should raise an exception
+        with self.assertRaises(Exception) as context:
+            set_duplicate(finding_new, finding_124)
+
+        self.assertIn("Found a regression", str(context.exception))
+
+    def test_duplicate_attached_to_mitigated_finding_unique_id(self):
+        """
+        Test the exception when attaching a finding that is already marked as duplicate
+        to a mitigated finding.
+
+        Related to: https://github.com/DefectDojo/django-DefectDojo/issues/14010
+        """
+        # Get an existing finding with unique_id (finding 124)
+        finding_124 = Finding.objects.get(id=124)
+        original_unique_id = finding_124.unique_id_from_tool
+
+        # Mitigate the existing finding
+        finding_124.active = False
+        finding_124.is_mitigated = True
+        finding_124.mitigated = timezone.now()
+        finding_124.save(dedupe_option=False)
+
+        # Create a new finding that is already marked as duplicate AND also mitigated
+        # This ensures that is_duplicate_reopen doesn't trigger first
+        finding_new, _ = self.copy_and_reset_finding(find_id=124)
+        finding_new.duplicate = True
+        finding_new.active = False
+        finding_new.is_mitigated = True
+        finding_new.mitigated = timezone.now()
+        finding_new.unique_id_from_tool = original_unique_id
+        finding_new.save(dedupe_option=False)
+
+        # Try to deduplicate - this should raise an exception
+        with self.assertRaises(Exception) as context:
+            set_duplicate(finding_new, finding_124)
+
+        self.assertIn("Skip this finding as we do not want to attach a new duplicate to a mitigated finding", str(context.exception))
+
+    def test_multiple_findings_same_unique_id_mixed_states_unique_id(self):
+        """
+        Test deduplication with multiple findings having the same unique_id,
+        where some are mitigated and some are active. The deduplication should
+        skip mitigated ones and use the first active one.
+
+        Related to: https://github.com/DefectDojo/django-DefectDojo/issues/14010
+        """
+        # Get an existing finding with unique_id (finding 124)
+        finding_124 = Finding.objects.get(id=124)
+        original_unique_id = finding_124.unique_id_from_tool
+
+        # Mitigate the original finding so it's not a candidate
+        finding_124.active = False
+        finding_124.is_mitigated = True
+        finding_124.mitigated = timezone.now()
+        finding_124.save()
+
+        # Create multiple findings with the same unique_id
+        # First: active finding (this should become the original)
+        finding_active, _ = self.copy_and_reset_finding(find_id=124)
+        finding_active.active = True
+        finding_active.is_mitigated = False
+        finding_active.mitigated = None
+        finding_active.unique_id_from_tool = original_unique_id
+        finding_active.save(dedupe_option=False)  # Don't deduplicate so it remains active
+
+        # Second: mitigated finding
+        finding_mitigated, _ = self.copy_and_reset_finding(find_id=124)
+        finding_mitigated.active = False
+        finding_mitigated.is_mitigated = True
+        finding_mitigated.mitigated = timezone.now()
+        finding_mitigated.out_of_scope = False
+        finding_mitigated.false_p = False
+        finding_mitigated.unique_id_from_tool = original_unique_id
+        finding_mitigated.save(dedupe_option=False)
+
+        # Ensure mitigated finding has lower ID (will be checked first)
+        if finding_mitigated.id > finding_active.id:
+            # Swap by deleting and recreating in correct order
+            finding_mitigated.delete()
+            finding_mitigated, _ = self.copy_and_reset_finding(find_id=124)
+            finding_mitigated.active = False
+            finding_mitigated.is_mitigated = True
+            finding_mitigated.mitigated = timezone.now()
+            finding_mitigated.out_of_scope = False
+            finding_mitigated.false_p = False
+            finding_mitigated.unique_id_from_tool = original_unique_id
+            finding_mitigated.save(dedupe_option=False)
+
+        # Create a new finding with the same unique_id
+        # It should skip the mitigated finding and deduplicate against the active one
+        finding_new, _ = self.copy_and_reset_finding(find_id=124)
+        finding_new.active = True
+        finding_new.is_mitigated = False
+        finding_new.unique_id_from_tool = original_unique_id
+        finding_new.save()
+
+        # The new finding should be marked as duplicate of the active finding,
+        # not the mitigated one (even though mitigated has lower ID)
+        self.assert_finding(finding_new, duplicate=True, duplicate_finding_id=finding_active.id)
+        self.assertNotEqual(finding_new.duplicate_finding.id, finding_mitigated.id)
+
     # algo unique_id_or_hash_code Veracode scan
 
     def test_identical_unique_id_or_hash_code(self):
diff --git a/unittests/test_import_reimport.py b/unittests/test_import_reimport.py
