fix(api): prevent /tmp saturation from compliance report generation (#10874)

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)

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:
+        <tmp_output_root>/<tenant_id>/<scan_id>/...
+
+    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)

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}")

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."""

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,