diff --git a/dojo/importers/base_importer.py b/dojo/importers/base_importer.py index d383de7b5a..d5cb727b8e 100644 --- a/dojo/importers/base_importer.py +++ b/dojo/importers/base_importer.py @@ -889,6 +889,20 @@ def sanitize_severity( # Return the finding if all else is good return finding + def persist_new_findings(self, prepared_findings: list[Finding]) -> list[Finding]: + """ + Persist a batch of new findings that have already been fully prepared + (scalar overrides applied, hash_code computed) and have no primary key yet. + + Default: an ordinary per-instance save, in the order given. Exists as an + override seam so a downstream edition can swap the write strategy (e.g. a + bulk insert) without reimplementing the grouping/tagging/location/ + vulnerability-id processing that runs on the returned, now-saved findings. + """ + for finding in prepared_findings: + finding.save_no_options() + return prepared_findings + def process_finding_groups( self, finding: Finding, diff --git a/dojo/importers/default_importer.py b/dojo/importers/default_importer.py index 105cb1bf9c..98e504cf8b 100644 --- a/dojo/importers/default_importer.py +++ b/dojo/importers/default_importer.py @@ -218,116 +218,124 @@ def _process_findings_internal( continue cleaned_findings.append(sanitized) - for idx, unsaved_finding in enumerate(cleaned_findings): - is_final_finding = idx == len(cleaned_findings) - 1 - - # Some parsers provide "mitigated" field but do not set timezone (because they are probably not available in the report) - # Finding.mitigated is DateTimeField and it requires timezone - if unsaved_finding.mitigated and not unsaved_finding.mitigated.tzinfo: - unsaved_finding.mitigated = unsaved_finding.mitigated.replace(tzinfo=self.now.tzinfo) - # Set some explicit fields on the finding - unsaved_finding.test = self.test - unsaved_finding.reporter = self.user - unsaved_finding.last_reviewed_by = self.user - unsaved_finding.last_reviewed = self.now - logger.debug("process_parsed_finding: unique_id_from_tool: %s, hash_code: %s, active from report: %s, verified from report: %s", unsaved_finding.unique_id_from_tool, unsaved_finding.hash_code, unsaved_finding.active, unsaved_finding.verified) - # indicates an override. Otherwise, do not change the value of unsaved_finding.active - if self.active is not None: - unsaved_finding.active = self.active - # indicates an override. Otherwise, do not change the value of verified - if self.verified is not None: - unsaved_finding.verified = self.verified - # scan_date was provided, override value from parser - if self.scan_date_override: - unsaved_finding.date = self.scan_date.date() - if self.service is not None: - unsaved_finding.service = self.service - - # Parsers shouldn't use the tags field, and use unsaved_tags instead. - # Merge any tags set by parser into unsaved_tags - tags_from_parser = unsaved_finding.tags if isinstance(unsaved_finding.tags, list) else [] - unsaved_tags_from_parser = unsaved_finding.unsaved_tags if isinstance(unsaved_finding.unsaved_tags, list) else [] - merged_tags = unsaved_tags_from_parser + tags_from_parser - if merged_tags: - unsaved_finding.unsaved_tags = merged_tags - unsaved_finding.tags = None - finding = self.process_cve(unsaved_finding) - # Calculate hash_code before saving based on unsaved_endpoints/unsaved_locations and unsaved_vulnerability_ids - finding.set_hash_code(True) + for batch_start in range(0, len(cleaned_findings), batch_max_size): + batch_end = min(batch_start + batch_max_size, len(cleaned_findings)) + is_final_batch = batch_end == len(cleaned_findings) + + # Prepare this batch's findings (scalar overrides, tag merge, hash_code) before + # any of them are persisted, so persist_new_findings() can write the whole batch + # in one call instead of one row at a time. + prepared_findings = [] + for unsaved_finding in cleaned_findings[batch_start:batch_end]: + # Some parsers provide "mitigated" field but do not set timezone (because they are probably not available in the report) + # Finding.mitigated is DateTimeField and it requires timezone + if unsaved_finding.mitigated and not unsaved_finding.mitigated.tzinfo: + unsaved_finding.mitigated = unsaved_finding.mitigated.replace(tzinfo=self.now.tzinfo) + # Set some explicit fields on the finding + unsaved_finding.test = self.test + unsaved_finding.reporter = self.user + unsaved_finding.last_reviewed_by = self.user + unsaved_finding.last_reviewed = self.now + logger.debug("process_parsed_finding: unique_id_from_tool: %s, hash_code: %s, active from report: %s, verified from report: %s", unsaved_finding.unique_id_from_tool, unsaved_finding.hash_code, unsaved_finding.active, unsaved_finding.verified) + # indicates an override. Otherwise, do not change the value of unsaved_finding.active + if self.active is not None: + unsaved_finding.active = self.active + # indicates an override. Otherwise, do not change the value of verified + if self.verified is not None: + unsaved_finding.verified = self.verified + # scan_date was provided, override value from parser + if self.scan_date_override: + unsaved_finding.date = self.scan_date.date() + if self.service is not None: + unsaved_finding.service = self.service + + # Parsers shouldn't use the tags field, and use unsaved_tags instead. + # Merge any tags set by parser into unsaved_tags + tags_from_parser = unsaved_finding.tags if isinstance(unsaved_finding.tags, list) else [] + unsaved_tags_from_parser = unsaved_finding.unsaved_tags if isinstance(unsaved_finding.unsaved_tags, list) else [] + merged_tags = unsaved_tags_from_parser + tags_from_parser + if merged_tags: + unsaved_finding.unsaved_tags = merged_tags + unsaved_finding.tags = None + finding = self.process_cve(unsaved_finding) + # Calculate hash_code before saving based on unsaved_endpoints/unsaved_locations and unsaved_vulnerability_ids + finding.set_hash_code(True) + prepared_findings.append(finding) # postprocessing will be done after processing related fields like locations, vulnerability ids, etc. - unsaved_finding.save_no_options() - - # Determine how the finding should be grouped - finding_will_be_grouped = self.process_finding_groups( - finding, - group_names_to_findings_dict, - ) - # Process any request/response pairs - self.process_request_response_pairs(finding) - self.process_locations(finding, self.endpoints_to_add) - # Parsers must use unsaved_tags to store tags, so we can clean them. - # Accumulate for bulk application after the loop (O(unique_tags) instead of O(N·T)). - cleaned_tags = clean_tags(finding.unsaved_tags) - if isinstance(cleaned_tags, list): - findings_with_parser_tags.append((finding, cleaned_tags)) - elif isinstance(cleaned_tags, str): - findings_with_parser_tags.append((finding, [cleaned_tags])) - # Process any files - self.process_files(finding) - # Process vulnerability IDs - finding = self.store_vulnerability_ids(finding) - # Categorize this finding as a new one - new_findings.append(finding) - # all data is already saved on the finding, we only need to trigger post processing in batches - logger.debug("process_findings: self.push_to_jira=%s, self.findings_groups_enabled=%s, self.group_by=%s", - self.push_to_jira, self.findings_groups_enabled, self.group_by) - push_to_jira = self.push_to_jira and ((not self.findings_groups_enabled or not self.group_by) or not finding_will_be_grouped) - logger.debug("process_findings: computed push_to_jira=%s", push_to_jira) - batch_finding_ids.append((finding.id, push_to_jira)) - batch_findings.append(finding) - - # If batch is full or we're at the end, persist locations/endpoints and dispatch - if len(batch_finding_ids) >= batch_max_size or is_final_finding: - 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 this batch's findings (and - # their endpoints/locations) BEFORE post_process_findings_batch - # dispatches, so rules/dedup see inherited tags on .tags. - apply_inherited_tags_for_findings(batch_findings) - 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(): - logger.debug("process_findings: dispatching batch with push_to_jira=%s (batch_size=%d, is_final=%s)", - push_to_jira_batch, len(finding_ids_batch), is_final_finding) - 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, - # '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) + saved_findings = self.persist_new_findings(prepared_findings) + + for saved_finding in saved_findings: + finding = saved_finding + # Determine how the finding should be grouped + finding_will_be_grouped = self.process_finding_groups( + finding, + group_names_to_findings_dict, + ) + # Process any request/response pairs + self.process_request_response_pairs(finding) + self.process_locations(finding, self.endpoints_to_add) + # Parsers must use unsaved_tags to store tags, so we can clean them. + # Accumulate for bulk application after the loop (O(unique_tags) instead of O(N·T)). + cleaned_tags = clean_tags(finding.unsaved_tags) + if isinstance(cleaned_tags, list): + findings_with_parser_tags.append((finding, cleaned_tags)) + elif isinstance(cleaned_tags, str): + findings_with_parser_tags.append((finding, [cleaned_tags])) + # Process any files + self.process_files(finding) + # Process vulnerability IDs + finding = self.store_vulnerability_ids(finding) + # Categorize this finding as a new one + new_findings.append(finding) + # all data is already saved on the finding, we only need to trigger post processing in batches + logger.debug("process_findings: self.push_to_jira=%s, self.findings_groups_enabled=%s, self.group_by=%s", + self.push_to_jira, self.findings_groups_enabled, self.group_by) + push_to_jira = self.push_to_jira and ((not self.findings_groups_enabled or not self.group_by) or not finding_will_be_grouped) + logger.debug("process_findings: computed push_to_jira=%s", push_to_jira) + batch_finding_ids.append((finding.id, push_to_jira)) + batch_findings.append(finding) + + # Persist locations/endpoints and dispatch post-processing for this batch + 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 this batch's findings (and + # their endpoints/locations) BEFORE post_process_findings_batch + # dispatches, so rules/dedup see inherited tags on .tags. + apply_inherited_tags_for_findings(batch_findings) + 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(): + logger.debug("process_findings: dispatching batch with push_to_jira=%s (batch_size=%d, is_final=%s)", + push_to_jira_batch, len(finding_ids_batch), is_final_batch) + 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, + # '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) # No chord: tasks are dispatched immediately above per batch diff --git a/dojo/importers/default_reimporter.py b/dojo/importers/default_reimporter.py index 4f42cbe46d..7ca3895bce 100644 --- a/dojo/importers/default_reimporter.py +++ b/dojo/importers/default_reimporter.py @@ -328,6 +328,10 @@ def _process_findings_internal( self.reactivated_items = [] self.unchanged_items = [] self.group_names_to_findings_dict = {} + # New findings queued by process_finding_that_was_not_matched, drained (persisted + # via persist_new_findings, then post-processed) once per matching batch -- see + # _drain_pending_new_findings. + self._pending_new_findings: list[tuple[Finding, Finding, bool]] = [] logger.debug(f"starting reimport of {len(parsed_findings) if parsed_findings else 0} items.") logger.debug("STEP 1: looping over findings from the reimported report and trying to match them to existing findings") @@ -426,6 +430,18 @@ def _process_findings_internal( # Determine how to proceed based on whether matches were found or not if matched_findings: existing_finding = matched_findings[0] + if existing_finding.pk is None: + # existing_finding is a same-report duplicate this same matching + # batch queued earlier (add_new_finding_to_candidates) and hasn't + # drained yet. Persist and post-process it now -- process_matched_finding + # and finding_post_processing below both require a real primary key. + self._finalize_specific_pending_new_finding( + existing_finding, + batch_findings_to_dispatch, + batch_findings, + new_findings_in_batch, + findings_with_parser_tags, + ) finding, force_continue = self.process_matched_finding( unsaved_finding, existing_finding, @@ -446,48 +462,23 @@ def _process_findings_internal( unsaved_finding, self.user, ) - else: - finding, finding_will_be_grouped = self.process_finding_that_was_not_matched(unsaved_finding) - - # Add newly created finding to candidates for subsequent findings in this batch - self.add_new_finding_to_candidates( - finding, - candidates_by_hash, - candidates_by_uid, - candidates_by_key, - ) - if finding: - new_findings_in_batch.append(finding) - - # This condition __appears__ to always be true, but am afraid to remove it - if finding: - # Process the rest of the items on the finding + # Matched findings already have a primary key: process and dispatch + # immediately, exactly as before. + # + # Post-processing batches (deduplication, rules, etc.) are separate from matching batches. + # In reimport scenarios, typically most findings match existing ones, so this check + # crossing dedupe_batch_max_size is normally driven by matched findings, not new ones. + # New findings are queued (see the else branch) and drained once per matching batch + # instead -- see _drain_pending_new_findings. finding = self.finding_post_processing( finding, unsaved_finding, - is_matched_finding=bool(matched_findings), + is_matched_finding=True, tag_accumulator=findings_with_parser_tags, ) - # all data is already saved on the finding, we only need to trigger post processing in batches push_to_jira = self.push_to_jira and ((not self.findings_groups_enabled or not self.group_by) or not finding_will_be_grouped) batch_findings_to_dispatch.append((finding, push_to_jira)) batch_findings.append(finding) - - # Post-processing batches (deduplication, rules, etc.) are separate from matching batches. - # These batches only contain "new" findings that were saved (not matched to existing findings). - # In reimport scenarios, typically most findings match existing ones, so only a small fraction - # of findings in each matching batch become new findings that need deduplication. - # - # We accumulate finding IDs across matching batches rather than dispatching at the end of each - # matching batch. This ensures deduplication batches stay close to the intended batch size - # (e.g., 1000 findings) for optimal bulk operation efficiency, even when only ~10% of findings - # in matching batches are new. If we dispatched at the end of each matching batch, we would - # end up with many small deduplication batches (e.g., ~100 findings each), reducing efficiency. - # - # The two batch types serve different purposes: - # - 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_findings_to_dispatch) >= dedupe_batch_max_size: self._flush_post_processing_batch( batch_findings_to_dispatch, @@ -496,6 +487,44 @@ def _process_findings_internal( findings_with_parser_tags, **kwargs, ) + else: + finding, finding_will_be_grouped = self.process_finding_that_was_not_matched(unsaved_finding) + + # Add newly created finding to candidates for subsequent findings in this + # batch. finding has no primary key yet -- match_finding_to_candidate_reimport + # and add_new_finding_to_candidates are written to tolerate that. + self.add_new_finding_to_candidates( + finding, + candidates_by_hash, + candidates_by_uid, + candidates_by_key, + ) + if finding: + # Deferred: finding_post_processing() (vulnerability-id/CWE + # reconciliation, file attachment) needs a real primary key, which + # persist_new_findings() only assigns once this matching batch + # finishes -- see _drain_pending_new_findings. + self._pending_new_findings.append((finding, unsaved_finding, finding_will_be_grouped)) + + # Persist this matching batch's new findings (bulk, via persist_new_findings) and + # run their post-processing before the NEXT matching batch's + # get_reimport_match_candidates_for_batch() queries the database -- otherwise a + # same-report duplicate created in this batch would still be unsaved and invisible + # to that fresh query. + self._drain_pending_new_findings( + batch_findings_to_dispatch, + batch_findings, + new_findings_in_batch, + findings_with_parser_tags, + ) + if len(batch_findings_to_dispatch) >= dedupe_batch_max_size: + self._flush_post_processing_batch( + batch_findings_to_dispatch, + batch_findings, + new_findings_in_batch, + findings_with_parser_tags, + **kwargs, + ) # A final drain instead of an is_final flag inside the loop. The matched branch's # force_continue skips the rest of the loop body, so a report whose last sorted @@ -534,6 +563,112 @@ def _process_findings_internal( return self.new_items, self.reactivated_items, self.to_mitigate, self.untouched + def _finalize_pending_new_finding( + self, + finding: Finding, + unsaved_finding: Finding, + batch_findings_to_dispatch: list[tuple[Finding, bool]], + batch_findings: list[Finding], + new_findings_in_batch: list[Finding], + findings_with_parser_tags: list[tuple], + *, + finding_will_be_grouped: bool, + ) -> None: + """ + Run one now-persisted new finding's post-processing and queue it for dispatch. + + `finding` must already have a primary key (persist_new_findings() has run on it). + Shared by _drain_pending_new_findings (the normal per-matching-batch case) and + _finalize_specific_pending_new_finding (the on-demand case: a later finding in + this report just matched against this one while it was still pending). + """ + self.new_items.append(finding) + new_findings_in_batch.append(finding) + finding = self.finding_post_processing( + finding, + unsaved_finding, + is_matched_finding=False, + tag_accumulator=findings_with_parser_tags, + ) + push_to_jira = self.push_to_jira and ((not self.findings_groups_enabled or not self.group_by) or not finding_will_be_grouped) + batch_findings_to_dispatch.append((finding, push_to_jira)) + batch_findings.append(finding) + + def _finalize_specific_pending_new_finding( + self, + existing_finding: Finding, + batch_findings_to_dispatch: list[tuple[Finding, bool]], + batch_findings: list[Finding], + new_findings_in_batch: list[Finding], + findings_with_parser_tags: list[tuple], + ) -> None: + """ + Persist and post-process one specific still-pending new finding immediately, + out of its normal per-matching-batch drain order. + + Called when a later finding in this report matches `existing_finding` while it + is still sitting in `_pending_new_findings` (a same-report duplicate queued by + add_new_finding_to_candidates earlier in this same matching batch, per issue + #3958): process_matched_finding and finding_post_processing both need a real + primary key for `existing_finding` right now, not whenever this batch happens to + drain -- see process_matched_active_finding's `.save_no_options()`/`.notes.add()` + calls and finding_post_processing's CWE/vulnerability-id reconciliation. + """ + for index, (finding, unsaved_finding, finding_will_be_grouped) in enumerate(self._pending_new_findings): + if finding is existing_finding: + del self._pending_new_findings[index] + (saved_finding,) = self.persist_new_findings([finding]) + self._finalize_pending_new_finding( + saved_finding, + unsaved_finding, + batch_findings_to_dispatch, + batch_findings, + new_findings_in_batch, + findings_with_parser_tags, + finding_will_be_grouped=finding_will_be_grouped, + ) + return + + def _drain_pending_new_findings( + self, + batch_findings_to_dispatch: list[tuple[Finding, bool]], + batch_findings: list[Finding], + new_findings_in_batch: list[Finding], + findings_with_parser_tags: list[tuple], + ) -> None: + """ + Persist this matching batch's queued new findings and run their post-processing. + + Called once per matching batch -- not per finding (like matched findings get, + since they already have a primary key) and not at the larger dedupe-batch boundary + (like the dispatch flush) -- because get_reimport_match_candidates_for_batch()'s + database query for the NEXT matching batch would not find a same-report duplicate + created here while it is still unsaved. finding_post_processing() itself tolerates + an unsaved instance (see the finding.pk guard before its CVE save), which is what + lets a downstream edition defer the write further still, via persist_new_finding() + or persist_new_findings() -- but the candidate query has no such tolerance, so the + stock path saves eagerly here regardless. + + A finding matched against earlier in this same batch (see + _finalize_specific_pending_new_finding) is already gone from + `_pending_new_findings` by the time this runs. + """ + if not self._pending_new_findings: + return + pending, self._pending_new_findings = self._pending_new_findings, [] + prepared_findings = [finding for finding, _unsaved_finding, _grouped in pending] + saved_findings = self.persist_new_findings(prepared_findings) + for (_finding, unsaved_finding, finding_will_be_grouped), saved_finding in zip(pending, saved_findings, strict=True): + self._finalize_pending_new_finding( + saved_finding, + unsaved_finding, + batch_findings_to_dispatch, + batch_findings, + new_findings_in_batch, + findings_with_parser_tags, + finding_will_be_grouped=finding_will_be_grouped, + ) + def _flush_post_processing_batch( self, batch_findings_to_dispatch: list[tuple[Finding, bool]], @@ -737,7 +872,17 @@ def match_finding_to_candidate_reimport( candidates_by_key: Dictionary mapping (title_lower, severity) to list of findings (for legacy algorithm) Returns: - List of matching findings, ordered by id + List of matching findings, in priority order. Deliberately NOT re-sorted by id + here: a candidate can be a same-report duplicate queued earlier in this matching + batch by add_new_finding_to_candidates, which has no primary key yet (persist_new_findings + only assigns one once the whole batch is queued -- see _drain_pending_new_findings), + and `.id` sorting a list containing None alongside real ids raises. The lists in + candidates_by_hash/candidates_by_uid/candidates_by_key are already in the right + order by construction: get_reimport_match_candidates_for_batch fetches existing + candidates pre-sorted by id, and add_new_finding_to_candidates only ever appends + to that same list afterward, in the order findings are processed -- so returning + them as-is preserves "existing findings first, then same-report ones in the order + they were created" without needing every candidate to have a real id yet. """ deduplicationLogger.debug("matching finding for reimport using algorithm: %s", self.deduplication_algorithm) @@ -745,14 +890,12 @@ def match_finding_to_candidate_reimport( if self.deduplication_algorithm == "hash_code": if candidates_by_hash is None or unsaved_finding.hash_code is None: return [] - matches = candidates_by_hash.get(unsaved_finding.hash_code, []) - return sorted(matches, key=lambda f: f.id) + return list(candidates_by_hash.get(unsaved_finding.hash_code, [])) if self.deduplication_algorithm == "unique_id_from_tool": if candidates_by_uid is None or unsaved_finding.unique_id_from_tool is None: return [] - matches = candidates_by_uid.get(unsaved_finding.unique_id_from_tool, []) - return sorted(matches, key=lambda f: f.id) + return list(candidates_by_uid.get(unsaved_finding.unique_id_from_tool, [])) if self.deduplication_algorithm == "unique_id_from_tool_or_hash_code": if candidates_by_hash is None and candidates_by_uid is None: @@ -761,28 +904,29 @@ def match_finding_to_candidate_reimport( if unsaved_finding.hash_code is None and unsaved_finding.unique_id_from_tool is None: return [] - # Collect matches from both hash_code and unique_id_from_tool - matches_by_id = {} + # Collect matches from both hash_code and unique_id_from_tool, de-duplicated by + # object identity rather than `.id`: a same-report candidate matched via both + # keys has no id yet to de-duplicate on, but is still exactly one Python object + # within this candidate-building pass, so `id()` is a safe, always-available key. + matches_by_identity = {} if unsaved_finding.hash_code is not None: hash_matches = candidates_by_hash.get(unsaved_finding.hash_code, []) for match in hash_matches: - matches_by_id[match.id] = match + matches_by_identity[id(match)] = match if unsaved_finding.unique_id_from_tool is not None: uid_matches = candidates_by_uid.get(unsaved_finding.unique_id_from_tool, []) for match in uid_matches: - matches_by_id[match.id] = match + matches_by_identity[id(match)] = match - matches = list(matches_by_id.values()) - return sorted(matches, key=lambda f: f.id) + return list(matches_by_identity.values()) if self.deduplication_algorithm == "legacy": if candidates_by_key is None or not unsaved_finding.title: return [] key = (unsaved_finding.title.lower(), unsaved_finding.severity) - matches = candidates_by_key.get(key, []) - return sorted(matches, key=lambda f: f.id) + return list(candidates_by_key.get(key, [])) logger.error(f'Internal error: unexpected deduplication_algorithm: "{self.deduplication_algorithm}"') return [] @@ -1030,7 +1174,15 @@ def process_finding_that_was_not_matched( self, unsaved_finding: Finding, ) -> tuple[Finding, bool]: - """Create a new finding from the one parsed from the report""" + """ + Prepare a new finding from the one parsed from the report. Not persisted here: + the caller queues it and persist_new_findings() writes it (batched, via + _drain_pending_new_findings()) once this matching batch finishes, so a + same-report duplicate later in the batch can still be added to this batch's + candidates via add_new_finding_to_candidates -- see that method and + match_finding_to_candidate_reimport for why the candidate lists tolerate an + unsaved finding. + """ # Set some explicit settings unsaved_finding.reporter = self.user unsaved_finding.last_reviewed = self.now @@ -1045,48 +1197,61 @@ def process_finding_that_was_not_matched( if self.scan_date_override: unsaved_finding.date = self.scan_date.date() 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. - 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 logger.debug( - "Reimport created new finding as no existing finding match: " - f"{finding.id}: {finding.title} " - f"({finding.component_name} - {finding.component_version})", + "Reimport found no existing match; will create a new finding: " + f"{finding.title} ({finding.component_name} - {finding.component_version})", ) # Manage the finding grouping selection finding_will_be_grouped = self.process_finding_groups( unsaved_finding, self.group_names_to_findings_dict, ) - # Add the new finding to the list - self.new_items.append(unsaved_finding) # Process any request/response pairs 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. + Write a single finding the report did not match to an existing one. + + Called once per finding, by this class's persist_new_findings() default below -- + not directly by process_finding_that_was_not_matched() any more, now that new + findings are queued (see _pending_new_findings) and persist_new_findings() writes + the whole matching batch at once via _drain_pending_new_findings(). 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. + get_reimport_match_candidates_for_batch) so a downstream edition can defer just + the write for a single finding -- accumulate here, and drain the buffer in + _flush_post_processing_batch (before calling super(), so the rows exist by the + time anything in that block reads a primary key) -- without reimplementing + persist_new_findings() or copying process_finding_that_was_not_matched(). An + edition that instead wants one bulk multi-row write for the whole batch overrides + persist_new_findings() directly. + + Overriding this is a supported way to defer the write. Everything between the + original call site and here -- grouping, the new_items list, request/response + pairs -- is safe on an unwritten finding, and that work is deliberately kept + that way. """ finding.save_no_options() + def persist_new_findings(self, prepared_findings: list[Finding]) -> list[Finding]: + """ + Persist this matching batch's new findings, one at a time via persist_new_finding(). + + Overrides BaseImporter's per-instance save_no_options() loop so a downstream + edition wanting only per-finding deferral (see persist_new_finding()) does not + also have to override this method: the default here calls into it, so overriding + persist_new_finding() alone is enough. An edition wanting a genuine bulk write + instead overrides this method and ignores persist_new_finding() entirely. + """ + for finding in prepared_findings: + self.persist_new_finding(finding) + return prepared_findings + def reconcile_vulnerability_ids( self, finding: Finding, diff --git a/unittests/test_tag_inheritance_perf.py b/unittests/test_tag_inheritance_perf.py index fa4f5f9429..69076441b6 100644 --- a/unittests/test_tag_inheritance_perf.py +++ b/unittests/test_tag_inheritance_perf.py @@ -635,9 +635,15 @@ def test_baseline_zap_scan_reimport_with_new_findings_v3(self): # V2 is unaffected -- EndpointManager.persist() opens no transaction -- which is # why only the V3 reimport constants absorb it (the import path buffers locations, # so its persist() still opens the transaction and keeps the +2 in full). + # -1 on both reimport-with-new paths, on top of everything above: DefaultReImporter + # now queues a matching batch's new findings in _pending_new_findings and persists/ + # post-processes them together via _drain_pending_new_findings once the batch's + # matching loop finishes, instead of saving each one inline as soon as it fails to + # match (see process_finding_that_was_not_matched and _drain_pending_new_findings). + # Reimport-no-change is unaffected because it creates no new findings to defer. EXPECTED_ZAP_IMPORT_V2 = 293 EXPECTED_ZAP_IMPORT_V3 = 318 EXPECTED_ZAP_REIMPORT_NO_CHANGE_V2 = 74 EXPECTED_ZAP_REIMPORT_NO_CHANGE_V3 = 85 - EXPECTED_ZAP_REIMPORT_WITH_NEW_V2 = 158 - EXPECTED_ZAP_REIMPORT_WITH_NEW_V3 = 186 + EXPECTED_ZAP_REIMPORT_WITH_NEW_V2 = 157 + EXPECTED_ZAP_REIMPORT_WITH_NEW_V3 = 185