diff --git a/dojo/finding/models.py b/dojo/finding/models.py index 0335d3b0395..9ef698631ff 100644 --- a/dojo/finding/models.py +++ b/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)}) diff --git a/dojo/importers/base_importer.py b/dojo/importers/base_importer.py index e647e1c8947..4e3d408e4cb 100644 --- a/dojo/importers/base_importer.py +++ b/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: diff --git a/dojo/importers/default_reimporter.py b/dojo/importers/default_reimporter.py index d18dfab1dfd..ff8a0c6af50 100644 --- a/dojo/importers/default_reimporter.py +++ b/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 diff --git a/dojo/importers/location_manager.py b/dojo/importers/location_manager.py index a2e08e5272e..525cce54ac3 100644 --- a/dojo/importers/location_manager.py +++ b/dojo/importers/location_manager.py @@ -38,8 +38,22 @@ class LocationManager(BaseLocationManager): def __init__(self, product: Product) -> None: super().__init__(product) - # Maps findings to a list of cleaned locations - self._locations_by_finding: dict[Finding, list[AbstractLocation]] = {} + # Findings paired with their cleaned locations, in first-recorded order. + # + # A list of pairs rather than a dict keyed by the finding, because a Finding that has not + # been written yet has no primary key and Django's Model.__hash__ raises on it + # ("Model instances without primary key value are unhashable"). Keying by the object would + # therefore forbid recording locations for a finding before it is saved -- and with it any + # importer that buffers inserts to write them in bulk at a batch boundary. Nothing here + # needs the object to be hashable: both consumers in _persist_locations() simply iterate, + # and they run inside persist(), by which point the findings are written. The tag + # accumulator threaded through finding_post_processing() already uses this same shape. + self._locations_by_finding: list[tuple[Finding, list[AbstractLocation]]] = [] + # Slot index into the list above so repeated records for one finding coalesce, keyed by + # object identity rather than by the finding itself for the reason above. id() is safe + # here specifically because the list holds a strong reference to every finding recorded, + # so no entry can be collected and have its id reused while this accumulator is live. + self._location_slot_by_finding: dict[int, int] = {} # All locations needing product refs (finding-associated + product-only), cleaned at record time self._product_locations: list[AbstractLocation] = [] # Status update entries, which we'll use at persist-time to determine Location statuses by comparing @@ -61,8 +75,21 @@ def record_locations_for_finding( ) -> None: """Record locations to be associated with a finding (and its product). Flushed by persist().""" if locations: - cleaned = self.clean_unsaved_locations(locations) - self._locations_by_finding.setdefault(finding, []).extend(cleaned) + self._record_cleaned_locations(finding, self.clean_unsaved_locations(locations)) + + def _record_cleaned_locations( + self, + finding: Finding, + cleaned: list[AbstractLocation], + ) -> None: + """Record locations that have already been through clean_unsaved_locations().""" + if cleaned: + slot = self._location_slot_by_finding.get(id(finding)) + if slot is None: + self._location_slot_by_finding[id(finding)] = len(self._locations_by_finding) + self._locations_by_finding.append((finding, list(cleaned))) + else: + self._locations_by_finding[slot][1].extend(cleaned) self._product_locations.extend(cleaned) def update_location_status( @@ -132,8 +159,7 @@ def _persist_locations(self) -> None: # full set — _product_locations is the superset of all locations (finding-associated + product-only). all_locations = list({(type(loc), loc.identity_hash): loc for loc in self._product_locations}.values()) if not all_locations: - self._locations_by_finding.clear() - self._product_locations.clear() + self._clear_location_accumulators() return # Bulk persist all locations to the database @@ -172,7 +198,7 @@ def _persist_locations(self) -> None: # Determine necessary finding refs to create if self._locations_by_finding: - all_finding_ids = [finding.id for finding in self._locations_by_finding] + all_finding_ids = [finding.id for finding, _ in self._locations_by_finding] # Strictly speaking this returns more rows than we need (it's the cross of the location/finding lists rather # than scoped per-finding), but more straightforward than constructing a per-finding lookup. We won't create # any unwanted associations below anyway. @@ -183,7 +209,7 @@ def _persist_locations(self) -> None: ).values_list("finding_id", "location_id"), ) - for finding, cleaned_locations in self._locations_by_finding.items(): + for finding, cleaned_locations in self._locations_by_finding: # Locations were already cleaned at record time — identity_hash is set, so we can # look up the persisted location directly. for location in cleaned_locations: @@ -224,7 +250,17 @@ def _persist_locations(self) -> None: ) # Clear accumulators + self._clear_location_accumulators() + + def _clear_location_accumulators(self) -> None: + """ + Reset the location accumulators together. + + The slot index is derived from the pair list, so clearing one without the other would + leave stale slots pointing past the end of the list. + """ self._locations_by_finding.clear() + self._location_slot_by_finding.clear() self._product_locations.clear() def _persist_status_updates(self) -> None: diff --git a/unittests/test_bulk_locations.py b/unittests/test_bulk_locations.py index 1e354645b66..2917aa2fe5d 100644 --- a/unittests/test_bulk_locations.py +++ b/unittests/test_bulk_locations.py @@ -539,3 +539,97 @@ def test_product_ref_reactivated_when_finding_ref_reactivated(self): product_ref.refresh_from_db() self.assertEqual(product_ref.status, ProductLocationStatus.Active) + + +@skip_unless_v3 +class TestRecordBeforeFindingIsSaved(DojoTestCase): + + """ + Locations must be recordable for a finding that has not been written yet. + + An importer that buffers inserts and writes them in bulk at a batch boundary records a + finding's locations while it is still unsaved. That used to be impossible for a reason + unrelated to locations: the accumulator was a dict keyed by the finding, and Django's + Model.__hash__ raises on an instance with no primary key. Nothing about persisting + locations needs the finding hashable -- persist() runs after the findings are written -- + so the accumulator holds pairs instead. + """ + + def _unsaved_finding_on_a_real_product(self): + """A Finding with a real test/product but deliberately never saved.""" + saved = _make_finding() + return Finding(test=saved.test, title="Unsaved", severity="Medium", reporter=saved.reporter), saved + + def test_recording_for_an_unsaved_finding_does_not_raise(self): + unsaved, saved = self._unsaved_finding_on_a_real_product() + mgr = LocationManager(saved.test.engagement.product) + + self.assertIsNone(unsaved.pk, "premise: the finding under test must be unsaved") + mgr.record_locations_for_finding(unsaved, [_make_url("unsaved-record.example.com")]) + + self.assertEqual(len(mgr._locations_by_finding), 1) + recorded_finding, recorded_locations = mgr._locations_by_finding[0] + self.assertIs(recorded_finding, unsaved) + self.assertEqual(len(recorded_locations), 1) + + def test_repeated_records_for_one_unsaved_finding_coalesce(self): + """record_for_finding() records twice (unsaved_locations, then extras) for one finding.""" + unsaved, saved = self._unsaved_finding_on_a_real_product() + mgr = LocationManager(saved.test.engagement.product) + + mgr.record_locations_for_finding(unsaved, [_make_url("coalesce-a.example.com")]) + mgr.record_locations_for_finding(unsaved, [_make_url("coalesce-b.example.com")]) + + self.assertEqual(len(mgr._locations_by_finding), 1, "one finding must occupy one slot") + self.assertEqual(len(mgr._locations_by_finding[0][1]), 2) + + def test_two_distinct_unsaved_findings_do_not_collide(self): + """ + Identity keying, not equality keying. + + Two unsaved findings are `==` to each other under Django's Model.__eq__ (both have + pk None), so an equality-keyed accumulator would merge their locations. Keying on + object identity keeps them apart. + """ + first, saved = self._unsaved_finding_on_a_real_product() + second = Finding(test=saved.test, title="Unsaved two", severity="Medium", reporter=saved.reporter) + mgr = LocationManager(saved.test.engagement.product) + + mgr.record_locations_for_finding(first, [_make_url("distinct-a.example.com")]) + mgr.record_locations_for_finding(second, [_make_url("distinct-b.example.com")]) + + self.assertEqual(len(mgr._locations_by_finding), 2) + self.assertEqual([entry[0] for entry in mgr._locations_by_finding], [first, second]) + + def test_locations_recorded_before_the_write_persist_after_it(self): + """The end-to-end shape a batched writer needs: record unsaved, write, then persist.""" + unsaved, saved = self._unsaved_finding_on_a_real_product() + product = saved.test.engagement.product + mgr = LocationManager(product) + + mgr.record_locations_for_finding(unsaved, [_make_url("deferred-write.example.com", "/api")]) + # The buffered write happens here, exactly as a batched writer would flush it. + unsaved.save() + mgr.persist() + + refs = LocationFindingReference.objects.filter(finding=unsaved) + self.assertEqual(refs.count(), 1) + self.assertEqual(refs.first().location.url.host, "deferred-write.example.com") + + def test_accumulators_are_cleared_together(self): + """A stale slot index would point past the end of the emptied pair list.""" + unsaved, saved = self._unsaved_finding_on_a_real_product() + mgr = LocationManager(saved.test.engagement.product) + + mgr.record_locations_for_finding(unsaved, [_make_url("cleared.example.com")]) + unsaved.save() + mgr.persist() + + self.assertEqual(mgr._locations_by_finding, []) + self.assertEqual(mgr._location_slot_by_finding, {}) + + # A second cycle on the same manager must start from a clean slate, not reuse a slot. + again = Finding(test=saved.test, title="Second cycle", severity="Medium", reporter=saved.reporter) + mgr.record_locations_for_finding(again, [_make_url("second-cycle.example.com")]) + self.assertEqual(len(mgr._locations_by_finding), 1) + self.assertIs(mgr._locations_by_finding[0][0], again) diff --git a/unittests/test_importers_deleted_finding_child_rows.py b/unittests/test_importers_deleted_finding_child_rows.py index 9341df14773..c0ac1fc8baf 100644 --- a/unittests/test_importers_deleted_finding_child_rows.py +++ b/unittests/test_importers_deleted_finding_child_rows.py @@ -252,3 +252,139 @@ def test_flush_burp_request_response_writes_everything_when_nothing_was_deleted( self._buffered_request_response_count(finding), msg=f"finding {finding.pk} lost its request/response pair", ) + + +class TestReconcileBeforeFindingIsWritten(DojoTestCase): + + """ + A finding's child rows must be reconcilable before the finding itself is written. + + Vulnerability ids and CWEs are already buffered and flushed at the batch boundary rather + than written per finding. Reconciling them, though, read the finding's existing rows back + through a reverse relation, and Django refuses that on an instance with no primary key + ("instance needs to have a primary key value before this relationship can be used"). So an + importer that buffers the finding inserts themselves -- writing them in bulk at the same + batch boundary -- could not reconcile at all. + + For a finding with no row there is nothing to read: no persisted vulnerability ids, no + persisted CWEs, and nothing to delete. Treating the existing set as empty is exact, not a + workaround. Every path here already runs today with a saved finding; these tests cover the + case that previously raised. + """ + + def setUp(self): + super().setUp() + self.user, _ = User.objects.get_or_create(username="admin") + product_type, _ = Product_Type.objects.get_or_create(name="reconcile_before_write") + self.environment, _ = Development_Environment.objects.get_or_create(name="Development") + self.product, _ = Product.objects.get_or_create( + name="TestReconcileBeforeFindingIsWritten", + description="Test", + prod_type=product_type, + ) + self.engagement, _ = Engagement.objects.get_or_create( + name="Reconcile Before Write", + product=self.product, + target_start=timezone.now(), + target_end=timezone.now(), + ) + with (get_unit_tests_scans_path("acunetix") / SCAN_FILE).open(encoding="utf-8") as scan: + self.test, _, _, _, _, _, _ = DefaultImporter( + close_old_findings=False, + user=self.user, + lead=self.user, + scan_date=None, + environment=self.environment, + active=True, + verified=False, + scan_type=SCAN_TYPE, + engagement=self.engagement, + ).process_scan(scan) + + def _reimporter(self): + return DefaultReImporter( + close_old_findings=False, + user=self.user, + lead=self.user, + scan_date=None, + environment=self.environment, + active=True, + verified=False, + scan_type=SCAN_TYPE, + test=self.test, + ) + + def _unsaved_finding(self, title="Buffered finding"): + return Finding(test=self.test, title=title, severity="Medium", reporter=self.user) + + def test_reconcile_vulnerability_ids_accepts_an_unsaved_finding(self): + reimporter = self._reimporter() + finding = self._unsaved_finding() + finding.unsaved_vulnerability_ids = ["CVE-2026-0001"] + + self.assertIsNone(finding.pk, "premise: the finding under test must be unsaved") + reimporter.reconcile_vulnerability_ids(finding) + + self.assertEqual("CVE-2026-0001", finding.cve) + self.assertIsNone(finding.pk, "reconciling must not write the finding") + + def test_buffered_vulnerability_ids_are_written_once_the_finding_is(self): + """The sequence a batched writer uses: reconcile while unsaved, write, then flush.""" + reimporter = self._reimporter() + finding = self._unsaved_finding("Buffered then written") + finding.unsaved_vulnerability_ids = ["CVE-2026-0002"] + + reimporter.reconcile_vulnerability_ids(finding) + # The buffered insert happens here, exactly as a batched writer would flush it. + finding.save() + reimporter.flush_vulnerability_ids() + + self.assertEqual( + ["CVE-2026-0002"], + [ + ref.vulnerability.vulnerability_id + for ref in FindingVulnerabilityReference.objects.filter( + finding_id=finding.pk, + ).select_related("vulnerability").order_by("order") + ], + ) + + def test_reconcile_cwes_accepts_an_unsaved_finding_and_queues_no_delete(self): + reimporter = self._reimporter() + finding = self._unsaved_finding("Buffered with cwes") + finding.cwe = 79 + + self.assertIsNone(finding.pk, "premise: the finding under test must be unsaved") + reimporter.reconcile_cwes(finding) + + # Stored as canonical CWE- labels, not the raw Finding.cwe integer. + self.assertEqual({"CWE-79"}, {row.cwe for row in reimporter.pending_cwes}) + self.assertEqual( + [], + reimporter.pending_cwe_deletes, + msg="a finding with no row has no CWE rows to delete, so nothing may be queued", + ) + + def test_a_changed_cve_does_not_write_an_unsaved_finding(self): + """ + finding_post_processing() saves on a cve change to update an existing row. + + A buffered finding has no row to update, and saving here would insert it early -- + defeating the buffering by splitting one batched INSERT into per-finding ones. + """ + reimporter = self._reimporter() + from_report = self._unsaved_finding("Report side") + from_report.unsaved_vulnerability_ids = ["CVE-2026-0003"] + finding = self._unsaved_finding("Buffered cve change") + + finding_count_before = Finding.objects.count() + reimporter.finding_post_processing( + finding, + from_report, + is_matched_finding=False, + tag_accumulator=[], + ) + + self.assertEqual("CVE-2026-0003", finding.cve, "premise: the cve must actually have changed") + self.assertIsNone(finding.pk, "the buffered finding must not have been written") + self.assertEqual(finding_count_before, Finding.objects.count()) diff --git a/unittests/test_reimport_final_drain.py b/unittests/test_reimport_final_drain.py new file mode 100644 index 00000000000..03e2275ecbf --- /dev/null +++ b/unittests/test_reimport_final_drain.py @@ -0,0 +1,115 @@ +""" +Regression test: a reimport must flush its last partial batch however the loop ended. + +process_findings() accumulates per-batch work (location status updates, vulnerability +ids, burp request/response pairs, parser and inherited tags, and the ids to dispatch to +post_process_findings_batch) and flushes it when the batch fills up. + +The matched branch ends in `continue`. That skip is what makes the tail fragile: with the +flush living inside the loop and gated on an is_final flag, a report whose LAST finding +took the matched branch never reached it, and everything appended since the previous +size-triggered flush was dropped -- silently, because nothing raises. Those findings got +no deduplication, no rules, no issue updater, no JIRA dispatch, and none of their parser +or inherited tags. + +The fix drains once after the loop instead, which runs however the last iteration ended. +""" + +from unittest.mock import patch + +from crum import impersonate +from django.utils import timezone + +from dojo.importers.default_reimporter import DefaultReImporter +from dojo.models import ( + Development_Environment, + Dojo_User, + Engagement, + Finding, + Product, + Product_Type, + Test, + Test_Type, + User, + UserContactInfo, +) + +from .dojo_test_case import DojoTestCase, get_unit_tests_scans_path + +SCAN_TYPE = "StackHawk HawkScan" +SCAN = get_unit_tests_scans_path("stackhawk") / "stackhawk_two_vul_same_hashcode_fabricated.json" + + +class TestReimportFinalDrain(DojoTestCase): + + """The last partial batch has to be flushed even when the final finding is a match.""" + + def setUp(self): + super().setUp() + testuser, _ = User.objects.get_or_create(username="admin") + UserContactInfo.objects.get_or_create(user=testuser, defaults={"block_execution": True}) + self.system_settings(enable_deduplication=True) + self.system_settings(enable_product_grade=False) + + product_type, _ = Product_Type.objects.get_or_create(name="test") + product, _ = Product.objects.get_or_create( + name="ReimportFinalDrainTest", description="Test", prod_type=product_type, + ) + engagement, _ = Engagement.objects.get_or_create( + name="Test Engagement", product=product, + target_start=timezone.now(), target_end=timezone.now(), + ) + self.lead, _ = User.objects.get_or_create(username="admin") + environment, _ = Development_Environment.objects.get_or_create(name="Development") + test_type, _ = Test_Type.objects.get_or_create(name=SCAN_TYPE) + self.test = Test.objects.create( + engagement=engagement, test_type=test_type, scan_type=SCAN_TYPE, + target_start=timezone.now(), target_end=timezone.now(), environment=environment, + ) + + def _reimport(self): + with impersonate(Dojo_User.objects.get(username="admin")), SCAN.open(encoding="utf-8") as scan: + reimporter = DefaultReImporter( + test=self.test, user=self.lead, lead=self.lead, scan_date=None, + minimum_severity="Info", active=True, verified=True, + force_sync=True, scan_type=SCAN_TYPE, + ) + return reimporter.process_scan(scan) + + def test_last_finding_taking_the_matched_branch_still_flushes_the_batch(self): + """ + Every finding in the report must reach post-processing dispatch. + + force_continue is forced on for every match so the loop's final iteration is + guaranteed to take the skipping branch. That is the precondition the bug needed; + forcing it here keeps the test from depending on which finding happens to sort + last, which is a property of the corpus rather than of the code under test. + """ + self._reimport() + self.assertGreater(Finding.objects.filter(test=self.test).count(), 0, + "the first pass must create findings, or the reimport matches nothing") + + real_process_matched = DefaultReImporter.process_matched_finding + + def always_force_continue(importer, unsaved_finding, existing_finding, *args, **kwargs): + finding, _ = real_process_matched(importer, unsaved_finding, existing_finding, *args, **kwargs) + return finding, True + + # bulk_apply_parser_tags is called from exactly one place, inside the flush, so it + # stands in for "the batch was flushed" without naming the flush itself. That keeps + # the test meaningful against the unfixed code, where no such method exists to patch. + with ( + patch.object(DefaultReImporter, "process_matched_finding", always_force_continue), + patch("dojo.importers.default_reimporter.bulk_apply_parser_tags") as apply_tags, + ): + self._reimport() + + self.assertTrue( + apply_tags.called, + msg=( + "the batch was never flushed: the last finding took the matched branch's " + "continue, so everything accumulated since the previous flush was dropped -- " + "no deduplication, rules, issue updater or JIRA dispatch, and no parser or " + "inherited tags for those findings" + ), + ) diff --git a/unittests/test_reimporter_persist_seam.py b/unittests/test_reimporter_persist_seam.py new file mode 100644 index 00000000000..8048eca1047 --- /dev/null +++ b/unittests/test_reimporter_persist_seam.py @@ -0,0 +1,161 @@ +from django.utils import timezone + +from dojo.importers.default_importer import DefaultImporter +from dojo.importers.default_reimporter import DefaultReImporter +from dojo.models import Development_Environment, Engagement, Finding, Product, Product_Type, User + +from .dojo_test_case import DojoTestCase, get_unit_tests_scans_path + +SCAN_TYPE = "Acunetix Scan" +SCAN_FILE = "many_findings.xml" + + +class BufferingReImporter(DefaultReImporter): + + """ + A downstream edition that defers new-finding writes to the batch boundary. + + This is the shape persist_new_finding() exists for, written out in full so the seam is + tested by a real user of it rather than by a mock. It buffers instead of writing, and + flushes the buffer at the start of the batch flush -- before anything in that block + (locations, vulnerability ids, tags, post-processing dispatch) reads a primary key. + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.buffered: list[Finding] = [] + self.flush_calls = 0 + self.max_buffer_seen = 0 + + def persist_new_finding(self, finding: Finding) -> None: + self.buffered.append(finding) + self.max_buffer_seen = max(self.max_buffer_seen, len(self.buffered)) + + def _flush_post_processing_batch(self, *args, **kwargs) -> None: + self.flush_calls += 1 + for finding in self.buffered: + finding.save_no_options() + self.buffered.clear() + super()._flush_post_processing_batch(*args, **kwargs) + + +class TestNewFindingPersistSeam(DojoTestCase): + + """ + persist_new_finding() must let a downstream edition defer the write. + + The importer writes each unmatched finding as it is processed. An edition that wants to + write them in bulk cannot do that without a seam: overriding + process_finding_that_was_not_matched() means copying its whole body, which then drifts + from upstream. This pins that overriding the one call is sufficient, and that the caller's + remaining work stays safe on a finding that has not been written yet. + """ + + def setUp(self): + super().setUp() + self.user, _ = User.objects.get_or_create(username="admin") + product_type, _ = Product_Type.objects.get_or_create(name="persist_seam") + self.environment, _ = Development_Environment.objects.get_or_create(name="Development") + self.product, _ = Product.objects.get_or_create( + name="TestNewFindingPersistSeam", + description="Test", + prod_type=product_type, + ) + self.engagement, _ = Engagement.objects.get_or_create( + name="Persist Seam", + product=self.product, + target_start=timezone.now(), + target_end=timezone.now(), + ) + + def _options(self, **overrides): + options = { + "user": self.user, + "lead": self.user, + "scan_date": None, + "environment": self.environment, + "active": True, + "verified": False, + "scan_type": SCAN_TYPE, + } + options.update(overrides) + return options + + def _empty_test(self, name): + """A Test with no findings, created by importing into a fresh engagement.""" + engagement = Engagement.objects.create( + name=name, + product=self.product, + target_start=timezone.now(), + target_end=timezone.now(), + ) + with (get_unit_tests_scans_path("acunetix") / "one_finding.xml").open(encoding="utf-8") as scan: + test, _, _, _, _, _, _ = DefaultImporter( + close_old_findings=False, **self._options(engagement=engagement), + ).process_scan(scan) + Finding.objects.filter(test=test).delete() + return test + + def _reimport(self, test, importer_class): + with (get_unit_tests_scans_path("acunetix") / SCAN_FILE).open(encoding="utf-8") as scan: + importer = importer_class(close_old_findings=False, **self._options(test=test)) + importer.process_scan(scan) + return importer + + @staticmethod + def _comparable(test): + """Findings as comparable tuples -- ids differ between runs, so they are excluded.""" + return sorted( + (f.title, f.severity, f.cve, f.component_name, f.component_version, f.active, f.verified) + for f in Finding.objects.filter(test=test) + ) + + def test_the_default_still_writes_each_finding_as_it_is_processed(self): + """The seam must not change stock behaviour.""" + test = self._empty_test("seam-default") + self._reimport(test, DefaultReImporter) + self.assertGreater(Finding.objects.filter(test=test).count(), 0) + + def test_an_override_can_defer_the_write_to_the_batch_boundary(self): + test = self._empty_test("seam-buffered") + + importer = self._reimport(test, BufferingReImporter) + + self.assertGreater( + importer.max_buffer_seen, + 0, + msg="premise: the override must actually have buffered something, or this proves nothing", + ) + self.assertEqual(importer.buffered, [], "the buffer must be drained by the final flush") + self.assertGreater(Finding.objects.filter(test=test).count(), 0) + + def test_deferring_the_write_produces_the_same_findings(self): + """The point of the seam: same result, different write timing.""" + stock_test = self._empty_test("seam-equiv-stock") + buffered_test = self._empty_test("seam-equiv-buffered") + + self._reimport(stock_test, DefaultReImporter) + self._reimport(buffered_test, BufferingReImporter) + + stock = self._comparable(stock_test) + buffered = self._comparable(buffered_test) + self.assertEqual(len(stock), len(buffered)) + self.assertEqual(stock, buffered) + + def test_child_rows_still_land_for_a_deferred_finding(self): + """ + The flush block writes locations, vulnerability ids and tags off the finding. + + Those all need a primary key, which is why the buffer drains at the *start* of the + flush. If that ordering broke, the findings would exist with no vulnerability ids. + """ + stock_test = self._empty_test("seam-children-stock") + buffered_test = self._empty_test("seam-children-buffered") + + self._reimport(stock_test, DefaultReImporter) + self._reimport(buffered_test, BufferingReImporter) + + def cve_set(test): + return {f.cve for f in Finding.objects.filter(test=test) if f.cve} + + self.assertEqual(cve_set(stock_test), cve_set(buffered_test))