fix(api): mark attack paths scan as failed when celery task fails (#10065)

Author: josema-xyz

api/CHANGELOG.md

@@ -18,6 +18,7 @@ All notable changes to the **Prowler API** are documented in this file.
 - Support CSA CCM 4.0 for the Azure provider [(#10039)](https://github.com/prowler-cloud/prowler/pull/10039)
 - Support CSA CCM 4.0 for the Oracle Cloud provider [(#10057)](https://github.com/prowler-cloud/prowler/pull/10057)
 - Support CSA CCM 4.0 for the Alibaba Cloud provider [(#10061)](https://github.com/prowler-cloud/prowler/pull/10061)
+- Attack Paths: Mark attack Paths scan as failed when Celery task fails outside job error handling [(#10065)](https://github.com/prowler-cloud/prowler/pull/10065)
 
 ### 🔐 Security
 

api/poetry.lock

@@ -86,7 +86,11 @@ def finish_attack_paths_scan(
 ) -> None:
     with rls_transaction(attack_paths_scan.tenant_id):
         now = datetime.now(tz=timezone.utc)
-        duration = int((now - attack_paths_scan.started_at).total_seconds())
+        duration = (
+            int((now - attack_paths_scan.started_at).total_seconds())
+            if attack_paths_scan.started_at
+            else 0
+        )
 
         attack_paths_scan.state = state
         attack_paths_scan.progress = 100
@@ -144,3 +148,24 @@ def update_old_attack_paths_scan(
     with rls_transaction(old_attack_paths_scan.tenant_id):
         old_attack_paths_scan.is_graph_database_deleted = True
         old_attack_paths_scan.save(update_fields=["is_graph_database_deleted"])
+
+
+def fail_attack_paths_scan(
+    tenant_id: str,
+    scan_id: str,
+    error: str,
+) -> None:
+    """
+    Mark the `AttackPathsScan` row as `FAILED` unless it's already `COMPLETED` or `FAILED`.
+    Used as a safety net when the Celery task fails outside the job's own error handling.
+    """
+    attack_paths_scan = retrieve_attack_paths_scan(tenant_id, scan_id)
+    if attack_paths_scan and attack_paths_scan.state not in (
+        StateChoices.COMPLETED,
+        StateChoices.FAILED,
+    ):
+        finish_attack_paths_scan(
+            attack_paths_scan,
+            StateChoices.FAILED,
+            {"global_error": error},
+        )

api/src/backend/tasks/jobs/attack_paths/db_utils.py

@@ -228,10 +228,16 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
     except Exception as e:
         exception_message = utils.stringify_exception(e, "Cartography failed")
         logger.error(exception_message)
-        ingestion_exceptions["global_cartography_error"] = exception_message
+        ingestion_exceptions["global_error"] = exception_message
 
         # Handling databases changes
-        graph_database.drop_database(tmp_cartography_config.neo4j_database)
+        try:
+            graph_database.drop_database(tmp_cartography_config.neo4j_database)
+        except Exception:
+            logger.exception(
+                f"Failed to drop temporary Neo4j database {tmp_cartography_config.neo4j_database} during cleanup"
+            )
+
         db_utils.finish_attack_paths_scan(
             attack_paths_scan, StateChoices.FAILED, ingestion_exceptions
         )

api/src/backend/tasks/jobs/attack_paths/scan.py

@@ -10,6 +10,7 @@
 from django_celery_beat.models import PeriodicTask
 from tasks.jobs.attack_paths import (
     attack_paths_scan,
+    db_utils as attack_paths_db_utils,
     can_provider_run_attack_paths_scan,
 )
 from tasks.jobs.backfill import (
@@ -359,8 +360,25 @@ def perform_scan_summary_task(tenant_id: str, scan_id: str):
     return aggregate_findings(tenant_id=tenant_id, scan_id=scan_id)
 
 
+class AttackPathsScanRLSTask(RLSTask):
+    """
+    RLS task that marks the `AttackPathsScan` DB row as `FAILED` when the Celery task fails.
+
+    Covers failures that happen outside the job's own try/except (e.g. provider lookup,
+    SDK initialization, or Neo4j configuration errors during setup).
+    """
+
+    def on_failure(self, exc, task_id, args, kwargs, _einfo):
+        tenant_id = kwargs.get("tenant_id")
+        scan_id = kwargs.get("scan_id")
+
+        if tenant_id and scan_id:
+            logger.error(f"Attack paths scan task {task_id} failed: {exc}")
+            attack_paths_db_utils.fail_attack_paths_scan(tenant_id, scan_id, str(exc))
+
+
 @shared_task(
-    base=RLSTask,
+    base=AttackPathsScanRLSTask,
     bind=True,
     name="attack-paths-scan-perform",
     queue="attack-paths-scans",