backport(importers): make bulk_new_finding_writes functional on 3.2.x

dojo/finding/models.py

@@ -565,21 +565,45 @@ def __init__(self, *args, **kwargs):
     def __str__(self):
         return self.title
 
-    def save(self, dedupe_option=True, rules_option=True, product_grading_option=True,  # noqa: FBT002
-             issue_updater_option=True, push_to_jira=False, user=None, *args, **kwargs):  # noqa: FBT002 - this is bit hard to fix nice have this universally fixed
-        logger.debug("Start saving finding of id " + str(self.id) + " dedupe_option:" + str(dedupe_option) + " (self.pk is %s)", "None" if self.pk is None else "not None")
-        from dojo.finding import helper as finding_helper  # noqa: PLC0415 -- lazy import, avoids circular dependency
-
-        is_new_finding = self.pk is None
-
-        # if not isinstance(self.date, (datetime, date)):
-        #     raise ValidationError(_("The 'date' field must be a valid date or datetime object."))
-
-        if not user:
-            from dojo.utils import get_current_user  # noqa: PLC0415 -- lazy import, avoids circular dependency
-            user = get_current_user()
+    @classmethod
+    def persisted_title(cls, title: str | None) -> str:
+        """
+        Return a title in the exact form a persisted finding carries.
+
+        The single source of truth for the title transform. Anything that needs to know
+        what a title *will* look like once stored -- notably a hash computed before the
+        row is written, which must match the hash computed after -- calls this instead of
+        repeating the transform.
+
+        Note that ``titlecase()`` is not merely a case change: it also normalizes
+        whitespace, collapsing consecutive newlines and turning tabs into spaces. A
+        reimplementation that truncated but did not titlecase therefore diverged for any
+        multi-line title, which is a bug that has already been paid for once.
+        """
+        return titlecase((title or "")[:cls._meta.get_field("title").max_length])
+
+    def derive_persisted_fields(self, *, dedupe_option: bool = True, is_new_finding: bool = False) -> None:
+        """
+        Normalize and derive the fields that must hold for any persisted finding.
+
+        This is the transform ``save()`` applies to a finding's own columns before the row
+        is written: title casing/truncation, blank-component normalization, the date
+        default, numerical severity, CVSS vector parsing, and the same-tool hash. It reads
+        configuration but performs no writes, touches no relations, and dispatches nothing,
+        so it is safe to call on an unsaved instance and on many instances in a loop.
+
+        It exists as a separate method so that batched writers -- anything using
+        ``bulk_create``/``bulk_update``, which bypass ``save()`` and its signals entirely --
+        can produce rows identical to the ones ``save()`` produces, by calling this rather
+        than reimplementing it.
+
+        Anything requiring a primary key -- ``found_by``, location/endpoint queries, SLA
+        expiry, status bookkeeping, post-save dispatch -- deliberately stays in ``save()``,
+        because a batched writer needs a genuinely different (set-based) implementation of
+        those rather than a shared one.
+        """
         # Title Casing
-        self.title = titlecase(self.title[:511])
+        self.title = Finding.persisted_title(self.title)
         # Normalize blank component fields to NULL so that findings without a component
         # group together. An empty string is treated as a distinct value from NULL by the
         # database, which would otherwise produce a separate "None" component group (SC-13073).
