From cf433128edaa389a56103f18c27c651a1029c6d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Fri, 24 Jul 2026 10:18:18 +0200 Subject: [PATCH] fix(api): duplicate finding rows in outputs on tasks re-run (#12097) --- .../output-generation-duplicate-rows.fixed.md | 1 + api/src/backend/tasks/tasks.py | 24 +++- api/src/backend/tasks/tests/test_tasks.py | 118 ++++++++++++++++++ 3 files changed, 142 insertions(+), 1 deletion(-) create mode 100644 api/changelog.d/output-generation-duplicate-rows.fixed.md diff --git a/api/changelog.d/output-generation-duplicate-rows.fixed.md b/api/changelog.d/output-generation-duplicate-rows.fixed.md new file mode 100644 index 0000000000..a503a56a03 --- /dev/null +++ b/api/changelog.d/output-generation-duplicate-rows.fixed.md @@ -0,0 +1 @@ +Output generation now removes the scan's temporary output directory before writing, so a re-run of the task for the same scan (e.g. broker redelivery after a worker is killed mid-run) no longer appends to the previous run's files and duplicates finding rows in the exported CSV and other outputs diff --git a/api/src/backend/tasks/tasks.py b/api/src/backend/tasks/tasks.py index a91ff85c01..e9101ff1bb 100644 --- a/api/src/backend/tasks/tasks.py +++ b/api/src/backend/tasks/tasks.py @@ -797,12 +797,34 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): if name not in frameworks_bulk and universal_bulk[name].outputs } frameworks_avail = get_compliance_frameworks(provider_type) + # Idempotency: a previous run of this task for the same scan may have left + # output files behind (e.g. broker redelivery after a worker was killed + # mid-run with task_acks_late, or a successful run on a deployment without + # S3 where the tmp dir is not removed). Output writers open files in append + # mode with a deterministic path (derived from scan.started_at), so reusing + # them would append every finding row again and duplicate the CSV/output + # rows. Start from a clean slate before (re)generating. + scan_tmp_dir = _scan_tmp_output_directory(tenant_id, scan_id) + if os.path.exists(scan_tmp_dir): + rmtree(scan_tmp_dir, ignore_errors=True) + # The writers below open output files in append mode with deterministic + # paths (derived from scan.started_at). Any stale file that survives the + # cleanup would get every finding row appended again, which is the exact + # duplication this guards against. Continuing is therefore unsafe: abort + # so `ScanReportRLSTask.on_failure` removes the tmp dir and the retry + # starts from a clean slate instead of publishing duplicated rows. + if os.path.exists(scan_tmp_dir): + raise RuntimeError( + "Could not remove stale output directory for scan " + f"{scan_id} before generating outputs; aborting to avoid " + "duplicated rows in appended outputs." + ) + out_dir, comp_dir = _generate_output_directory( DJANGO_TMP_OUTPUT_DIRECTORY, provider_uid, tenant_id, scan_id ) # Removed on success here and on failure by ScanReportRLSTask.on_failure, # so partial artifacts do not accumulate and fill the disk (ENOSPC). - scan_tmp_dir = _scan_tmp_output_directory(tenant_id, scan_id) def get_writer(writer_map, name, factory, is_last): """ diff --git a/api/src/backend/tasks/tests/test_tasks.py b/api/src/backend/tasks/tests/test_tasks.py index 476444cb00..8c846be805 100644 --- a/api/src/backend/tasks/tests/test_tasks.py +++ b/api/src/backend/tasks/tests/test_tasks.py @@ -420,6 +420,124 @@ class TestGenerateOutputs: assert result == {"upload": False} mock_scan_update.return_value.update.assert_called_once() + def test_generate_outputs_removes_previous_run_artifacts(self): + """Regression for PROWLER-2266. + + Output writers open files in append mode with a deterministic path + (derived from scan.started_at). If this task runs again for the same + scan (e.g. broker redelivery after a worker is killed mid-run with + task_acks_late), reusing the leftover files appends every finding row + again, duplicating rows in the CSV/output while the API console keeps + showing a single finding. The task must start from a clean slate by + removing the scan's tmp output directory before (re)generating. + """ + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_root: + # Simulate artifacts left behind by a previous run of the same scan. + scan_tmp_dir = Path(tmp_root) / self.tenant_id / self.scan_id + scan_tmp_dir.mkdir(parents=True) + stale_artifact = scan_tmp_dir / "prowler-output-aws-20260723120000.csv" + stale_artifact.write_text("HEADER\nold-finding-row\n") + + with ( + patch("tasks.tasks.DJANGO_TMP_OUTPUT_DIRECTORY", tmp_root), + patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter, + patch("tasks.tasks.Provider.objects.get"), + patch("tasks.tasks.initialize_prowler_provider"), + patch("tasks.tasks.Compliance.get_bulk"), + patch("tasks.tasks.get_compliance_frameworks"), + patch("tasks.tasks.get_prowler_provider_compliance", return_value={}), + patch("tasks.tasks.Finding.all_objects.filter") as mock_findings, + patch( + "tasks.tasks._generate_output_directory", + return_value=("/tmp/test/out", "/tmp/test/comp"), + ), + patch("tasks.tasks.FindingOutput._transform_findings_stats"), + patch("tasks.tasks.FindingOutput.transform_api_finding"), + patch( + "tasks.tasks.OUTPUT_FORMATS_MAPPING", + { + "json": { + "class": MagicMock(name="Writer"), + "suffix": ".json", + "kwargs": {}, + } + }, + ), + patch("tasks.tasks.COMPLIANCE_CLASS_MAP", {"aws": []}), + patch( + "tasks.tasks._compress_output_files", return_value="/tmp/compressed" + ), + patch("tasks.tasks._upload_to_s3", return_value=None), + patch("tasks.tasks.Scan.all_objects.filter"), + ): + mock_filter.return_value.exists.return_value = True + mock_findings.return_value.order_by.return_value.iterator.return_value = [ + [MagicMock()], + True, + ] + + generate_outputs_task( + scan_id=self.scan_id, + provider_id=self.provider_id, + tenant_id=self.tenant_id, + ) + + # The stale artifacts from the previous run must be gone, so the + # append-mode writers cannot duplicate rows onto them. + assert not stale_artifact.exists() + assert not scan_tmp_dir.exists() + + def test_generate_outputs_aborts_when_stale_cleanup_fails(self): + """Regression for PROWLER-2266. + + If the stale output directory cannot be removed (e.g. permission error), + the leftover files would be reopened in append mode and every finding + row would be duplicated. The task must abort instead of continuing and + publishing duplicated rows, so the retry can start from a clean slate. + """ + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_root: + scan_tmp_dir = Path(tmp_root) / self.tenant_id / self.scan_id + scan_tmp_dir.mkdir(parents=True) + stale_artifact = scan_tmp_dir / "prowler-output-aws-20260723120000.csv" + stale_artifact.write_text("HEADER\nold-finding-row\n") + + with ( + patch("tasks.tasks.DJANGO_TMP_OUTPUT_DIRECTORY", tmp_root), + patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter, + patch("tasks.tasks.Provider.objects.get"), + patch("tasks.tasks.initialize_prowler_provider"), + patch("tasks.tasks.Compliance.get_bulk"), + patch("tasks.tasks.get_compliance_frameworks"), + patch("tasks.tasks.get_prowler_provider_compliance", return_value={}), + # `rmtree(ignore_errors=True)` swallows the failure and leaves the + # directory behind; simulate that with a no-op so the guard fires. + patch("tasks.tasks.rmtree"), + patch("tasks.tasks._generate_output_directory") as mock_gen_dir, + patch("tasks.tasks._compress_output_files") as mock_compress, + patch("tasks.tasks._upload_to_s3") as mock_upload, + patch("tasks.tasks.Scan.all_objects.filter") as mock_scan_update, + ): + mock_filter.return_value.exists.return_value = True + + with pytest.raises(RuntimeError, match="stale output directory"): + generate_outputs_task( + scan_id=self.scan_id, + provider_id=self.provider_id, + tenant_id=self.tenant_id, + ) + + # The task must abort before generating/publishing any output. + mock_gen_dir.assert_not_called() + mock_compress.assert_not_called() + mock_upload.assert_not_called() + mock_scan_update.assert_not_called() + def test_generate_outputs_triggers_html_extra_update(self): mock_finding_output = MagicMock() mock_finding_output.compliance = {"cis": ["requirement-1", "requirement-2"]}