From fb6da427f8f14fb7013e856ff8e18987085d9a4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adri=C3=A1n=20Pe=C3=B1a?= Date: Mon, 27 Apr 2026 11:05:34 +0200 Subject: [PATCH] fix(api): prevent /tmp saturation from compliance report generation (#10874) --- api/CHANGELOG.md | 4 + api/src/backend/tasks/jobs/report.py | 619 ++++++++++++++++---- api/src/backend/tasks/tasks.py | 19 +- api/src/backend/tasks/tests/test_reports.py | 534 ++++++++++++++++- api/src/backend/tasks/tests/test_tasks.py | 57 +- 5 files changed, 1101 insertions(+), 132 deletions(-) diff --git a/api/CHANGELOG.md b/api/CHANGELOG.md index a742d9ca61..24b43deb62 100644 --- a/api/CHANGELOG.md +++ b/api/CHANGELOG.md @@ -28,6 +28,10 @@ All notable changes to the **Prowler API** are documented in this file. - Attack Paths: Neo4j driver `connection_acquisition_timeout` is now configurable via `NEO4J_CONN_ACQUISITION_TIMEOUT` (default lowered from 120 s to 15 s) [(#10873)](https://github.com/prowler-cloud/prowler/pull/10873) +### 🐞 Fixed + +- `/tmp/prowler_api_output` saturation in compliance report workers: the final `rmtree` in `generate_compliance_reports` now only waits on frameworks actually generated for the provider (so unsupported frameworks no longer leave a placeholder `results` entry that blocks cleanup), output directories are created lazily per enabled framework, and both `generate_compliance_reports` and `generate_outputs_task` run an opportunistic stale cleanup at task start with a 48h age threshold, a per-host `fcntl` throttle, a 50-deletions-per-run cap, and guards that protect EXECUTING scans and scans whose `output_location` still points to a local path (metadata lookups routed through the admin DB so RLS does not hide those rows) [(#10874)](https://github.com/prowler-cloud/prowler/pull/10874) + --- ## [1.25.3] (Prowler v5.24.3) diff --git a/api/src/backend/tasks/jobs/report.py b/api/src/backend/tasks/jobs/report.py index 005df623f2..12f48b7040 100644 --- a/api/src/backend/tasks/jobs/report.py +++ b/api/src/backend/tasks/jobs/report.py @@ -1,9 +1,13 @@ import gc +import os import re +import time from collections.abc import Iterable from pathlib import Path from shutil import rmtree +from uuid import UUID +import fcntl from celery.utils.log import get_task_logger from config.django.base import DJANGO_TMP_OUTPUT_DIRECTORY from tasks.jobs.export import _generate_compliance_output_directory, _upload_to_s3 @@ -18,13 +22,38 @@ from tasks.jobs.reports import ( from tasks.jobs.threatscore import compute_threatscore_metrics from tasks.jobs.threatscore_utils import _aggregate_requirement_statistics_from_database -from api.db_router import READ_REPLICA_ALIAS +from api.db_router import READ_REPLICA_ALIAS, MainRouter from api.db_utils import rls_transaction -from api.models import Provider, ScanSummary, ThreatScoreSnapshot +from api.models import Provider, Scan, ScanSummary, StateChoices, ThreatScoreSnapshot from prowler.lib.check.compliance_models import Compliance from prowler.lib.outputs.finding import Finding as FindingOutput logger = get_task_logger(__name__) +STALE_TMP_OUTPUT_MAX_AGE_HOURS = 48 +STALE_TMP_OUTPUT_MAX_DELETIONS_PER_RUN = 50 +STALE_TMP_OUTPUT_THROTTLE_SECONDS = 60 * 60 +STALE_TMP_OUTPUT_LOCK_FILE_NAME = ".stale_tmp_cleanup.lock" + +# Refuse to ever run rmtree against shared system roots; the configured +# DJANGO_TMP_OUTPUT_DIRECTORY must be a dedicated subdirectory. +_FORBIDDEN_CLEANUP_ROOTS = frozenset( + Path(p).resolve() + for p in ("/", "/tmp", "/var", "/var/tmp", "/home", "/root", "/etc", "/usr") +) + + +def _resolve_stale_tmp_safe_root() -> Path | None: + """Resolve the configured tmp output directory, rejecting unsafe roots.""" + try: + configured_root = Path(DJANGO_TMP_OUTPUT_DIRECTORY).resolve() + except OSError: + return None + if configured_root in _FORBIDDEN_CLEANUP_ROOTS: + return None + return configured_root + + +STALE_TMP_OUTPUT_SAFE_ROOT = _resolve_stale_tmp_safe_root() # Matches CIS compliance_ids like "cis_1.4_aws", "cis_5.0_azure", # "cis_1.10_kubernetes", "cis_3.0.1_aws". Requires at least one dotted @@ -69,6 +98,324 @@ def _pick_latest_cis_variant(compliance_ids: Iterable[str]) -> str | None: return best_name +def _should_run_stale_cleanup( + root_path: Path, + throttle_seconds: int = STALE_TMP_OUTPUT_THROTTLE_SECONDS, +) -> bool: + """Throttle stale cleanup to at most once per hour per host.""" + lock_file_path = root_path / STALE_TMP_OUTPUT_LOCK_FILE_NAME + now_timestamp = int(time.time()) + + try: + with lock_file_path.open("a+", encoding="ascii") as lock_file: + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return False + lock_file.seek(0) + previous_value = lock_file.read().strip() + try: + last_run_timestamp = int(previous_value) if previous_value else 0 + except ValueError: + last_run_timestamp = 0 + + if now_timestamp - last_run_timestamp < throttle_seconds: + return False + + lock_file.seek(0) + lock_file.truncate() + lock_file.write(str(now_timestamp)) + lock_file.flush() + os.fsync(lock_file.fileno()) + except OSError as error: + logger.warning("Skipping stale tmp cleanup: lock file error (%s)", error) + return False + + return True + + +def _is_scan_metadata_protected( + scan_path: Path, + scan_state: str | None, + output_location: str | None, +) -> bool: + """ + Return True when metadata indicates the directory must not be deleted. + + Protected cases: + - Scan is still EXECUTING. + - Scan has a local output artifact path (non-S3) under this scan directory. + """ + if scan_state == StateChoices.EXECUTING.value: + return True + + output_location = output_location or "" + if output_location and not output_location.startswith("s3://"): + try: + resolved_output_location = Path(output_location).resolve() + except OSError: + # Conservative fallback: if we cannot resolve a local output path, + # keep the directory to avoid deleting potentially needed artifacts. + return True + + if ( + resolved_output_location == scan_path + or scan_path in resolved_output_location.parents + ): + return True + + return False + + +def _is_scan_directory_protected( + tenant_id: str, + scan_id: str, + scan_path: Path, +) -> bool: + """ + DB-backed wrapper used when batch metadata is not already available. + """ + try: + scan_uuid = UUID(scan_id) + except ValueError: + return False + + try: + scan = ( + Scan.all_objects.using(MainRouter.admin_db) + .filter(tenant_id=tenant_id, id=scan_uuid) + .only("state", "output_location") + .first() + ) + except Exception as error: + logger.warning( + "Skipping stale tmp cleanup for %s/%s due to scan lookup error: %s", + tenant_id, + scan_id, + error, + ) + return True + + if not scan: + return False + + return _is_scan_metadata_protected( + scan_path=scan_path, + scan_state=scan.state, + output_location=scan.output_location, + ) + + +def _cleanup_stale_tmp_output_directories( + tmp_output_root: str, + max_age_hours: int = STALE_TMP_OUTPUT_MAX_AGE_HOURS, + exclude_scan: tuple[str, str] | None = None, + max_deletions_per_run: int = STALE_TMP_OUTPUT_MAX_DELETIONS_PER_RUN, +) -> int: + """ + Opportunistically delete stale scan directories under the tmp output root. + + Expected directory layout: + ///... + + Each run that wins the per-host throttle sweeps every tenant directory so + leftover artifacts cannot pile up for tenants whose own tasks happen to + lose the throttle race. + + Args: + tmp_output_root: Base tmp output path. + max_age_hours: Directory max age before deletion. + exclude_scan: Optional (tenant_id, scan_id) that must never be deleted. + max_deletions_per_run: Max number of scan directories deleted per run. + + Returns: + Number of deleted scan directories. + """ + try: + if max_age_hours <= 0: + return 0 + + try: + root_path = Path(tmp_output_root).resolve() + except OSError as error: + logger.warning( + "Skipping stale tmp cleanup: unable to resolve %s (%s)", + tmp_output_root, + error, + ) + return 0 + + if ( + STALE_TMP_OUTPUT_SAFE_ROOT is None + or root_path != STALE_TMP_OUTPUT_SAFE_ROOT + ): + logger.warning( + "Skipping stale tmp cleanup: unsupported root %s (allowed: %s)", + root_path, + STALE_TMP_OUTPUT_SAFE_ROOT, + ) + return 0 + + if not root_path.exists() or not root_path.is_dir(): + return 0 + + if max_deletions_per_run <= 0: + return 0 + + if not _should_run_stale_cleanup(root_path): + return 0 + + cutoff_timestamp = time.time() - (max_age_hours * 60 * 60) + deleted_scan_dirs = 0 + + try: + tenant_dirs = list(root_path.iterdir()) + except OSError as error: + logger.warning( + "Skipping stale tmp cleanup: unable to list %s (%s)", + root_path, + error, + ) + return 0 + + for tenant_dir in tenant_dirs: + if deleted_scan_dirs >= max_deletions_per_run: + break + + if not tenant_dir.is_dir() or tenant_dir.is_symlink(): + continue + + try: + scan_dirs = list(tenant_dir.iterdir()) + except OSError: + continue + + stale_candidates: list[tuple[str, Path, UUID | None]] = [] + for scan_dir in scan_dirs: + if not scan_dir.is_dir() or scan_dir.is_symlink(): + continue + + if exclude_scan and ( + tenant_dir.name == exclude_scan[0] + and scan_dir.name == exclude_scan[1] + ): + continue + + try: + if scan_dir.stat().st_mtime >= cutoff_timestamp: + continue + except OSError: + continue + + try: + resolved_scan_dir = scan_dir.resolve() + except OSError: + continue + + if root_path not in resolved_scan_dir.parents: + logger.warning( + "Skipping stale tmp cleanup for path outside root: %s", + resolved_scan_dir, + ) + continue + + try: + scan_uuid: UUID | None = UUID(scan_dir.name) + except ValueError: + scan_uuid = None + + stale_candidates.append((scan_dir.name, resolved_scan_dir, scan_uuid)) + + if not stale_candidates: + continue + + scan_metadata_by_id: dict[UUID, tuple[str | None, str | None]] = {} + metadata_preload_succeeded = False + candidate_scan_ids = [ + candidate[2] for candidate in stale_candidates if candidate[2] + ] + if candidate_scan_ids: + try: + scan_rows = ( + Scan.all_objects.using(MainRouter.admin_db) + .filter( + tenant_id=tenant_dir.name, + id__in=candidate_scan_ids, + ) + .values_list("id", "state", "output_location") + ) + scan_metadata_by_id = { + scan_id: (scan_state, output_location) + for scan_id, scan_state, output_location in scan_rows + } + metadata_preload_succeeded = True + except Exception as error: + logger.warning( + "Skipping stale tmp cleanup metadata preload for tenant %s: %s", + tenant_dir.name, + error, + ) + else: + metadata_preload_succeeded = True + + for scan_name, resolved_scan_dir, scan_uuid in stale_candidates: + if deleted_scan_dirs >= max_deletions_per_run: + break + + should_check_scan_fallback = True + if scan_uuid and metadata_preload_succeeded: + should_check_scan_fallback = False + scan_metadata = scan_metadata_by_id.get(scan_uuid) + if scan_metadata: + scan_state, output_location = scan_metadata + if _is_scan_metadata_protected( + scan_path=resolved_scan_dir, + scan_state=scan_state, + output_location=output_location, + ): + continue + + if should_check_scan_fallback and _is_scan_directory_protected( + tenant_id=tenant_dir.name, + scan_id=scan_name, + scan_path=resolved_scan_dir, + ): + continue + + try: + rmtree(resolved_scan_dir, ignore_errors=True) + deleted_scan_dirs += 1 + except Exception as error: + logger.warning( + "Error cleaning stale tmp directory %s: %s", + resolved_scan_dir, + error, + ) + + if deleted_scan_dirs: + logger.info( + "Deleted %s stale tmp output directories older than %sh from %s", + deleted_scan_dirs, + max_age_hours, + root_path, + ) + if deleted_scan_dirs >= max_deletions_per_run: + logger.info( + "Stale tmp cleanup hit deletion limit (%s) for root %s", + max_deletions_per_run, + root_path, + ) + + return deleted_scan_dirs + except Exception as error: + logger.warning( + "Skipping stale tmp cleanup due to unexpected error: %s", + error, + exc_info=True, + ) + return 0 + + def generate_threatscore_report( tenant_id: str, scan_id: str, @@ -355,6 +702,19 @@ def generate_compliance_reports( generate_cis, ) + try: + _cleanup_stale_tmp_output_directories( + DJANGO_TMP_OUTPUT_DIRECTORY, + max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS, + exclude_scan=(tenant_id, scan_id), + ) + except Exception as error: + logger.warning( + "Skipping stale tmp cleanup before compliance reports for scan %s: %s", + scan_id, + error, + ) + results: dict = {} # Validate that the scan has findings and get provider info @@ -412,10 +772,31 @@ def generate_compliance_reports( generate_csa = False # For CIS we do NOT pre-check the provider against a hard-coded whitelist - # (that list drifts the moment a new CIS JSON ships). Instead, we let - # `_pick_latest_cis_variant` over `Compliance.get_bulk(provider_type)` - # return None for providers that lack CIS, and treat that as "nothing to - # do" below. + # (that list drifts the moment a new CIS JSON ships). Instead, we inspect + # the dynamically loaded framework map and pick the latest available CIS + # version, if any. + latest_cis: str | None = None + if generate_cis: + try: + frameworks_bulk = Compliance.get_bulk(provider_type) + latest_cis = _pick_latest_cis_variant( + name for name in frameworks_bulk.keys() if name.startswith("cis_") + ) + except Exception as e: + logger.error("Error discovering CIS variants for %s: %s", provider_type, e) + results["cis"] = {"upload": False, "path": "", "error": str(e)} + generate_cis = False + else: + if latest_cis is None: + logger.info("No CIS variants available for provider %s", provider_type) + results["cis"] = {"upload": False, "path": ""} + generate_cis = False + else: + logger.info( + "Selected latest CIS variant for provider %s: %s", + provider_type, + latest_cis, + ) if ( not generate_threatscore @@ -438,45 +819,56 @@ def generate_compliance_reports( findings_cache = {} logger.info("Created shared findings cache for all reports") - # Generate output directories + generated_report_keys: list[str] = [] + output_paths: dict[str, str] = {} + out_dir: str | None = None + + # Generate output directories only for enabled and supported report types. try: logger.info("Generating output directories") - threatscore_path = _generate_compliance_output_directory( - DJANGO_TMP_OUTPUT_DIRECTORY, - provider_uid, - tenant_id, - scan_id, - compliance_framework="threatscore", - ) - ens_path = _generate_compliance_output_directory( - DJANGO_TMP_OUTPUT_DIRECTORY, - provider_uid, - tenant_id, - scan_id, - compliance_framework="ens", - ) - nis2_path = _generate_compliance_output_directory( - DJANGO_TMP_OUTPUT_DIRECTORY, - provider_uid, - tenant_id, - scan_id, - compliance_framework="nis2", - ) - csa_path = _generate_compliance_output_directory( - DJANGO_TMP_OUTPUT_DIRECTORY, - provider_uid, - tenant_id, - scan_id, - compliance_framework="csa", - ) - cis_path = _generate_compliance_output_directory( - DJANGO_TMP_OUTPUT_DIRECTORY, - provider_uid, - tenant_id, - scan_id, - compliance_framework="cis", - ) - out_dir = str(Path(threatscore_path).parent.parent) + if generate_threatscore: + output_paths["threatscore"] = _generate_compliance_output_directory( + DJANGO_TMP_OUTPUT_DIRECTORY, + provider_uid, + tenant_id, + scan_id, + compliance_framework="threatscore", + ) + if generate_ens: + output_paths["ens"] = _generate_compliance_output_directory( + DJANGO_TMP_OUTPUT_DIRECTORY, + provider_uid, + tenant_id, + scan_id, + compliance_framework="ens", + ) + if generate_nis2: + output_paths["nis2"] = _generate_compliance_output_directory( + DJANGO_TMP_OUTPUT_DIRECTORY, + provider_uid, + tenant_id, + scan_id, + compliance_framework="nis2", + ) + if generate_csa: + output_paths["csa"] = _generate_compliance_output_directory( + DJANGO_TMP_OUTPUT_DIRECTORY, + provider_uid, + tenant_id, + scan_id, + compliance_framework="csa", + ) + if generate_cis and latest_cis: + output_paths["cis"] = _generate_compliance_output_directory( + DJANGO_TMP_OUTPUT_DIRECTORY, + provider_uid, + tenant_id, + scan_id, + compliance_framework="cis", + ) + if output_paths: + first_output_path = next(iter(output_paths.values())) + out_dir = str(Path(first_output_path).parent.parent) except Exception as e: logger.error("Error generating output directory: %s", e) error_dict = {"error": str(e), "upload": False, "path": ""} @@ -494,6 +886,8 @@ def generate_compliance_reports( # Generate ThreatScore report if generate_threatscore: + generated_report_keys.append("threatscore") + threatscore_path = output_paths["threatscore"] compliance_id_threatscore = f"prowler_threatscore_{provider_type}" pdf_path_threatscore = f"{threatscore_path}_threatscore_report.pdf" logger.info( @@ -595,6 +989,8 @@ def generate_compliance_reports( # Generate ENS report if generate_ens: + generated_report_keys.append("ens") + ens_path = output_paths["ens"] compliance_id_ens = f"ens_rd2022_{provider_type}" pdf_path_ens = f"{ens_path}_ens_report.pdf" logger.info("Generating ENS report with compliance %s", compliance_id_ens) @@ -629,6 +1025,8 @@ def generate_compliance_reports( # Generate NIS2 report if generate_nis2: + generated_report_keys.append("nis2") + nis2_path = output_paths["nis2"] compliance_id_nis2 = f"nis2_{provider_type}" pdf_path_nis2 = f"{nis2_path}_nis2_report.pdf" logger.info("Generating NIS2 report with compliance %s", compliance_id_nis2) @@ -664,6 +1062,8 @@ def generate_compliance_reports( # Generate CSA CCM report if generate_csa: + generated_report_keys.append("csa") + csa_path = output_paths["csa"] compliance_id_csa = f"csa_ccm_4.0_{provider_type}" pdf_path_csa = f"{csa_path}_csa_report.pdf" logger.info("Generating CSA CCM report with compliance %s", compliance_id_csa) @@ -700,91 +1100,72 @@ def generate_compliance_reports( # Generate CIS Benchmark report for the latest available version only. # CIS ships multiple versions per provider (e.g. cis_1.4_aws, cis_5.0_aws, # cis_6.0_aws); we dynamically pick the highest semantic version at run - # time rather than hard-coding a per-provider mapping. `Compliance.get_bulk` - # is the single source of truth for which providers have CIS. - if generate_cis: - latest_cis: str | None = None + # time rather than hard-coding a per-provider mapping. + if generate_cis and latest_cis: + generated_report_keys.append("cis") + cis_path = output_paths["cis"] + if out_dir is None: + out_dir = str(Path(cis_path).parent.parent) + pdf_path_cis = f"{cis_path}_cis_report.pdf" try: - frameworks_bulk = Compliance.get_bulk(provider_type) - latest_cis = _pick_latest_cis_variant( - name for name in frameworks_bulk.keys() if name.startswith("cis_") + generate_cis_report( + tenant_id=tenant_id, + scan_id=scan_id, + compliance_id=latest_cis, + output_path=pdf_path_cis, + provider_id=provider_id, + only_failed=only_failed_cis, + include_manual=include_manual_cis, + provider_obj=provider_obj, + requirement_statistics=requirement_statistics, + findings_cache=findings_cache, ) - except Exception as e: - logger.error("Error discovering CIS variants for %s: %s", provider_type, e) - results["cis"] = {"upload": False, "path": "", "error": str(e)} - if "cis" not in results: - if latest_cis is None: - logger.info("No CIS variants available for provider %s", provider_type) - results["cis"] = {"upload": False, "path": ""} - else: + upload_uri_cis = _upload_to_s3( + tenant_id, + scan_id, + pdf_path_cis, + f"cis/{Path(pdf_path_cis).name}", + ) + + if upload_uri_cis: + results["cis"] = { + "upload": True, + "path": upload_uri_cis, + } logger.info( - "Selected latest CIS variant for provider %s: %s", - provider_type, + "CIS report %s uploaded to %s", latest_cis, + upload_uri_cis, + ) + else: + results["cis"] = {"upload": False, "path": out_dir} + logger.warning( + "CIS report %s saved locally at %s", + latest_cis, + out_dir, ) - pdf_path_cis = f"{cis_path}_cis_report.pdf" - try: - generate_cis_report( - tenant_id=tenant_id, - scan_id=scan_id, - compliance_id=latest_cis, - output_path=pdf_path_cis, - provider_id=provider_id, - only_failed=only_failed_cis, - include_manual=include_manual_cis, - provider_obj=provider_obj, - requirement_statistics=requirement_statistics, - findings_cache=findings_cache, - ) - upload_uri_cis = _upload_to_s3( - tenant_id, - scan_id, - pdf_path_cis, - f"cis/{Path(pdf_path_cis).name}", - ) + except Exception as e: + logger.error("Error generating CIS report %s: %s", latest_cis, e) + results["cis"] = { + "upload": False, + "path": "", + "error": str(e), + } + finally: + # Free ReportLab/matplotlib memory before moving on. + gc.collect() - if upload_uri_cis: - results["cis"] = { - "upload": True, - "path": upload_uri_cis, - } - logger.info( - "CIS report %s uploaded to %s", - latest_cis, - upload_uri_cis, - ) - else: - results["cis"] = {"upload": False, "path": out_dir} - logger.warning( - "CIS report %s saved locally at %s", - latest_cis, - out_dir, - ) + # Clean up temporary files only if all generated reports were + # uploaded successfully. Reports skipped for provider incompatibility + # or missing CIS variants must not block cleanup. + all_uploaded = bool(generated_report_keys) and all( + results.get(report_key, {}).get("upload", False) + for report_key in generated_report_keys + ) - except Exception as e: - logger.error("Error generating CIS report %s: %s", latest_cis, e) - results["cis"] = { - "upload": False, - "path": "", - "error": str(e), - } - finally: - # Free ReportLab/matplotlib memory before moving on. - gc.collect() - - # Clean up temporary files only if every requested report has been - # successfully uploaded. All result entries now share the same flat - # shape, so the check is a single comprehension. - upload_flags = [ - bool(entry.get("upload", False)) - for entry in results.values() - if isinstance(entry, dict) and entry.get("upload") is not None - ] - all_uploaded = bool(upload_flags) and all(upload_flags) - - if all_uploaded: + if all_uploaded and out_dir: try: rmtree(Path(out_dir), ignore_errors=True) logger.info("Cleaned up temporary files at %s", out_dir) diff --git a/api/src/backend/tasks/tasks.py b/api/src/backend/tasks/tasks.py index df6c257949..6dfabec103 100644 --- a/api/src/backend/tasks/tasks.py +++ b/api/src/backend/tasks/tasks.py @@ -46,7 +46,11 @@ from tasks.jobs.lighthouse_providers import ( refresh_lighthouse_provider_models, ) from tasks.jobs.muting import mute_historical_findings -from tasks.jobs.report import generate_compliance_reports_job +from tasks.jobs.report import ( + STALE_TMP_OUTPUT_MAX_AGE_HOURS, + _cleanup_stale_tmp_output_directories, + generate_compliance_reports_job, +) from tasks.jobs.scan import ( aggregate_attack_surface, aggregate_daily_severity, @@ -440,6 +444,19 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): scan_id (str): The scan identifier. provider_id (str): The provider_id id to be used in generating outputs. """ + try: + _cleanup_stale_tmp_output_directories( + DJANGO_TMP_OUTPUT_DIRECTORY, + max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS, + exclude_scan=(tenant_id, scan_id), + ) + except Exception as error: + logger.warning( + "Skipping stale tmp cleanup before output generation for scan %s: %s", + scan_id, + error, + ) + # Check if the scan has findings if not ScanSummary.objects.filter(scan_id=scan_id).exists(): logger.info(f"No findings found for scan {scan_id}") diff --git a/api/src/backend/tasks/tests/test_reports.py b/api/src/backend/tasks/tests/test_reports.py index a25df876f3..b5cb0cc38e 100644 --- a/api/src/backend/tasks/tests/test_reports.py +++ b/api/src/backend/tasks/tests/test_reports.py @@ -1,3 +1,5 @@ +import os +import time import uuid from unittest.mock import Mock, patch @@ -5,7 +7,12 @@ import matplotlib import pytest from reportlab.lib import colors from tasks.jobs.report import ( + STALE_TMP_OUTPUT_MAX_AGE_HOURS, + STALE_TMP_OUTPUT_LOCK_FILE_NAME, + _cleanup_stale_tmp_output_directories, + _is_scan_directory_protected, _pick_latest_cis_variant, + _should_run_stale_cleanup, generate_compliance_reports, generate_threatscore_report, ) @@ -33,7 +40,13 @@ from tasks.jobs.threatscore_utils import ( _load_findings_for_requirement_checks, ) -from api.models import Finding, Resource, ResourceFindingMapping, StatusChoices +from api.models import ( + Finding, + Resource, + ResourceFindingMapping, + StateChoices, + StatusChoices, +) from prowler.lib.check.models import Severity matplotlib.use("Agg") # Use non-interactive backend for tests @@ -355,6 +368,366 @@ class TestLoadFindingsForChecks: assert result == {} +class TestCleanupStaleTmpOutputDirectories: + """Unit tests for opportunistic stale cleanup under tmp output root.""" + + def test_removes_only_scan_dirs_older_than_ttl(self, tmp_path, monkeypatch): + """Should remove stale scan directories and keep recent ones.""" + root_dir = tmp_path / "prowler_api_output" + + old_scan_dir = root_dir / "tenant-a" / "scan-old" + old_scan_dir.mkdir(parents=True) + (old_scan_dir / "artifact.txt").write_text("old") + + recent_scan_dir = root_dir / "tenant-a" / "scan-recent" + recent_scan_dir.mkdir(parents=True) + (recent_scan_dir / "artifact.txt").write_text("recent") + + now = time.time() + stale_ts = now - ((STALE_TMP_OUTPUT_MAX_AGE_HOURS + 1) * 60 * 60) + os.utime(old_scan_dir, (stale_ts, stale_ts)) + + monkeypatch.setattr( + "tasks.jobs.report.STALE_TMP_OUTPUT_SAFE_ROOT", root_dir.resolve() + ) + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", lambda *_: True + ) + monkeypatch.setattr( + "tasks.jobs.report._is_scan_directory_protected", lambda **_: False + ) + + removed = _cleanup_stale_tmp_output_directories( + str(root_dir), max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS + ) + + assert removed == 1 + assert not old_scan_dir.exists() + assert recent_scan_dir.exists() + + def test_skips_current_scan_even_when_stale(self, tmp_path, monkeypatch): + """Should not delete stale directory for the currently processed scan.""" + root_dir = tmp_path / "prowler_api_output" + + current_scan_dir = root_dir / "tenant-current" / "scan-current" + current_scan_dir.mkdir(parents=True) + (current_scan_dir / "artifact.txt").write_text("current") + + other_stale_scan_dir = root_dir / "tenant-other" / "scan-old" + other_stale_scan_dir.mkdir(parents=True) + (other_stale_scan_dir / "artifact.txt").write_text("other") + + now = time.time() + stale_ts = now - ((STALE_TMP_OUTPUT_MAX_AGE_HOURS + 1) * 60 * 60) + os.utime(current_scan_dir, (stale_ts, stale_ts)) + os.utime(other_stale_scan_dir, (stale_ts, stale_ts)) + + monkeypatch.setattr( + "tasks.jobs.report.STALE_TMP_OUTPUT_SAFE_ROOT", root_dir.resolve() + ) + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", lambda *_: True + ) + monkeypatch.setattr( + "tasks.jobs.report._is_scan_directory_protected", lambda **_: False + ) + + removed = _cleanup_stale_tmp_output_directories( + str(root_dir), + max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS, + exclude_scan=("tenant-current", "scan-current"), + ) + + assert removed == 1 + assert current_scan_dir.exists() + assert not other_stale_scan_dir.exists() + + def test_respects_max_deletions_per_run(self, tmp_path, monkeypatch): + """Cleanup should stop deleting when max_deletions_per_run is reached.""" + root_dir = tmp_path / "prowler_api_output" + + stale_dir_1 = root_dir / "tenant-a" / "scan-old-1" + stale_dir_2 = root_dir / "tenant-a" / "scan-old-2" + stale_dir_1.mkdir(parents=True) + stale_dir_2.mkdir(parents=True) + (stale_dir_1 / "artifact.txt").write_text("old-1") + (stale_dir_2 / "artifact.txt").write_text("old-2") + + now = time.time() + stale_ts = now - ((STALE_TMP_OUTPUT_MAX_AGE_HOURS + 1) * 60 * 60) + os.utime(stale_dir_1, (stale_ts, stale_ts)) + os.utime(stale_dir_2, (stale_ts, stale_ts)) + + monkeypatch.setattr( + "tasks.jobs.report.STALE_TMP_OUTPUT_SAFE_ROOT", root_dir.resolve() + ) + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", lambda *_: True + ) + monkeypatch.setattr( + "tasks.jobs.report._is_scan_directory_protected", lambda **_: False + ) + + removed = _cleanup_stale_tmp_output_directories( + str(root_dir), + max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS, + max_deletions_per_run=1, + ) + + assert removed == 1 + remaining = sum( + 1 for scan_dir in (stale_dir_1, stale_dir_2) if scan_dir.exists() + ) + assert remaining == 1 + + def test_rejects_non_safe_root(self, tmp_path, monkeypatch): + """Cleanup must no-op when called with a root outside the allowed safe root.""" + root_dir = tmp_path / "prowler_api_output" + root_dir.mkdir(parents=True) + + monkeypatch.setattr( + "tasks.jobs.report.STALE_TMP_OUTPUT_SAFE_ROOT", + (tmp_path / "another-root").resolve(), + ) + + def _fail_should_run(*_args, **_kwargs): + raise AssertionError("_should_run_stale_cleanup should not be called") + + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", _fail_should_run + ) + + removed = _cleanup_stale_tmp_output_directories(str(root_dir), max_age_hours=48) + + assert removed == 0 + + def test_ignores_symlink_scan_directories(self, tmp_path, monkeypatch): + """Symlinked scan directories must never be deleted by cleanup.""" + root_dir = tmp_path / "prowler_api_output" + stale_real_scan_dir = root_dir / "tenant-a" / "scan-old-real" + stale_real_scan_dir.mkdir(parents=True) + (stale_real_scan_dir / "artifact.txt").write_text("old") + + symlink_target = tmp_path / "symlink-target" + symlink_target.mkdir(parents=True) + (symlink_target / "artifact.txt").write_text("target") + symlink_scan_dir = root_dir / "tenant-a" / "scan-link" + symlink_scan_dir.symlink_to(symlink_target, target_is_directory=True) + + now = time.time() + stale_ts = now - ((STALE_TMP_OUTPUT_MAX_AGE_HOURS + 1) * 60 * 60) + os.utime(stale_real_scan_dir, (stale_ts, stale_ts)) + + monkeypatch.setattr( + "tasks.jobs.report.STALE_TMP_OUTPUT_SAFE_ROOT", root_dir.resolve() + ) + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", lambda *_: True + ) + monkeypatch.setattr( + "tasks.jobs.report._is_scan_directory_protected", lambda **_: False + ) + + removed = _cleanup_stale_tmp_output_directories( + str(root_dir), max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS + ) + + assert removed == 1 + assert not stale_real_scan_dir.exists() + assert symlink_scan_dir.exists() + assert symlink_target.exists() + + def test_handles_internal_exception_without_propagating( + self, tmp_path, monkeypatch + ): + """Cleanup errors must be swallowed so callers are not interrupted.""" + root_dir = tmp_path / "prowler_api_output" + stale_scan_dir = root_dir / "tenant-a" / "scan-old" + stale_scan_dir.mkdir(parents=True) + + now = time.time() + stale_ts = now - ((STALE_TMP_OUTPUT_MAX_AGE_HOURS + 1) * 60 * 60) + os.utime(stale_scan_dir, (stale_ts, stale_ts)) + + monkeypatch.setattr( + "tasks.jobs.report.STALE_TMP_OUTPUT_SAFE_ROOT", root_dir.resolve() + ) + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", lambda *_: True + ) + + def _raise(*_args, **_kwargs): + raise RuntimeError("db timeout") + + monkeypatch.setattr("tasks.jobs.report._is_scan_directory_protected", _raise) + + removed = _cleanup_stale_tmp_output_directories( + str(root_dir), max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS + ) + + assert removed == 0 + assert stale_scan_dir.exists() + + def test_safe_root_follows_custom_tmp_output_directory(self, tmp_path, monkeypatch): + """Custom DJANGO_TMP_OUTPUT_DIRECTORY must be honored as the safe root.""" + from tasks.jobs import report as report_module + + custom_root = tmp_path / "custom_tmp_output" + custom_root.mkdir(parents=True) + + monkeypatch.setattr( + report_module, "DJANGO_TMP_OUTPUT_DIRECTORY", str(custom_root) + ) + + resolved_root = report_module._resolve_stale_tmp_safe_root() + assert resolved_root == custom_root.resolve() + + stale_scan_dir = custom_root / "tenant-a" / "scan-old" + stale_scan_dir.mkdir(parents=True) + (stale_scan_dir / "artifact.txt").write_text("old") + + stale_ts = time.time() - ((STALE_TMP_OUTPUT_MAX_AGE_HOURS + 1) * 60 * 60) + os.utime(stale_scan_dir, (stale_ts, stale_ts)) + + monkeypatch.setattr(report_module, "STALE_TMP_OUTPUT_SAFE_ROOT", resolved_root) + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", lambda *_: True + ) + monkeypatch.setattr( + "tasks.jobs.report._is_scan_directory_protected", lambda **_: False + ) + + removed = _cleanup_stale_tmp_output_directories( + str(custom_root), max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS + ) + + assert removed == 1 + assert not stale_scan_dir.exists() + + @pytest.mark.parametrize( + "forbidden_root", + ["/", "/tmp", "/var", "/var/tmp", "/home", "/root", "/etc", "/usr"], + ) + def test_safe_root_rejects_forbidden_system_roots( + self, forbidden_root, monkeypatch + ): + """Cleanup must refuse to operate against shared system roots.""" + from tasks.jobs import report as report_module + + monkeypatch.setattr( + report_module, "DJANGO_TMP_OUTPUT_DIRECTORY", forbidden_root + ) + + assert report_module._resolve_stale_tmp_safe_root() is None + + def test_skips_cleanup_when_safe_root_is_none(self, tmp_path, monkeypatch): + """A None safe root (forbidden config) must short-circuit the cleanup.""" + root_dir = tmp_path / "prowler_api_output" + root_dir.mkdir(parents=True) + + monkeypatch.setattr("tasks.jobs.report.STALE_TMP_OUTPUT_SAFE_ROOT", None) + + def _fail_should_run(*_args, **_kwargs): + raise AssertionError("_should_run_stale_cleanup should not be called") + + monkeypatch.setattr( + "tasks.jobs.report._should_run_stale_cleanup", _fail_should_run + ) + + removed = _cleanup_stale_tmp_output_directories( + str(root_dir), max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS + ) + + assert removed == 0 + + +class TestStaleCleanupProtectionHelpers: + """Unit tests for stale cleanup helper guard logic.""" + + def test_should_run_cleanup_is_throttled(self, tmp_path): + root_dir = tmp_path / "prowler_api_output" + root_dir.mkdir(parents=True) + + assert _should_run_stale_cleanup(root_dir, throttle_seconds=3600) is True + assert _should_run_stale_cleanup(root_dir, throttle_seconds=3600) is False + + lock_file = root_dir / STALE_TMP_OUTPUT_LOCK_FILE_NAME + lock_file.write_text(str(int(time.time()) - 7200), encoding="ascii") + + assert _should_run_stale_cleanup(root_dir, throttle_seconds=3600) is True + + @patch("tasks.jobs.report.fcntl.flock", side_effect=BlockingIOError) + def test_should_run_cleanup_returns_false_when_lock_is_busy( + self, _mock_flock, tmp_path + ): + root_dir = tmp_path / "prowler_api_output" + root_dir.mkdir(parents=True) + + assert _should_run_stale_cleanup(root_dir, throttle_seconds=3600) is False + + @patch("tasks.jobs.report.Scan.all_objects.using") + def test_is_scan_directory_protected_for_executing_scan( + self, mock_scan_using, tmp_path + ): + scan_id = str(uuid.uuid4()) + scan_path = tmp_path / scan_id + scan_path.mkdir(parents=True) + mock_scan_using.return_value.filter.return_value.only.return_value.first.return_value = Mock( + state=StateChoices.EXECUTING, output_location=None + ) + + assert ( + _is_scan_directory_protected( + tenant_id="tenant-a", + scan_id=scan_id, + scan_path=scan_path, + ) + is True + ) + + @patch("tasks.jobs.report.Scan.all_objects.using") + def test_is_scan_directory_protected_for_local_output( + self, mock_scan_using, tmp_path + ): + scan_id = str(uuid.uuid4()) + scan_path = tmp_path / scan_id + scan_path.mkdir(parents=True) + local_output_path = scan_path / "outputs.zip" + mock_scan_using.return_value.filter.return_value.only.return_value.first.return_value = Mock( + state=StateChoices.COMPLETED, output_location=str(local_output_path) + ) + + assert ( + _is_scan_directory_protected( + tenant_id="tenant-a", + scan_id=scan_id, + scan_path=scan_path.resolve(), + ) + is True + ) + + @patch("tasks.jobs.report.Scan.all_objects.using") + def test_is_scan_directory_not_protected_for_s3_output( + self, mock_scan_using, tmp_path + ): + scan_id = str(uuid.uuid4()) + scan_path = tmp_path / scan_id + scan_path.mkdir(parents=True) + mock_scan_using.return_value.filter.return_value.only.return_value.first.return_value = Mock( + state=StateChoices.COMPLETED, + output_location="s3://bucket/path/report.zip", + ) + + assert ( + _is_scan_directory_protected( + tenant_id="tenant-a", + scan_id=scan_id, + scan_path=scan_path, + ) + is False + ) + + @pytest.mark.django_db class TestGenerateThreatscoreReportFunction: """Test suite for generate_threatscore_report function.""" @@ -426,6 +799,31 @@ class TestGenerateComplianceReportsOptimized: mock_ens.assert_not_called() mock_nis2.assert_not_called() + @patch( + "tasks.jobs.report._cleanup_stale_tmp_output_directories", + side_effect=RuntimeError("cleanup boom"), + ) + def test_cleanup_exception_does_not_break_no_findings_flow(self, _mock_cleanup): + """Unexpected cleanup failures must not abort report generation.""" + random_tenant = str(uuid.uuid4()) + random_scan = str(uuid.uuid4()) + random_provider = str(uuid.uuid4()) + + with patch("tasks.jobs.report.ScanSummary.objects.filter") as mock_filter: + mock_filter.return_value.exists.return_value = False + result = generate_compliance_reports( + tenant_id=random_tenant, + scan_id=random_scan, + provider_id=random_provider, + generate_threatscore=True, + generate_ens=False, + generate_nis2=False, + generate_csa=False, + generate_cis=False, + ) + + assert result["threatscore"] == {"upload": False, "path": ""} + @patch("tasks.jobs.report._upload_to_s3") @patch("tasks.jobs.report.generate_cis_report") def test_no_findings_returns_flat_cis_entry( @@ -457,6 +855,103 @@ class TestGenerateComplianceReportsOptimized: assert result["cis"] == {"upload": False, "path": ""} mock_cis.assert_not_called() + @patch("tasks.jobs.report.rmtree") + @patch("tasks.jobs.report._upload_to_s3") + @patch("tasks.jobs.report.generate_threatscore_report") + @patch("tasks.jobs.report._generate_compliance_output_directory") + @patch("tasks.jobs.report._aggregate_requirement_statistics_from_database") + @patch("tasks.jobs.report.Compliance.get_bulk") + @patch("tasks.jobs.report.Provider.objects.get") + @patch("tasks.jobs.report.ScanSummary.objects.filter") + def test_cleanup_runs_when_supported_reports_upload_successfully( + self, + mock_scan_summary_filter, + mock_provider_get, + mock_get_bulk, + mock_aggregate_stats, + mock_generate_output_dir, + mock_threatscore, + mock_upload_to_s3, + mock_rmtree, + ): + """Cleanup must run when all generated (supported) reports are uploaded.""" + mock_scan_summary_filter.return_value.exists.return_value = True + mock_provider_get.return_value = Mock(uid="provider-uid", provider="m365") + mock_get_bulk.return_value = {} + mock_aggregate_stats.return_value = {} + mock_generate_output_dir.return_value = ( + "/tmp/tenant/scan/threatscore/prowler-output-provider-20240101000000" + ) + mock_upload_to_s3.return_value = ( + "s3://bucket/tenant/scan/threatscore/report.pdf" + ) + + result = generate_compliance_reports( + tenant_id=str(uuid.uuid4()), + scan_id=str(uuid.uuid4()), + provider_id=str(uuid.uuid4()), + generate_threatscore=True, + generate_ens=True, + generate_nis2=True, + generate_csa=True, + generate_cis=True, + ) + + assert result["threatscore"]["upload"] is True + assert result["ens"]["upload"] is False + assert result["nis2"]["upload"] is False + assert result["csa"]["upload"] is False + assert result["cis"] == {"upload": False, "path": ""} + mock_generate_output_dir.assert_called_once() + mock_threatscore.assert_called_once() + mock_rmtree.assert_called_once() + + @patch("tasks.jobs.report.rmtree") + @patch("tasks.jobs.report._upload_to_s3") + @patch("tasks.jobs.report.generate_threatscore_report") + @patch("tasks.jobs.report._generate_compliance_output_directory") + @patch("tasks.jobs.report._aggregate_requirement_statistics_from_database") + @patch("tasks.jobs.report.Compliance.get_bulk") + @patch("tasks.jobs.report.Provider.objects.get") + @patch("tasks.jobs.report.ScanSummary.objects.filter") + def test_cleanup_skipped_when_supported_upload_fails( + self, + mock_scan_summary_filter, + mock_provider_get, + mock_get_bulk, + mock_aggregate_stats, + mock_generate_output_dir, + mock_threatscore, + mock_upload_to_s3, + mock_rmtree, + ): + """Cleanup must not run when a generated report upload fails.""" + mock_scan_summary_filter.return_value.exists.return_value = True + mock_provider_get.return_value = Mock(uid="provider-uid", provider="m365") + mock_get_bulk.return_value = {} + mock_aggregate_stats.return_value = {} + mock_generate_output_dir.return_value = ( + "/tmp/tenant/scan/threatscore/prowler-output-provider-20240101000000" + ) + mock_upload_to_s3.return_value = None + + result = generate_compliance_reports( + tenant_id=str(uuid.uuid4()), + scan_id=str(uuid.uuid4()), + provider_id=str(uuid.uuid4()), + generate_threatscore=True, + generate_ens=True, + generate_nis2=True, + generate_csa=True, + generate_cis=True, + ) + + assert result["threatscore"]["upload"] is False + assert result["cis"] == {"upload": False, "path": ""} + mock_generate_output_dir.assert_called_once() + mock_threatscore.assert_called_once() + mock_rmtree.assert_not_called() + @pytest.mark.django_db class TestGenerateComplianceReportsCIS: @@ -622,6 +1117,43 @@ class TestGenerateComplianceReportsCIS: assert result["cis"] == {"upload": False, "path": ""} mock_cis.assert_not_called() + @patch("tasks.jobs.report._aggregate_requirement_statistics_from_database") + @patch("tasks.jobs.report._generate_compliance_output_directory") + @patch("tasks.jobs.report.Compliance.get_bulk") + def test_cis_output_directory_failure_is_captured( + self, + mock_get_bulk, + mock_generate_output_dir, + mock_stats, + monkeypatch, + tenants_fixture, + scans_fixture, + providers_fixture, + ): + """CIS output dir errors must be captured in results (not raised).""" + tenant = tenants_fixture[0] + scan = scans_fixture[0] + provider = providers_fixture[0] + + self._force_scan_has_findings(monkeypatch) + mock_stats.return_value = {} + mock_get_bulk.return_value = {"cis_5.0_aws": Mock()} + mock_generate_output_dir.side_effect = RuntimeError("dir boom") + + result = generate_compliance_reports( + tenant_id=str(tenant.id), + scan_id=str(scan.id), + provider_id=str(provider.id), + generate_threatscore=False, + generate_ens=False, + generate_nis2=False, + generate_csa=False, + generate_cis=True, + ) + + assert result["cis"]["upload"] is False + assert result["cis"]["error"] == "dir boom" + class TestPickLatestCisVariant: """Unit tests for `_pick_latest_cis_variant` helper.""" diff --git a/api/src/backend/tasks/tests/test_tasks.py b/api/src/backend/tasks/tests/test_tasks.py index 88ef895f65..04bb91c916 100644 --- a/api/src/backend/tasks/tests/test_tasks.py +++ b/api/src/backend/tasks/tests/test_tasks.py @@ -13,6 +13,8 @@ from tasks.jobs.lighthouse_providers import ( _extract_bedrock_credentials, ) from tasks.tasks import ( + DJANGO_TMP_OUTPUT_DIRECTORY, + STALE_TMP_OUTPUT_MAX_AGE_HOURS, _cleanup_orphan_scheduled_scans, _perform_scan_complete_tasks, check_integrations_task, @@ -236,7 +238,8 @@ class TestGenerateOutputs: self.provider_id = str(uuid.uuid4()) self.tenant_id = str(uuid.uuid4()) - def test_no_findings_returns_early(self): + @patch("tasks.tasks._cleanup_stale_tmp_output_directories") + def test_no_findings_returns_early(self, mock_cleanup_stale_tmp_output_directories): with patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter: mock_filter.return_value.exists.return_value = False @@ -248,6 +251,34 @@ class TestGenerateOutputs: assert result == {"upload": False} mock_filter.assert_called_once_with(scan_id=self.scan_id) + mock_cleanup_stale_tmp_output_directories.assert_called_once_with( + DJANGO_TMP_OUTPUT_DIRECTORY, + max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS, + exclude_scan=(self.tenant_id, self.scan_id), + ) + + @patch( + "tasks.tasks._cleanup_stale_tmp_output_directories", + side_effect=RuntimeError("cleanup boom"), + ) + def test_cleanup_exception_does_not_break_no_findings_flow( + self, mock_cleanup_stale_tmp_output_directories + ): + with patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter: + mock_filter.return_value.exists.return_value = False + + result = generate_outputs_task( + scan_id=self.scan_id, + provider_id=self.provider_id, + tenant_id=self.tenant_id, + ) + + assert result == {"upload": False} + mock_cleanup_stale_tmp_output_directories.assert_called_once_with( + DJANGO_TMP_OUTPUT_DIRECTORY, + max_age_hours=STALE_TMP_OUTPUT_MAX_AGE_HOURS, + exclude_scan=(self.tenant_id, self.scan_id), + ) @patch("tasks.tasks._upload_to_s3") @patch("tasks.tasks._compress_output_files") @@ -309,7 +340,7 @@ class TestGenerateOutputs: ), patch( "tasks.tasks.COMPLIANCE_CLASS_MAP", - {"aws": [(lambda x: True, MagicMock(name="CSVCompliance"))]}, + {"aws": [(lambda _x: True, MagicMock(name="CSVCompliance"))]}, ), patch( "tasks.tasks._generate_output_directory", @@ -361,7 +392,7 @@ class TestGenerateOutputs: ), patch( "tasks.tasks.COMPLIANCE_CLASS_MAP", - {"aws": [(lambda x: True, MagicMock())]}, + {"aws": [(lambda _x: True, MagicMock())]}, ), patch("tasks.tasks._compress_output_files", return_value="/tmp/compressed"), patch("tasks.tasks._upload_to_s3", return_value=None), @@ -441,7 +472,7 @@ class TestGenerateOutputs: ), patch( "tasks.tasks.COMPLIANCE_CLASS_MAP", - {"aws": [(lambda x: True, mock_compliance_class)]}, + {"aws": [(lambda _x: True, mock_compliance_class)]}, ), ): mock_filter.return_value.exists.return_value = True @@ -470,6 +501,10 @@ class TestGenerateOutputs: class TrackingWriter: def __init__(self, findings, file_path, file_extension, from_cli): + self.findings = findings + self.file_path = file_path + self.file_extension = file_extension + self.from_cli = from_cli self.transform_called = 0 self.batch_write_data_to_file = MagicMock() self._data = [] @@ -578,13 +613,13 @@ class TestGenerateOutputs: patch("tasks.tasks.FindingOutput._transform_findings_stats"), patch( "tasks.tasks.FindingOutput.transform_api_finding", - side_effect=lambda f, prov: f, + side_effect=lambda f, _prov: f, ), patch("tasks.tasks._compress_output_files", return_value="outdir.zip"), patch("tasks.tasks._upload_to_s3", return_value="s3://bucket/outdir.zip"), patch( "tasks.tasks.Scan.all_objects.filter", - return_value=MagicMock(update=lambda **kw: None), + return_value=MagicMock(update=lambda **_kw: None), ), patch("tasks.tasks.batched", return_value=two_batches), patch("tasks.tasks.OUTPUT_FORMATS_MAPPING", {}), @@ -666,7 +701,7 @@ class TestGenerateOutputs: ), patch( "tasks.tasks.COMPLIANCE_CLASS_MAP", - {"aws": [(lambda x: True, mock_compliance_class)]}, + {"aws": [(lambda _x: True, mock_compliance_class)]}, ), ): mock_filter.return_value.exists.return_value = True @@ -748,7 +783,7 @@ class TestScanCompleteTasks: @patch("tasks.tasks.can_provider_run_attack_paths_scan", return_value=False) def test_scan_complete_tasks( self, - mock_can_run_attack_paths, + _mock_can_run_attack_paths, mock_attack_paths_task, mock_check_integrations_task, mock_compliance_reports_task, @@ -994,7 +1029,7 @@ class TestCheckIntegrationsTask: @patch("tasks.tasks.rmtree") def test_generate_outputs_with_asff_for_aws_with_security_hub( self, - mock_rmtree, + _mock_rmtree, mock_scan_update, mock_upload, mock_compress, @@ -1122,7 +1157,7 @@ class TestCheckIntegrationsTask: @patch("tasks.tasks.rmtree") def test_generate_outputs_no_asff_for_aws_without_security_hub( self, - mock_rmtree, + _mock_rmtree, mock_scan_update, mock_upload, mock_compress, @@ -1245,7 +1280,7 @@ class TestCheckIntegrationsTask: @patch("tasks.tasks.rmtree") def test_generate_outputs_no_asff_for_non_aws_provider( self, - mock_rmtree, + _mock_rmtree, mock_scan_update, mock_upload, mock_compress,