@@ -638,6 +662,22 @@ def save(self, dedupe_option=True, rules_option=True, product_grading_option=Tru
             elif (self.file_path is not None):
                 self.static_finding = True
 
+    def save(self, dedupe_option=True, rules_option=True, product_grading_option=True,  # noqa: FBT002
+             issue_updater_option=True, push_to_jira=False, user=None, *args, **kwargs):  # noqa: FBT002 - this is bit hard to fix nice have this universally fixed
+        logger.debug("Start saving finding of id " + str(self.id) + " dedupe_option:" + str(dedupe_option) + " (self.pk is %s)", "None" if self.pk is None else "not None")
+        from dojo.finding import helper as finding_helper  # noqa: PLC0415 -- lazy import, avoids circular dependency
+
+        is_new_finding = self.pk is None
+
+        # if not isinstance(self.date, (datetime, date)):
+        #     raise ValidationError(_("The 'date' field must be a valid date or datetime object."))
+
+        if not user:
+            from dojo.utils import get_current_user  # noqa: PLC0415 -- lazy import, avoids circular dependency
+            user = get_current_user()
+        self.derive_persisted_fields(dedupe_option=dedupe_option, is_new_finding=is_new_finding)
+
+        if is_new_finding:
             # because we have reduced the number of (super()).save() calls, the helper is no longer called for new findings
             # so we call it manually
             finding_helper.update_finding_status(self, user, changed_fields={"id": (None, None)})

dojo/importers/base_importer.py

@@ -1118,10 +1118,18 @@ def reconcile_cwes(self, finding: Finding) -> None:
         """Accumulate a delete+insert of Finding_CWE rows for a reimported finding when its CWEs changed."""
         new_cwes = set(self.finding_cwe_values(finding))
         # finding_cwe_set is prefetched on reimport candidates (build_candidate_scope_queryset).
-        existing_cwes = {row.cwe for row in finding.finding_cwe_set.all()}
+        # A finding that has not been written yet has no persisted CWEs, and reading the reverse
+        # relation on it raises ("instance needs to have a primary key value before this
+        # relationship can be used"). Treating it as empty is exact rather than a workaround:
+        # there are no rows to compare against, so every parsed CWE is new. This lets an importer
+        # that buffers inserts reconcile a finding's CWEs before flushing the buffer.
+        existing_cwes = {row.cwe for row in finding.finding_cwe_set.all()} if finding.pk else set()
         if existing_cwes == new_cwes:
             return
-        self.pending_cwe_deletes.append(finding.id)
+        # Nothing to delete for a finding with no row yet; appending None would put a NULL in the
+        # filter that flush_vulnerability_ids() builds.
+        if finding.pk:
+            self.pending_cwe_deletes.append(finding.id)
         self.pending_cwes.extend([Finding_CWE(finding=finding, cwe=cwe) for cwe in new_cwes])
 
     def flush_vulnerability_ids(self) -> None:

dojo/importers/default_reimporter.py

@@ -266,6 +266,64 @@ def add_new_finding_to_candidates(
                 f"Added finding {finding.id} (title: {finding.title}, severity: {finding.severity}) to candidates for next findings in this report",
             )
 
+    def _flush_post_processing_batch(
+        self,
+        batch_finding_ids,
+        batch_findings,
+        new_findings_in_batch,
+        findings_with_parser_tags,
+        **kwargs,
+    ) -> None:
+        """
+        Persist and dispatch everything accumulated since the last flush.
+
+        Extracted so it can run both on the size trigger inside the loop and once more
+        after it. Every step is a no-op on empty state, so the final call is safe and is
+        deliberately unconditional.
+        """
+        self.location_handler.persist()
+        self.flush_vulnerability_ids()
+        self.flush_burp_request_response()
+        # Apply parser-supplied tags for this batch before post-processing starts,
+        # so rules/deduplication tasks see the tags already on the findings.
+        bulk_apply_parser_tags(findings_with_parser_tags)
+        findings_with_parser_tags.clear()
+        # Apply import-time tags before post-processing so rules/deduplication see them.
+        self.apply_import_tags_for_batch(batch_findings)
+        # Apply inherited Product tags to NEWLY CREATED findings only
+        # (and their endpoints/locations) BEFORE post_process_findings_batch
+        # dispatches, so rules/dedup see inherited tags on .tags.
+        # Matched/existing findings already have inheritance applied from
+        # their original creation; re-running it on no-change reimports
+        # would be ~8 wasted queries per batch.
+        apply_inherited_tags_for_findings(new_findings_in_batch)
+        new_findings_in_batch.clear()
+        batch_findings.clear()
+        # Partition the batch by each finding's own push_to_jira flag so one
+        # finding's grouping state is not applied to the whole batch. Uniform
+        # batches (grouping disabled, or push_to_jira off) stay a single dispatch.
+        finding_ids_by_push: dict[bool, list[int]] = {}
+        for finding_id, finding_push_to_jira in batch_finding_ids:
+            finding_ids_by_push.setdefault(finding_push_to_jira, []).append(finding_id)
+        batch_finding_ids.clear()
+        for push_to_jira_batch, finding_ids_batch in finding_ids_by_push.items():
+            result = dojo_dispatch_task(
+                finding_helper.post_process_findings_batch,
+                finding_ids_batch,
+                dedupe_option=True,
+                rules_option=True,
+                product_grading_option=True,
+                issue_updater_option=True,
+                push_to_jira=push_to_jira_batch,
+                jira_instance_id=getattr(self.jira_instance, "id", None),
+                # 'async_wait' joins on this dispatch via AsyncResult.get(), so its
+                # result must be stored despite the global CELERY_TASK_IGNORE_RESULT.
+                **({"ignore_result": False} if self.deduplication_execution_mode == DEDUPLICATION_EXECUTION_MODE_ASYNC_WAIT else {}),
+                **self.post_processing_dispatch_kwargs(**kwargs),
+            )
+            if self.deduplication_execution_mode == DEDUPLICATION_EXECUTION_MODE_ASYNC_WAIT:
+                self.record_post_processing_result(result)
+
     def process_findings(
         self,
         parsed_findings: list[Finding],
@@ -394,7 +452,6 @@ def _process_findings_internal(
         for batch_start in range(0, len(cleaned_findings), match_batch_max_size):
             batch_end = min(batch_start + match_batch_max_size, len(cleaned_findings))
             unsaved_findings_batch = cleaned_findings[batch_start:batch_end]
-            is_final_batch = batch_end == len(cleaned_findings)
 
             logger.debug(f"Processing reimport batch {batch_start}-{batch_end} of {len(cleaned_findings)} findings")
 
@@ -406,8 +463,7 @@ def _process_findings_internal(
             )
 
             # Process each finding in the batch using pre-fetched candidates
-            for idx, unsaved_finding in enumerate(unsaved_findings_batch):
-                is_final = is_final_batch and idx == len(unsaved_findings_batch) - 1
+            for unsaved_finding in unsaved_findings_batch:
 
                 # Match any findings to this new one coming in using pre-fetched candidates
                 matched_findings = self.match_finding_to_candidate_reimport(
@@ -482,49 +538,28 @@ def _process_findings_internal(
                     # - Matching batches: optimize candidate fetching (solve 1+N query problem)
                     # - Deduplication batches: optimize bulk operations (larger batches = fewer queries)
                     # They don't need to be aligned since they optimize different operations.
-                    if len(batch_finding_ids) >= dedupe_batch_max_size or is_final:
-                        self.location_handler.persist()
-                        self.flush_vulnerability_ids()
-                        self.flush_burp_request_response()
-                        # Apply parser-supplied tags for this batch before post-processing starts,
-                        # so rules/deduplication tasks see the tags already on the findings.
-                        bulk_apply_parser_tags(findings_with_parser_tags)
-                        findings_with_parser_tags.clear()
-                        # Apply import-time tags before post-processing so rules/deduplication see them.
-                        self.apply_import_tags_for_batch(batch_findings)
-                        # Apply inherited Product tags to NEWLY CREATED findings only
-                        # (and their endpoints/locations) BEFORE post_process_findings_batch
-                        # dispatches, so rules/dedup see inherited tags on .tags.
-                        # Matched/existing findings already have inheritance applied from
-                        # their original creation; re-running it on no-change reimports
-                        # would be ~8 wasted queries per batch.
-                        apply_inherited_tags_for_findings(new_findings_in_batch)
-                        new_findings_in_batch.clear()
-                        batch_findings.clear()
-                        # Partition the batch by each finding's own push_to_jira flag so one
-                        # finding's grouping state is not applied to the whole batch. Uniform
-                        # batches (grouping disabled, or push_to_jira off) stay a single dispatch.
-                        finding_ids_by_push: dict[bool, list[int]] = {}
-                        for finding_id, finding_push_to_jira in batch_finding_ids:
-                            finding_ids_by_push.setdefault(finding_push_to_jira, []).append(finding_id)
-                        batch_finding_ids.clear()
-                        for push_to_jira_batch, finding_ids_batch in finding_ids_by_push.items():
-                            result = dojo_dispatch_task(
-                                finding_helper.post_process_findings_batch,
-                                finding_ids_batch,
-                                dedupe_option=True,
-                                rules_option=True,
-                                product_grading_option=True,
-                                issue_updater_option=True,
-                                push_to_jira=push_to_jira_batch,
-                                jira_instance_id=getattr(self.jira_instance, "id", None),
-                                # 'async_wait' joins on this dispatch via AsyncResult.get(), so its
-                                # result must be stored despite the global CELERY_TASK_IGNORE_RESULT.
-                                **({"ignore_result": False} if self.deduplication_execution_mode == DEDUPLICATION_EXECUTION_MODE_ASYNC_WAIT else {}),
-                                **self.post_processing_dispatch_kwargs(**kwargs),
-                            )
-                            if self.deduplication_execution_mode == DEDUPLICATION_EXECUTION_MODE_ASYNC_WAIT:
-                                self.record_post_processing_result(result)
+                    if len(batch_finding_ids) >= dedupe_batch_max_size:
+                        self._flush_post_processing_batch(
+                            batch_finding_ids,
+                            batch_findings,
+                            new_findings_in_batch,
+                            findings_with_parser_tags,
+                            **kwargs,
+                        )
+
+        # A final drain rather than an is_final flag inside the loop. The matched branch
+        # ends in `continue`, so a report whose last finding took that path never reached
+        # the in-loop is_final flush: everything appended since the previous size-triggered
+        # flush was silently dropped, with no deduplication, rules, issue updater or JIRA
+        # dispatch, and no parser or inherited tags for those findings. Draining here runs
+        # exactly once however the last iteration ended.
+        self._flush_post_processing_batch(
+            batch_finding_ids,
+            batch_findings,
+            new_findings_in_batch,
+            findings_with_parser_tags,
+            **kwargs,
+        )
 
         # No chord: tasks are dispatched immediately above per batch
 
@@ -997,7 +1032,7 @@ def process_finding_that_was_not_matched(
         unsaved_finding = self.process_cve(unsaved_finding)
         # Hash code is already calculated earlier as it's the primary matching criteria for reimport
         # Save it. Don't dedupe before endpoints/locations are added.
-        unsaved_finding.save_no_options()
+        self.persist_new_finding(unsaved_finding)
         finding = unsaved_finding
         # Force parsers to use unsaved_tags (stored in finding_post_processing function below)
         finding.tags = None
@@ -1017,6 +1052,26 @@ def process_finding_that_was_not_matched(
         self.process_request_response_pairs(unsaved_finding)
         return unsaved_finding, finding_will_be_grouped
 
+    def persist_new_finding(self, finding: Finding) -> None:
+        """
+        Write a finding the report did not match to an existing one.
+
+        This is intentionally a separate method (like get_original_findings and
+        get_reimport_match_candidates_for_batch) so downstream editions can override it
+        without copying the full process_finding_that_was_not_matched() implementation.
+
+        The override this exists for buffers new findings and writes them in bulk at the batch
+        boundary, where locations, vulnerability ids, tags and post-processing are already
+        flushed. Such an edition overrides this to accumulate, and _flush_post_processing_batch
+        to write the buffer before calling super() -- so the rows exist by the time anything in
+        that block reads a primary key.
+
+        Overriding this is the only supported way to defer the write. Everything after the call
+        in the caller -- grouping, the new_items list, request/response pairs -- is safe on an
+        unwritten finding, and the caller's remaining work is deliberately kept that way.
+        """
+        finding.save_no_options()
+
     def reconcile_vulnerability_ids(
         self,
         finding: Finding,
@@ -1038,7 +1093,12 @@ def reconcile_vulnerability_ids(
         # stays a no-query read.
         from dojo.vulnerability.queries import finding_vulnerability_id_strings  # noqa: PLC0415 -- avoid import cycle
 
-        existing_vuln_ids = set(finding_vulnerability_id_strings(finding))
+        # A finding that has not been written yet has no persisted vulnerability ids, and the read
+        # helper raises on it ("instance needs to have a primary key value before this relationship
+        # can be used"). Treating it as empty is exact rather than a workaround: there are no rows
+        # to compare against, so every parsed id is new. This lets an importer that buffers inserts
+        # reconcile a finding's vulnerability ids before flushing the buffer.
+        existing_vuln_ids = set(finding_vulnerability_id_strings(finding)) if finding.pk else set()
         new_vuln_ids = set(vulnerability_ids_to_process)
 
         # Early exit if unchanged — no DB work needed
@@ -1104,7 +1164,12 @@ def finding_post_processing(
         finding = self.reconcile_vulnerability_ids(finding)
         # Save the finding only if the cve field was changed by save_vulnerability_ids
         # This is temporary as the cve field will be phased out
-        if finding.cve != old_cve:
+        #
+        # Only for a finding that already has a row. This save exists to push a changed cve onto
+        # an existing record; for a finding an importer is still buffering there is nothing to
+        # update, and saving here would insert it early -- defeating the buffering and splitting
+        # one batched INSERT into per-finding ones. The value rides along when the buffer flushes.
+        if finding.cve != old_cve and finding.pk:
             finding.save()
         return finding