feat(attack-paths): The complete Attack Paths feature (#9805)

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: César Arroba <19954079+cesararroba@users.noreply.github.com>
Co-authored-by: Alan Buscaglia <gentlemanprogramming@gmail.com>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Andoni Alonso <14891798+andoniaf@users.noreply.github.com>
Co-authored-by: Rubén De la Torre Vico <ruben@prowler.com>
Co-authored-by: HugoPBrito <hugopbrit@gmail.com>
Co-authored-by: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com>
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
Co-authored-by: Chandrapal Badshah <Chan9390@users.noreply.github.com>
Co-authored-by: Chandrapal Badshah <12944530+Chan9390@users.noreply.github.com>
Co-authored-by: Adrián Peña <adrianjpr@gmail.com>
Co-authored-by: Pedro Martín <pedromarting3@gmail.com>
Co-authored-by: KonstGolfi <73020281+KonstGolfi@users.noreply.github.com>
Co-authored-by: lydiavilchez <114735608+lydiavilchez@users.noreply.github.com>
Co-authored-by: Prowler Bot <bot@prowler.com>
Co-authored-by: prowler-bot <179230569+prowler-bot@users.noreply.github.com>
Co-authored-by: StylusFrost <43682773+StylusFrost@users.noreply.github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: alejandrobailo <alejandrobailo94@gmail.com>
Co-authored-by: Alejandro Bailo <59607668+alejandrobailo@users.noreply.github.com>
Co-authored-by: Víctor Fernández Poyatos <victor@prowler.com>
Co-authored-by: bota4go <108249054+bota4go@users.noreply.github.com>
Co-authored-by: Daniel Barranquero <74871504+danibarranqueroo@users.noreply.github.com>
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
Co-authored-by: mchennai <50082780+mchennai@users.noreply.github.com>
Co-authored-by: Ryan Nolette <sonofagl1tch@users.noreply.github.com>
Co-authored-by: Ulissis Correa <123517149+ulissisc@users.noreply.github.com>
Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
Co-authored-by: Lee Trout <ltrout@watchpointlabs.com>
Co-authored-by: Sergio Garcia <sergargar1@gmail.com>
Co-authored-by: Alan-TheGentleman <alan@thegentleman.dev>

api/src/backend/tasks/beat.py

@@ -7,6 +7,7 @@ from tasks.tasks import perform_scheduled_scan_task
 from api.db_utils import rls_transaction
 from api.exceptions import ConflictException
 from api.models import Provider, Scan, StateChoices
+from tasks.jobs.attack_paths import db_utils as attack_paths_db_utils
 
 
 def schedule_provider_scan(provider_instance: Provider):
@@ -39,6 +40,12 @@ def schedule_provider_scan(provider_instance: Provider):
             scheduled_at=datetime.now(timezone.utc),
         )
 
+    attack_paths_db_utils.create_attack_paths_scan(
+        tenant_id=tenant_id,
+        scan_id=str(scheduled_scan.id),
+        provider_id=provider_id,
+    )
+
     # Schedule the task
     periodic_task_instance = PeriodicTask.objects.create(
         interval=schedule,

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

@@ -0,0 +1,7 @@
+from tasks.jobs.attack_paths.db_utils import can_provider_run_attack_paths_scan
+from tasks.jobs.attack_paths.scan import run as attack_paths_scan
+
+__all__ = [
+    "attack_paths_scan",
+    "can_provider_run_attack_paths_scan",
+]

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

@@ -0,0 +1,237 @@
+# Portions of this file are based on code from the Cartography project
+# (https://github.com/cartography-cncf/cartography), which is licensed under the Apache 2.0 License.
+
+from typing import Any
+
+import aioboto3
+import boto3
+import neo4j
+
+from cartography.config import Config as CartographyConfig
+from cartography.intel import aws as cartography_aws
+from celery.utils.log import get_task_logger
+
+from api.models import (
+    AttackPathsScan as ProwlerAPIAttackPathsScan,
+    Provider as ProwlerAPIProvider,
+)
+from prowler.providers.common.provider import Provider as ProwlerSDKProvider
+from tasks.jobs.attack_paths import db_utils, utils
+
+logger = get_task_logger(__name__)
+
+
+def start_aws_ingestion(
+    neo4j_session: neo4j.Session,
+    cartography_config: CartographyConfig,
+    prowler_api_provider: ProwlerAPIProvider,
+    prowler_sdk_provider: ProwlerSDKProvider,
+    attack_paths_scan: ProwlerAPIAttackPathsScan,
+) -> dict[str, dict[str, str]]:
+    """
+    Code based on Cartography version 0.122.0, specifically on `cartography.intel.aws.__init__.py`.
+
+    For the scan progress updates:
+        - The caller of this function (`tasks.jobs.attack_paths.scan.run`) has set it to 2.
+        - When the control returns to the caller, it will be set to 95.
+    """
+
+    # Initialize variables common to all jobs
+    common_job_parameters = {
+        "UPDATE_TAG": cartography_config.update_tag,
+        "permission_relationships_file": cartography_config.permission_relationships_file,
+        "aws_guardduty_severity_threshold": cartography_config.aws_guardduty_severity_threshold,
+        "aws_cloudtrail_management_events_lookback_hours": cartography_config.aws_cloudtrail_management_events_lookback_hours,
+        "experimental_aws_inspector_batch": cartography_config.experimental_aws_inspector_batch,
+    }
+
+    boto3_session = get_boto3_session(prowler_api_provider, prowler_sdk_provider)
+    regions: list[str] = list(prowler_sdk_provider._enabled_regions)
+    requested_syncs = list(cartography_aws.RESOURCE_FUNCTIONS.keys())
+
+    sync_args = cartography_aws._build_aws_sync_kwargs(
+        neo4j_session,
+        boto3_session,
+        regions,
+        prowler_api_provider.uid,
+        cartography_config.update_tag,
+        common_job_parameters,
+    )
+
+    # Starting with sync functions
+    cartography_aws.organizations.sync(
+        neo4j_session,
+        {prowler_api_provider.alias: prowler_api_provider.uid},
+        cartography_config.update_tag,
+        common_job_parameters,
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 3)
+
+    # Adding an extra field
+    common_job_parameters["AWS_ID"] = prowler_api_provider.uid
+
+    cartography_aws._autodiscover_accounts(
+        neo4j_session,
+        boto3_session,
+        prowler_api_provider.uid,
+        cartography_config.update_tag,
+        common_job_parameters,
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 4)
+
+    failed_syncs = sync_aws_account(
+        prowler_api_provider, requested_syncs, sync_args, attack_paths_scan
+    )
+
+    if "permission_relationships" in requested_syncs:
+        cartography_aws.RESOURCE_FUNCTIONS["permission_relationships"](**sync_args)
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 88)
+
+    if "resourcegroupstaggingapi" in requested_syncs:
+        cartography_aws.RESOURCE_FUNCTIONS["resourcegroupstaggingapi"](**sync_args)
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 89)
+
+    cartography_aws.run_scoped_analysis_job(
+        "aws_ec2_iaminstanceprofile.json",
+        neo4j_session,
+        common_job_parameters,
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 90)
+
+    cartography_aws.run_analysis_job(
+        "aws_lambda_ecr.json",
+        neo4j_session,
+        common_job_parameters,
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 91)
+
+    cartography_aws.merge_module_sync_metadata(
+        neo4j_session,
+        group_type="AWSAccount",
+        group_id=prowler_api_provider.uid,
+        synced_type="AWSAccount",
+        update_tag=cartography_config.update_tag,
+        stat_handler=cartography_aws.stat_handler,
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 92)
+
+    # Removing the added extra field
+    del common_job_parameters["AWS_ID"]
+
+    cartography_aws.run_cleanup_job(
+        "aws_post_ingestion_principals_cleanup.json",
+        neo4j_session,
+        common_job_parameters,
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 93)
+
+    cartography_aws._perform_aws_analysis(
+        requested_syncs, neo4j_session, common_job_parameters
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 94)
+
+    return failed_syncs
+
+
+def get_boto3_session(
+    prowler_api_provider: ProwlerAPIProvider, prowler_sdk_provider: ProwlerSDKProvider
+) -> boto3.Session:
+    boto3_session = prowler_sdk_provider.session.current_session
+
+    aws_accounts_from_session = cartography_aws.organizations.get_aws_account_default(
+        boto3_session
+    )
+    if not aws_accounts_from_session:
+        raise Exception(
+            "No valid AWS credentials could be found. No AWS accounts can be synced."
+        )
+
+    aws_account_id_from_session = list(aws_accounts_from_session.values())[0]
+    if prowler_api_provider.uid != aws_account_id_from_session:
+        raise Exception(
+            f"Provider {prowler_api_provider.uid} doesn't match AWS account {aws_account_id_from_session}."
+        )
+
+    if boto3_session.region_name is None:
+        global_region = prowler_sdk_provider.get_global_region()
+        boto3_session._session.set_config_variable("region", global_region)
+
+    return boto3_session
+
+
+def get_aioboto3_session(boto3_session: boto3.Session) -> aioboto3.Session:
+    return aioboto3.Session(botocore_session=boto3_session._session)
+
+
+def sync_aws_account(
+    prowler_api_provider: ProwlerAPIProvider,
+    requested_syncs: list[str],
+    sync_args: dict[str, Any],
+    attack_paths_scan: ProwlerAPIAttackPathsScan,
+) -> dict[str, str]:
+    current_progress = 4  # `cartography_aws._autodiscover_accounts`
+    max_progress = (
+        87  # `cartography_aws.RESOURCE_FUNCTIONS["permission_relationships"]` - 1
+    )
+    n_steps = (
+        len(requested_syncs) - 2
+    )  # Excluding `permission_relationships` and `resourcegroupstaggingapi`
+    progress_step = (max_progress - current_progress) / n_steps
+
+    failed_syncs = {}
+
+    for func_name in requested_syncs:
+        if func_name in cartography_aws.RESOURCE_FUNCTIONS:
+            logger.info(
+                f"Syncing function {func_name} for AWS account {prowler_api_provider.uid}"
+            )
+
+            # Updating progress, not really the right place but good enough
+            current_progress += progress_step
+            db_utils.update_attack_paths_scan_progress(
+                attack_paths_scan, int(current_progress)
+            )
+
+            try:
+                # `ecr:image_layers` uses `aioboto3_session` instead of `boto3_session`
+                if func_name == "ecr:image_layers":
+                    cartography_aws.RESOURCE_FUNCTIONS[func_name](
+                        neo4j_session=sync_args.get("neo4j_session"),
+                        aioboto3_session=get_aioboto3_session(
+                            sync_args.get("boto3_session")
+                        ),
+                        regions=sync_args.get("regions"),
+                        current_aws_account_id=sync_args.get("current_aws_account_id"),
+                        update_tag=sync_args.get("update_tag"),
+                        common_job_parameters=sync_args.get("common_job_parameters"),
+                    )
+
+                # Skip permission relationships and tags for now because they rely on data already being in the graph
+                elif func_name in [
+                    "permission_relationships",
+                    "resourcegroupstaggingapi",
+                ]:
+                    continue
+
+                else:
+                    cartography_aws.RESOURCE_FUNCTIONS[func_name](**sync_args)
+
+            except Exception as e:
+                exception_message = utils.stringify_exception(
+                    e, f"Exception for AWS sync function: {func_name}"
+                )
+                failed_syncs[func_name] = exception_message
+
+                logger.warning(
+                    f"Caught exception syncing function {func_name} from AWS account {prowler_api_provider.uid}. We "
+                    "are continuing on to the next AWS sync function.",
+                )
+
+                continue
+
+        else:
+            raise ValueError(
+                f'AWS sync function "{func_name}" was specified but does not exist. Did you misspell it?'
+            )
+
+    return failed_syncs

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

@@ -0,0 +1,161 @@
+from datetime import datetime, timezone
+from typing import Any
+
+from cartography.config import Config as CartographyConfig
+
+from api.db_utils import rls_transaction
+from api.models import (
+    AttackPathsScan as ProwlerAPIAttackPathsScan,
+    Provider as ProwlerAPIProvider,
+    StateChoices,
+)
+from tasks.jobs.attack_paths.providers import is_provider_available
+
+
+def can_provider_run_attack_paths_scan(tenant_id: str, provider_id: int) -> bool:
+    with rls_transaction(tenant_id):
+        prowler_api_provider = ProwlerAPIProvider.objects.get(id=provider_id)
+
+    return is_provider_available(prowler_api_provider.provider)
+
+
+def create_attack_paths_scan(
+    tenant_id: str,
+    scan_id: str,
+    provider_id: int,
+) -> ProwlerAPIAttackPathsScan | None:
+    if not can_provider_run_attack_paths_scan(tenant_id, provider_id):
+        return None
+
+    with rls_transaction(tenant_id):
+        attack_paths_scan = ProwlerAPIAttackPathsScan.objects.create(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+            scan_id=scan_id,
+            state=StateChoices.SCHEDULED,
+            started_at=datetime.now(tz=timezone.utc),
+        )
+        attack_paths_scan.save()
+
+    return attack_paths_scan
+
+
+def retrieve_attack_paths_scan(
+    tenant_id: str,
+    scan_id: str,
+) -> ProwlerAPIAttackPathsScan | None:
+    try:
+        with rls_transaction(tenant_id):
+            attack_paths_scan = ProwlerAPIAttackPathsScan.objects.get(
+                scan_id=scan_id,
+            )
+
+        return attack_paths_scan
+
+    except ProwlerAPIAttackPathsScan.DoesNotExist:
+        return None
+
+
+def starting_attack_paths_scan(
+    attack_paths_scan: ProwlerAPIAttackPathsScan,
+    task_id: str,
+    cartography_config: CartographyConfig,
+) -> None:
+    with rls_transaction(attack_paths_scan.tenant_id):
+        attack_paths_scan.task_id = task_id
+        attack_paths_scan.state = StateChoices.EXECUTING
+        attack_paths_scan.started_at = datetime.now(tz=timezone.utc)
+        attack_paths_scan.update_tag = cartography_config.update_tag
+        attack_paths_scan.graph_database = cartography_config.neo4j_database
+
+        attack_paths_scan.save(
+            update_fields=[
+                "task_id",
+                "state",
+                "started_at",
+                "update_tag",
+                "graph_database",
+            ]
+        )
+
+
+def finish_attack_paths_scan(
+    attack_paths_scan: ProwlerAPIAttackPathsScan,
+    state: StateChoices,
+    ingestion_exceptions: dict[str, Any],
+) -> 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())
+
+        attack_paths_scan.state = state
+        attack_paths_scan.progress = 100
+        attack_paths_scan.completed_at = now
+        attack_paths_scan.duration = duration
+        attack_paths_scan.ingestion_exceptions = ingestion_exceptions
+
+        attack_paths_scan.save(
+            update_fields=[
+                "state",
+                "progress",
+                "completed_at",
+                "duration",
+                "ingestion_exceptions",
+            ]
+        )
+
+
+def update_attack_paths_scan_progress(
+    attack_paths_scan: ProwlerAPIAttackPathsScan,
+    progress: int,
+) -> None:
+    with rls_transaction(attack_paths_scan.tenant_id):
+        attack_paths_scan.progress = progress
+        attack_paths_scan.save(update_fields=["progress"])
+
+
+def get_old_attack_paths_scans(
+    tenant_id: str,
+    provider_id: str,
+    attack_paths_scan_id: str,
+) -> list[ProwlerAPIAttackPathsScan]:
+    """
+    An `old_attack_paths_scan` is any `completed` Attack Paths scan for the same provider,
+    with its graph database not deleted, excluding the current Attack Paths scan.
+    """
+
+    with rls_transaction(tenant_id):
+        completed_scans_qs = (
+            ProwlerAPIAttackPathsScan.objects.filter(
+                provider_id=provider_id,
+                state=StateChoices.COMPLETED,
+                is_graph_database_deleted=False,
+            )
+            .exclude(id=attack_paths_scan_id)
+            .all()
+        )
+
+        return list(completed_scans_qs)
+
+
+def update_old_attack_paths_scan(
+    old_attack_paths_scan: ProwlerAPIAttackPathsScan,
+) -> None:
+    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 get_provider_graph_database_names(tenant_id: str, provider_id: str) -> list[str]:
+    """
+    Return existing graph database names for a tenant/provider.
+
+    Note: For accesing the `AttackPathsScan` we need to use `all_objects` manager because the provider is soft-deleted.
+    """
+    with rls_transaction(tenant_id):
+        graph_databases_names_qs = ProwlerAPIAttackPathsScan.all_objects.filter(
+            provider_id=provider_id,
+            is_graph_database_deleted=False,
+        ).values_list("graph_database", flat=True)
+
+        return list(graph_databases_names_qs)

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

@@ -0,0 +1,23 @@
+AVAILABLE_PROVIDERS: list[str] = [
+    "aws",
+]
+
+ROOT_NODE_LABELS: dict[str, str] = {
+    "aws": "AWSAccount",
+}
+
+NODE_UID_FIELDS: dict[str, str] = {
+    "aws": "arn",
+}
+
+
+def is_provider_available(provider_type: str) -> bool:
+    return provider_type in AVAILABLE_PROVIDERS
+
+
+def get_root_node_label(provider_type: str) -> str:
+    return ROOT_NODE_LABELS.get(provider_type, "UnknownProviderAccount")
+
+
+def get_node_uid_field(provider_type: str) -> str:
+    return NODE_UID_FIELDS.get(provider_type, "UnknownProviderUID")

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

@@ -0,0 +1,205 @@
+import neo4j
+
+from cartography.client.core.tx import run_write_query
+from cartography.config import Config as CartographyConfig
+from celery.utils.log import get_task_logger
+
+from api.db_utils import rls_transaction
+from api.models import Provider, ResourceFindingMapping
+from config.env import env
+from prowler.config import config as ProwlerConfig
+from tasks.jobs.attack_paths.providers import get_node_uid_field, get_root_node_label
+
+logger = get_task_logger(__name__)
+
+BATCH_SIZE = env.int("NEO4J_INSERT_BATCH_SIZE", 500)
+
+INDEX_STATEMENTS = [
+    "CREATE INDEX prowler_finding_id IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.id);",
+    "CREATE INDEX prowler_finding_provider_uid IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.provider_uid);",
+    "CREATE INDEX prowler_finding_lastupdated IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.lastupdated);",
+    "CREATE INDEX prowler_finding_check_id IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.status);",
+]
+
+INSERT_STATEMENT_TEMPLATE = """
+    UNWIND $findings_data AS finding_data
+
+    MATCH (account:__ROOT_NODE_LABEL__ {id: $provider_uid})
+    MATCH (account)-->(resource)
+        WHERE resource.__NODE_UID_FIELD__ = finding_data.resource_uid
+            OR resource.id = finding_data.resource_uid
+
+    MERGE (finding:ProwlerFinding {id: finding_data.id})
+        ON CREATE SET
+            finding.id = finding_data.id,
+            finding.uid = finding_data.uid,
+            finding.inserted_at = finding_data.inserted_at,
+            finding.updated_at = finding_data.updated_at,
+            finding.first_seen_at = finding_data.first_seen_at,
+            finding.scan_id = finding_data.scan_id,
+            finding.delta = finding_data.delta,
+            finding.status = finding_data.status,
+            finding.status_extended = finding_data.status_extended,
+            finding.severity = finding_data.severity,
+            finding.check_id = finding_data.check_id,
+            finding.check_title = finding_data.check_title,
+            finding.muted = finding_data.muted,
+            finding.muted_reason = finding_data.muted_reason,
+            finding.provider_uid = $provider_uid,
+            finding.firstseen = timestamp(),
+            finding.lastupdated = $last_updated,
+            finding._module_name = 'cartography:prowler',
+            finding._module_version = $prowler_version
+        ON MATCH SET
+            finding.status = finding_data.status,
+            finding.status_extended = finding_data.status_extended,
+            finding.lastupdated = $last_updated
+
+    MERGE (resource)-[rel:HAS_FINDING]->(finding)
+        ON CREATE SET
+            rel.provider_uid = $provider_uid,
+            rel.firstseen = timestamp(),
+            rel.lastupdated = $last_updated,
+            rel._module_name = 'cartography:prowler',
+            rel._module_version = $prowler_version
+        ON MATCH SET
+            rel.lastupdated = $last_updated
+"""
+
+CLEANUP_STATEMENT = """
+    MATCH (finding:ProwlerFinding {provider_uid: $provider_uid})
+        WHERE finding.lastupdated < $last_updated
+
+    WITH finding LIMIT $batch_size
+
+    DETACH DELETE finding
+
+    RETURN COUNT(finding) AS deleted_findings_count
+"""
+
+
+def create_indexes(neo4j_session: neo4j.Session) -> None:
+    """
+    Code based on Cartography version 0.122.0, specifically on `cartography.intel.create_indexes.run`.
+    """
+
+    logger.info("Creating indexes for Prowler node types.")
+    for statement in INDEX_STATEMENTS:
+        logger.debug("Executing statement: %s", statement)
+        run_write_query(neo4j_session, statement)
+
+
+def analysis(
+    neo4j_session: neo4j.Session,
+    prowler_api_provider: Provider,
+    scan_id: str,
+    config: CartographyConfig,
+) -> None:
+    findings_data = get_provider_last_scan_findings(prowler_api_provider, scan_id)
+    load_findings(neo4j_session, findings_data, prowler_api_provider, config)
+    cleanup_findings(neo4j_session, prowler_api_provider, config)
+
+
+def get_provider_last_scan_findings(
+    prowler_api_provider: Provider,
+    scan_id: str,
+) -> list[dict[str, str]]:
+    with rls_transaction(prowler_api_provider.tenant_id):
+        resource_finding_qs = ResourceFindingMapping.objects.filter(
+            finding__scan_id=scan_id,
+        ).values(
+            "resource__uid",
+            "finding__id",
+            "finding__uid",
+            "finding__inserted_at",
+            "finding__updated_at",
+            "finding__first_seen_at",
+            "finding__scan_id",
+            "finding__delta",
+            "finding__status",
+            "finding__status_extended",
+            "finding__severity",
+            "finding__check_id",
+            "finding__check_metadata__checktitle",
+            "finding__muted",
+            "finding__muted_reason",
+        )
+
+        findings = []
+        for resource_finding in resource_finding_qs:
+            findings.append(
+                {
+                    "resource_uid": str(resource_finding["resource__uid"]),
+                    "id": str(resource_finding["finding__id"]),
+                    "uid": resource_finding["finding__uid"],
+                    "inserted_at": resource_finding["finding__inserted_at"],
+                    "updated_at": resource_finding["finding__updated_at"],
+                    "first_seen_at": resource_finding["finding__first_seen_at"],
+                    "scan_id": str(resource_finding["finding__scan_id"]),
+                    "delta": resource_finding["finding__delta"],
+                    "status": resource_finding["finding__status"],
+                    "status_extended": resource_finding["finding__status_extended"],
+                    "severity": resource_finding["finding__severity"],
+                    "check_id": str(resource_finding["finding__check_id"]),
+                    "check_title": resource_finding[
+                        "finding__check_metadata__checktitle"
+                    ],
+                    "muted": resource_finding["finding__muted"],
+                    "muted_reason": resource_finding["finding__muted_reason"],
+                }
+            )
+
+        return findings
+
+
+def load_findings(
+    neo4j_session: neo4j.Session,
+    findings_data: list[dict[str, str]],
+    prowler_api_provider: Provider,
+    config: CartographyConfig,
+) -> None:
+    replacements = {
+        "__ROOT_NODE_LABEL__": get_root_node_label(prowler_api_provider.provider),
+        "__NODE_UID_FIELD__": get_node_uid_field(prowler_api_provider.provider),
+    }
+    query = INSERT_STATEMENT_TEMPLATE
+    for replace_key, replace_value in replacements.items():
+        query = query.replace(replace_key, replace_value)
+
+    parameters = {
+        "provider_uid": str(prowler_api_provider.uid),
+        "last_updated": config.update_tag,
+        "prowler_version": ProwlerConfig.prowler_version,
+    }
+
+    total_length = len(findings_data)
+    for i in range(0, total_length, BATCH_SIZE):
+        parameters["findings_data"] = findings_data[i : i + BATCH_SIZE]
+
+        logger.info(
+            f"Loading findings batch {i // BATCH_SIZE + 1} / {(total_length + BATCH_SIZE - 1) // BATCH_SIZE}"
+        )
+
+        neo4j_session.run(query, parameters)
+
+
+def cleanup_findings(
+    neo4j_session: neo4j.Session,
+    prowler_api_provider: Provider,
+    config: CartographyConfig,
+) -> None:
+    parameters = {
+        "provider_uid": str(prowler_api_provider.uid),
+        "last_updated": config.update_tag,
+        "batch_size": BATCH_SIZE,
+    }
+
+    batch = 1
+    deleted_count = 1
+    while deleted_count > 0:
+        logger.info(f"Cleaning findings batch {batch}")
+
+        result = neo4j_session.run(CLEANUP_STATEMENT, parameters)
+
+        deleted_count = result.single().get("deleted_findings_count", 0)
+        batch += 1

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

@@ -0,0 +1,183 @@
+import logging
+import time
+import asyncio
+
+from typing import Any, Callable
+
+from cartography.config import Config as CartographyConfig
+from cartography.intel import analysis as cartography_analysis
+from cartography.intel import create_indexes as cartography_create_indexes
+from cartography.intel import ontology as cartography_ontology
+from celery.utils.log import get_task_logger
+
+from api.attack_paths import database as graph_database
+from api.db_utils import rls_transaction
+from api.models import (
+    Provider as ProwlerAPIProvider,
+    StateChoices,
+)
+from api.utils import initialize_prowler_provider
+from tasks.jobs.attack_paths import aws, db_utils, prowler, utils
+
+# Without this Celery goes crazy with Cartography logging
+logging.getLogger("cartography").setLevel(logging.ERROR)
+logging.getLogger("neo4j").propagate = False
+
+logger = get_task_logger(__name__)
+
+CARTOGRAPHY_INGESTION_FUNCTIONS: dict[str, Callable] = {
+    "aws": aws.start_aws_ingestion,
+}
+
+
+def get_cartography_ingestion_function(provider_type: str) -> Callable | None:
+    return CARTOGRAPHY_INGESTION_FUNCTIONS.get(provider_type)
+
+
+def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
+    """
+    Code based on Cartography version 0.122.0, specifically on `cartography.cli.main`, `cartography.cli.CLI.main`,
+    `cartography.sync.run_with_config` and `cartography.sync.Sync.run`.
+    """
+    ingestion_exceptions = {}  # This will hold any exceptions raised during ingestion
+
+    # Prowler necessary objects
+    with rls_transaction(tenant_id):
+        prowler_api_provider = ProwlerAPIProvider.objects.get(scan__pk=scan_id)
+        prowler_sdk_provider = initialize_prowler_provider(prowler_api_provider)
+
+    # Attack Paths Scan necessary objects
+    cartography_ingestion_function = get_cartography_ingestion_function(
+        prowler_api_provider.provider
+    )
+    attack_paths_scan = db_utils.retrieve_attack_paths_scan(tenant_id, scan_id)
+
+    # Checks before starting the scan
+    if not cartography_ingestion_function:
+        ingestion_exceptions = {
+            "global_error": f"Provider {prowler_api_provider.provider} is not supported for Attack Paths scans"
+        }
+        if attack_paths_scan:
+            db_utils.finish_attack_paths_scan(
+                attack_paths_scan, StateChoices.COMPLETED, ingestion_exceptions
+            )
+
+        logger.warning(
+            f"Provider {prowler_api_provider.provider} is not supported for Attack Paths scans"
+        )
+        return ingestion_exceptions
+
+    else:
+        if not attack_paths_scan:
+            logger.warning(
+                f"No Attack Paths Scan found for scan {scan_id} and tenant {tenant_id}, let's create it then"
+            )
+            attack_paths_scan = db_utils.create_attack_paths_scan(
+                tenant_id, scan_id, prowler_api_provider.id
+            )
+
+    # While creating the Cartography configuration, attributes `neo4j_user` and `neo4j_password` are not really needed in this config object
+    cartography_config = CartographyConfig(
+        neo4j_uri=graph_database.get_uri(),
+        neo4j_database=graph_database.get_database_name(attack_paths_scan.id),
+        update_tag=int(time.time()),
+    )
+
+    # Starting the Attack Paths scan
+    db_utils.starting_attack_paths_scan(attack_paths_scan, task_id, cartography_config)
+
+    try:
+        logger.info(
+            f"Creating Neo4j database {cartography_config.neo4j_database} for tenant {prowler_api_provider.tenant_id}"
+        )
+
+        graph_database.create_database(cartography_config.neo4j_database)
+        db_utils.update_attack_paths_scan_progress(attack_paths_scan, 1)
+
+        logger.info(
+            f"Starting Cartography ({attack_paths_scan.id}) for "
+            f"{prowler_api_provider.provider.upper()} provider {prowler_api_provider.id}"
+        )
+        with graph_database.get_session(
+            cartography_config.neo4j_database
+        ) as neo4j_session:
+            # Indexes creation
+            cartography_create_indexes.run(neo4j_session, cartography_config)
+            prowler.create_indexes(neo4j_session)
+            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 2)
+
+            # The real scan, where iterates over cloud services
+            ingestion_exceptions = _call_within_event_loop(
+                cartography_ingestion_function,
+                neo4j_session,
+                cartography_config,
+                prowler_api_provider,
+                prowler_sdk_provider,
+                attack_paths_scan,
+            )
+
+            # Post-processing: Just keeping it to be more Cartography compliant
+            cartography_ontology.run(neo4j_session, cartography_config)
+            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 95)
+
+            cartography_analysis.run(neo4j_session, cartography_config)
+            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 96)
+
+            # Adding Prowler nodes and relationships
+            prowler.analysis(
+                neo4j_session, prowler_api_provider, scan_id, cartography_config
+            )
+
+        logger.info(
+            f"Completed Cartography ({attack_paths_scan.id}) for "
+            f"{prowler_api_provider.provider.upper()} provider {prowler_api_provider.id}"
+        )
+
+        # Handling databases changes
+        old_attack_paths_scans = db_utils.get_old_attack_paths_scans(
+            prowler_api_provider.tenant_id,
+            prowler_api_provider.id,
+            attack_paths_scan.id,
+        )
+        for old_attack_paths_scan in old_attack_paths_scans:
+            graph_database.drop_database(old_attack_paths_scan.graph_database)
+            db_utils.update_old_attack_paths_scan(old_attack_paths_scan)
+
+        db_utils.finish_attack_paths_scan(
+            attack_paths_scan, StateChoices.COMPLETED, ingestion_exceptions
+        )
+        return ingestion_exceptions
+
+    except Exception as e:
+        exception_message = utils.stringify_exception(e, "Cartography failed")
+        logger.error(exception_message)
+        ingestion_exceptions["global_cartography_error"] = exception_message
+
+        # Handling databases changes
+        graph_database.drop_database(cartography_config.neo4j_database)
+        db_utils.finish_attack_paths_scan(
+            attack_paths_scan, StateChoices.FAILED, ingestion_exceptions
+        )
+        raise
+
+
+def _call_within_event_loop(fn, *args, **kwargs):
+    """
+    Cartography needs a running event loop, so assuming there is none (Celery task or even regular DRF endpoint),
+    let's create a new one and set it as the current event loop for this thread.
+    """
+
+    loop = asyncio.new_event_loop()
+    try:
+        asyncio.set_event_loop(loop)
+        return fn(*args, **kwargs)
+
+    finally:
+        try:
+            loop.run_until_complete(loop.shutdown_asyncgens())
+
+        except Exception as e:
+            logger.warning(f"Failed to shutdown async generators cleanly: {e}")
+
+        loop.close()
+        asyncio.set_event_loop(None)

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

@@ -0,0 +1,10 @@
+import traceback
+
+from datetime import datetime, timezone
+
+
+def stringify_exception(exception: Exception, context: str) -> str:
+    timestamp = datetime.now(tz=timezone.utc)
+    exception_traceback = traceback.TracebackException.from_exception(exception)
+    traceback_string = "".join(exception_traceback.format())
+    return f"{timestamp} - {context}\n{traceback_string}"

api/src/backend/tasks/jobs/deletion.py

@@ -1,9 +1,19 @@
 from celery.utils.log import get_task_logger
 from django.db import DatabaseError
 
+from api.attack_paths import database as graph_database
 from api.db_router import MainRouter
 from api.db_utils import batch_delete, rls_transaction
-from api.models import Finding, Provider, Resource, Scan, ScanSummary, Tenant
+from api.models import (
+    AttackPathsScan,
+    Finding,
+    Provider,
+    Resource,
+    Scan,
+    ScanSummary,
+    Tenant,
+)
+from tasks.jobs.attack_paths.db_utils import get_provider_graph_database_names
 
 logger = get_task_logger(__name__)
 
@@ -23,16 +33,27 @@ def delete_provider(tenant_id: str, pk: str):
     Raises:
         Provider.DoesNotExist: If no instance with the provided primary key exists.
     """
+    # Delete the Attack Paths' graph databases related to the provider
+    graph_database_names = get_provider_graph_database_names(tenant_id, pk)
+    try:
+        for graph_database_name in graph_database_names:
+            graph_database.drop_database(graph_database_name)
+    except graph_database.GraphDatabaseQueryException as gdb_error:
+        logger.error(f"Error deleting Provider databases: {gdb_error}")
+        raise
+
+    # Get all provider related data and delete them in batches
     with rls_transaction(tenant_id):
         instance = Provider.all_objects.get(pk=pk)
-        deletion_summary = {}
         deletion_steps = [
             ("Scan Summaries", ScanSummary.all_objects.filter(scan__provider=instance)),
             ("Findings", Finding.all_objects.filter(scan__provider=instance)),
             ("Resources", Resource.all_objects.filter(provider=instance)),
             ("Scans", Scan.all_objects.filter(provider=instance)),
+            ("AttackPathsScans", AttackPathsScan.all_objects.filter(provider=instance)),
         ]
 
+    deletion_summary = {}
     for step_name, queryset in deletion_steps:
         try:
             _, step_summary = batch_delete(tenant_id, queryset)
@@ -48,6 +69,7 @@ def delete_provider(tenant_id: str, pk: str):
     except DatabaseError as db_error:
         logger.error(f"Error deleting Provider: {db_error}")
         raise
+
     return deletion_summary
 
 

api/src/backend/tasks/tasks.py

@@ -1,13 +1,29 @@
 import os
+
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from shutil import rmtree
 
 from celery import chain, group, shared_task
 from celery.utils.log import get_task_logger
+from django_celery_beat.models import PeriodicTask
+
+from api.compliance import get_compliance_frameworks
+from api.db_router import READ_REPLICA_ALIAS
+from api.db_utils import rls_transaction
+from api.decorators import handle_provider_deletion, set_tenant
+from api.models import Finding, Integration, Provider, Scan, ScanSummary, StateChoices
+from api.utils import initialize_prowler_provider
+from api.v1.serializers import ScanTaskSerializer
 from config.celery import RLSTask
 from config.django.base import DJANGO_FINDINGS_BATCH_SIZE, DJANGO_TMP_OUTPUT_DIRECTORY
-from django_celery_beat.models import PeriodicTask
+from prowler.lib.check.compliance_models import Compliance
+from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
+from prowler.lib.outputs.finding import Finding as FindingOutput
+from tasks.jobs.attack_paths import (
+    attack_paths_scan,
+    can_provider_run_attack_paths_scan,
+)
 from tasks.jobs.backfill import (
     backfill_compliance_summaries,
     backfill_daily_severity_summaries,
@@ -50,17 +66,6 @@ from tasks.jobs.scan import (
 )
 from tasks.utils import batched, get_next_execution_datetime
 
-from api.compliance import get_compliance_frameworks
-from api.db_router import READ_REPLICA_ALIAS
-from api.db_utils import rls_transaction
-from api.decorators import handle_provider_deletion, set_tenant
-from api.models import Finding, Integration, Provider, Scan, ScanSummary, StateChoices
-from api.utils import initialize_prowler_provider
-from api.v1.serializers import ScanTaskSerializer
-from prowler.lib.check.compliance_models import Compliance
-from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
-from prowler.lib.outputs.finding import Finding as FindingOutput
-
 logger = get_task_logger(__name__)
 
 
@@ -153,6 +158,11 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str)
         ),
     ).apply_async()
 
+    if can_provider_run_attack_paths_scan(tenant_id, provider_id):
+        perform_attack_paths_scan_task.apply_async(
+            kwargs={"tenant_id": tenant_id, "scan_id": scan_id}
+        )
+
 
 @shared_task(base=RLSTask, name="provider-connection-check")
 @set_tenant
@@ -358,6 +368,25 @@ def perform_scan_summary_task(tenant_id: str, scan_id: str):
     return aggregate_findings(tenant_id=tenant_id, scan_id=scan_id)
 
 
+# TODO: This task must be queued at the `attack-paths` queue, don't forget to add it to the `docker-entrypoint.sh` file
+@shared_task(base=RLSTask, bind=True, name="attack-paths-scan-perform", queue="scans")
+def perform_attack_paths_scan_task(self, tenant_id: str, scan_id: str):
+    """
+    Execute an Attack Paths scan for the given provider within the current tenant RLS context.
+
+    Args:
+        self: The task instance (automatically passed when bind=True).
+        tenant_id (str): The tenant identifier for RLS context.
+        scan_id (str): The Prowler scan identifier for obtaining the tenant and provider context.
+
+    Returns:
+        Any: The result from `attack_paths_scan`, including any per-scan failure details.
+    """
+    return attack_paths_scan(
+        tenant_id=tenant_id, scan_id=scan_id, task_id=self.request.id
+    )
+
+
 @shared_task(name="tenant-deletion", queue="deletion", autoretry_for=(Exception,))
 def delete_tenant_task(tenant_id: str):
     return delete_tenant(pk=tenant_id)

api/src/backend/tasks/tests/test_attack_paths_scan.py

@@ -0,0 +1,416 @@
+from contextlib import nullcontext
+from types import SimpleNamespace
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+from api.models import (
+    AttackPathsScan,
+    Finding,
+    Provider,
+    Resource,
+    ResourceFindingMapping,
+    Scan,
+    StateChoices,
+    StatusChoices,
+)
+from prowler.lib.check.models import Severity
+from tasks.jobs.attack_paths import prowler as prowler_module
+from tasks.jobs.attack_paths.scan import run as attack_paths_run
+
+
+@pytest.mark.django_db
+class TestAttackPathsRun:
+    def test_run_success_flow(self, tenants_fixture, providers_fixture, scans_fixture):
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+        scan = scans_fixture[0]
+        scan.provider = provider
+        scan.save()
+
+        attack_paths_scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider,
+            scan=scan,
+            state=StateChoices.SCHEDULED,
+        )
+
+        mock_session = MagicMock()
+        session_ctx = MagicMock()
+        session_ctx.__enter__.return_value = mock_session
+        session_ctx.__exit__.return_value = False
+        ingestion_result = {"organizations": "warning"}
+        ingestion_fn = MagicMock(return_value=ingestion_result)
+
+        with (
+            patch(
+                "tasks.jobs.attack_paths.scan.rls_transaction",
+                new=lambda *args, **kwargs: nullcontext(),
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.initialize_prowler_provider",
+                return_value=MagicMock(_enabled_regions=["us-east-1"]),
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.graph_database.get_uri",
+                return_value="bolt://neo4j",
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.graph_database.get_database_name",
+                return_value="db-scan-id",
+            ) as mock_get_db_name,
+            patch(
+                "tasks.jobs.attack_paths.scan.graph_database.create_database"
+            ) as mock_create_db,
+            patch(
+                "tasks.jobs.attack_paths.scan.graph_database.get_session",
+                return_value=session_ctx,
+            ) as mock_get_session,
+            patch(
+                "tasks.jobs.attack_paths.scan.cartography_create_indexes.run"
+            ) as mock_cartography_indexes,
+            patch(
+                "tasks.jobs.attack_paths.scan.cartography_analysis.run"
+            ) as mock_cartography_analysis,
+            patch(
+                "tasks.jobs.attack_paths.scan.cartography_ontology.run"
+            ) as mock_cartography_ontology,
+            patch(
+                "tasks.jobs.attack_paths.scan.prowler.create_indexes"
+            ) as mock_prowler_indexes,
+            patch(
+                "tasks.jobs.attack_paths.scan.prowler.analysis"
+            ) as mock_prowler_analysis,
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.retrieve_attack_paths_scan",
+                return_value=attack_paths_scan,
+            ) as mock_retrieve_scan,
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan"
+            ) as mock_starting,
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress"
+            ) as mock_update_progress,
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan"
+            ) as mock_finish,
+            patch(
+                "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function",
+                return_value=ingestion_fn,
+            ) as mock_get_ingestion,
+            patch(
+                "tasks.jobs.attack_paths.scan._call_within_event_loop",
+                side_effect=lambda fn, *a, **kw: fn(*a, **kw),
+            ) as mock_event_loop,
+        ):
+            result = attack_paths_run(str(tenant.id), str(scan.id), "task-123")
+
+        assert result == ingestion_result
+        mock_retrieve_scan.assert_called_once_with(str(tenant.id), str(scan.id))
+        mock_starting.assert_called_once()
+        config = mock_starting.call_args[0][2]
+        assert config.neo4j_database == "db-scan-id"
+
+        mock_create_db.assert_called_once_with("db-scan-id")
+        mock_get_session.assert_called_once_with("db-scan-id")
+        mock_cartography_indexes.assert_called_once_with(mock_session, config)
+        mock_prowler_indexes.assert_called_once_with(mock_session)
+        mock_cartography_analysis.assert_called_once_with(mock_session, config)
+        mock_cartography_ontology.assert_called_once_with(mock_session, config)
+        mock_prowler_analysis.assert_called_once_with(
+            mock_session,
+            provider,
+            str(scan.id),
+            config,
+        )
+        mock_get_ingestion.assert_called_once_with(provider.provider)
+        mock_event_loop.assert_called_once()
+        mock_update_progress.assert_any_call(attack_paths_scan, 1)
+        mock_update_progress.assert_any_call(attack_paths_scan, 2)
+        mock_update_progress.assert_any_call(attack_paths_scan, 95)
+        mock_finish.assert_called_once_with(
+            attack_paths_scan, StateChoices.COMPLETED, ingestion_result
+        )
+        mock_get_db_name.assert_called_once_with(attack_paths_scan.id)
+
+    def test_run_failure_marks_scan_failed(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+        scan = scans_fixture[0]
+        scan.provider = provider
+        scan.save()
+
+        attack_paths_scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider,
+            scan=scan,
+            state=StateChoices.SCHEDULED,
+        )
+
+        mock_session = MagicMock()
+        session_ctx = MagicMock()
+        session_ctx.__enter__.return_value = mock_session
+        session_ctx.__exit__.return_value = False
+        ingestion_fn = MagicMock(side_effect=RuntimeError("ingestion boom"))
+
+        with (
+            patch(
+                "tasks.jobs.attack_paths.scan.rls_transaction",
+                new=lambda *args, **kwargs: nullcontext(),
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.initialize_prowler_provider",
+                return_value=MagicMock(_enabled_regions=["us-east-1"]),
+            ),
+            patch("tasks.jobs.attack_paths.scan.graph_database.get_uri"),
+            patch(
+                "tasks.jobs.attack_paths.scan.graph_database.get_database_name",
+                return_value="db-scan-id",
+            ),
+            patch("tasks.jobs.attack_paths.scan.graph_database.create_database"),
+            patch(
+                "tasks.jobs.attack_paths.scan.graph_database.get_session",
+                return_value=session_ctx,
+            ),
+            patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run"),
+            patch("tasks.jobs.attack_paths.scan.cartography_analysis.run"),
+            patch("tasks.jobs.attack_paths.scan.prowler.create_indexes"),
+            patch("tasks.jobs.attack_paths.scan.prowler.analysis"),
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.retrieve_attack_paths_scan",
+                return_value=attack_paths_scan,
+            ),
+            patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan"),
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress"
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan"
+            ) as mock_finish,
+            patch(
+                "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function",
+                return_value=ingestion_fn,
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan._call_within_event_loop",
+                side_effect=lambda fn, *a, **kw: fn(*a, **kw),
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.utils.stringify_exception",
+                return_value="Cartography failed: ingestion boom",
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="ingestion boom"):
+                attack_paths_run(str(tenant.id), str(scan.id), "task-456")
+
+        failure_args = mock_finish.call_args[0]
+        assert failure_args[0] is attack_paths_scan
+        assert failure_args[1] == StateChoices.FAILED
+        assert failure_args[2] == {
+            "global_cartography_error": "Cartography failed: ingestion boom"
+        }
+
+    def test_run_returns_early_for_unsupported_provider(self, tenants_fixture):
+        tenant = tenants_fixture[0]
+        provider = Provider.objects.create(
+            provider=Provider.ProviderChoices.GCP,
+            uid="gcp-account",
+            alias="gcp",
+            tenant_id=tenant.id,
+        )
+        scan = Scan.objects.create(
+            name="GCP Scan",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.AVAILABLE,
+            tenant_id=tenant.id,
+        )
+
+        with (
+            patch(
+                "tasks.jobs.attack_paths.scan.rls_transaction",
+                new=lambda *args, **kwargs: nullcontext(),
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.initialize_prowler_provider",
+                return_value=MagicMock(),
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function",
+                return_value=None,
+            ) as mock_get_ingestion,
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.retrieve_attack_paths_scan"
+            ) as mock_retrieve,
+        ):
+            mock_retrieve.return_value = None
+            result = attack_paths_run(str(tenant.id), str(scan.id), "task-789")
+
+        assert result == {
+            "global_error": "Provider gcp is not supported for Attack Paths scans"
+        }
+        mock_get_ingestion.assert_called_once_with(provider.provider)
+        mock_retrieve.assert_called_once_with(str(tenant.id), str(scan.id))
+
+
+@pytest.mark.django_db
+class TestAttackPathsProwlerHelpers:
+    def test_create_indexes_executes_all_statements(self):
+        mock_session = MagicMock()
+        with patch("tasks.jobs.attack_paths.prowler.run_write_query") as mock_run_write:
+            prowler_module.create_indexes(mock_session)
+
+        assert mock_run_write.call_count == len(prowler_module.INDEX_STATEMENTS)
+        mock_run_write.assert_has_calls(
+            [call(mock_session, stmt) for stmt in prowler_module.INDEX_STATEMENTS]
+        )
+
+    def test_load_findings_batches_requests(self, providers_fixture):
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+
+        findings = [
+            {"id": "1", "resource_uid": "r-1"},
+            {"id": "2", "resource_uid": "r-2"},
+        ]
+        config = SimpleNamespace(update_tag=12345)
+        mock_session = MagicMock()
+
+        with (
+            patch.object(prowler_module, "BATCH_SIZE", 1),
+            patch(
+                "tasks.jobs.attack_paths.prowler.get_root_node_label",
+                return_value="AWSAccount",
+            ),
+            patch(
+                "tasks.jobs.attack_paths.prowler.get_node_uid_field",
+                return_value="arn",
+            ),
+        ):
+            prowler_module.load_findings(mock_session, findings, provider, config)
+
+        assert mock_session.run.call_count == 2
+        for call_args in mock_session.run.call_args_list:
+            params = call_args.args[1]
+            assert params["provider_uid"] == str(provider.uid)
+            assert params["last_updated"] == config.update_tag
+            assert "findings_data" in params
+
+    def test_cleanup_findings_runs_batches(self, providers_fixture):
+        provider = providers_fixture[0]
+        config = SimpleNamespace(update_tag=1024)
+        mock_session = MagicMock()
+
+        first_batch = MagicMock()
+        first_batch.single.return_value = {"deleted_findings_count": 3}
+        second_batch = MagicMock()
+        second_batch.single.return_value = {"deleted_findings_count": 0}
+        mock_session.run.side_effect = [first_batch, second_batch]
+
+        prowler_module.cleanup_findings(mock_session, provider, config)
+
+        assert mock_session.run.call_count == 2
+        params = mock_session.run.call_args.args[1]
+        assert params["provider_uid"] == str(provider.uid)
+        assert params["last_updated"] == config.update_tag
+
+    def test_get_provider_last_scan_findings_returns_latest_scan_data(
+        self,
+        tenants_fixture,
+        providers_fixture,
+    ):
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+
+        resource = Resource.objects.create(
+            tenant_id=tenant.id,
+            provider=provider,
+            uid="resource-uid",
+            name="Resource",
+            region="us-east-1",
+            service="ec2",
+            type="instance",
+        )
+
+        older_scan = Scan.objects.create(
+            name="Older",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant_id=tenant.id,
+        )
+        old_finding = Finding.objects.create(
+            tenant_id=tenant.id,
+            uid="older-finding",
+            scan=older_scan,
+            delta=Finding.DeltaChoices.NEW,
+            status=StatusChoices.PASS,
+            status_extended="ok",
+            severity=Severity.low,
+            impact=Severity.low,
+            impact_extended="",
+            raw_result={},
+            check_id="check-old",
+            check_metadata={"checktitle": "Old"},
+            first_seen_at=older_scan.inserted_at,
+        )
+        ResourceFindingMapping.objects.create(
+            tenant_id=tenant.id,
+            resource=resource,
+            finding=old_finding,
+        )
+
+        latest_scan = Scan.objects.create(
+            name="Latest",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant_id=tenant.id,
+        )
+        finding = Finding.objects.create(
+            tenant_id=tenant.id,
+            uid="finding-uid",
+            scan=latest_scan,
+            delta=Finding.DeltaChoices.NEW,
+            status=StatusChoices.FAIL,
+            status_extended="failed",
+            severity=Severity.high,
+            impact=Severity.high,
+            impact_extended="",
+            raw_result={},
+            check_id="check-1",
+            check_metadata={"checktitle": "Check title"},
+            first_seen_at=latest_scan.inserted_at,
+        )
+        ResourceFindingMapping.objects.create(
+            tenant_id=tenant.id,
+            resource=resource,
+            finding=finding,
+        )
+
+        latest_scan.refresh_from_db()
+
+        with patch(
+            "tasks.jobs.attack_paths.prowler.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            findings_data = prowler_module.get_provider_last_scan_findings(
+                provider,
+                str(latest_scan.id),
+            )
+
+        assert len(findings_data) == 1
+        finding_dict = findings_data[0]
+        assert finding_dict["id"] == str(finding.id)
+        assert finding_dict["resource_uid"] == resource.uid
+        assert finding_dict["check_title"] == "Check title"
+        assert finding_dict["scan_id"] == str(latest_scan.id)

api/src/backend/tasks/tests/test_connection.py

@@ -82,7 +82,7 @@ def test_check_provider_connection_exception(
     [
         {
             "name": "OpenAI",
-            "api_key_decoded": "sk-test1234567890T3BlbkFJtest1234567890",
+            "api_key_decoded": "sk-fake-test-key-for-unit-testing-only",
             "model": "gpt-4o",
             "temperature": 0,
             "max_tokens": 4000,

api/src/backend/tasks/tests/test_deletion.py

@@ -1,27 +1,60 @@
+from unittest.mock import call, patch
+
 import pytest
+
 from django.core.exceptions import ObjectDoesNotExist
-from tasks.jobs.deletion import delete_provider, delete_tenant
 
 from api.models import Provider, Tenant
+from tasks.jobs.deletion import delete_provider, delete_tenant
 
 
 @pytest.mark.django_db
 class TestDeleteProvider:
     def test_delete_provider_success(self, providers_fixture):
-        instance = providers_fixture[0]
-        tenant_id = str(instance.tenant_id)
-        result = delete_provider(tenant_id, instance.id)
+        with patch(
+            "tasks.jobs.deletion.get_provider_graph_database_names"
+        ) as mock_get_provider_graph_database_names, patch(
+            "tasks.jobs.deletion.graph_database.drop_database"
+        ) as mock_drop_database:
+            graph_db_names = ["graph-db-1", "graph-db-2"]
+            mock_get_provider_graph_database_names.return_value = graph_db_names
 
-        assert result
-        with pytest.raises(ObjectDoesNotExist):
-            Provider.objects.get(pk=instance.id)
+            instance = providers_fixture[0]
+            tenant_id = str(instance.tenant_id)
+            result = delete_provider(tenant_id, instance.id)
+
+            assert result
+            with pytest.raises(ObjectDoesNotExist):
+                Provider.objects.get(pk=instance.id)
+
+            mock_get_provider_graph_database_names.assert_called_once_with(
+                tenant_id, instance.id
+            )
+            mock_drop_database.assert_has_calls(
+                [call(graph_db_name) for graph_db_name in graph_db_names]
+            )
 
     def test_delete_provider_does_not_exist(self, tenants_fixture):
-        tenant_id = str(tenants_fixture[0].id)
-        non_existent_pk = "babf6796-cfcc-4fd3-9dcf-88d012247645"
+        with patch(
+            "tasks.jobs.deletion.get_provider_graph_database_names"
+        ) as mock_get_provider_graph_database_names, patch(
+            "tasks.jobs.deletion.graph_database.drop_database"
+        ) as mock_drop_database:
+            graph_db_names = ["graph-db-1"]
+            mock_get_provider_graph_database_names.return_value = graph_db_names
 
-        with pytest.raises(ObjectDoesNotExist):
-            delete_provider(tenant_id, non_existent_pk)
+            tenant_id = str(tenants_fixture[0].id)
+            non_existent_pk = "babf6796-cfcc-4fd3-9dcf-88d012247645"
+
+            with pytest.raises(ObjectDoesNotExist):
+                delete_provider(tenant_id, non_existent_pk)
+
+            mock_get_provider_graph_database_names.assert_called_once_with(
+                tenant_id, non_existent_pk
+            )
+            mock_drop_database.assert_has_calls(
+                [call(graph_db_name) for graph_db_name in graph_db_names]
+            )
 
 
 @pytest.mark.django_db
@@ -30,33 +63,68 @@ class TestDeleteTenant:
         """
         Test successful deletion of a tenant and its related data.
         """
-        tenant = tenants_fixture[0]
-        providers = Provider.objects.filter(tenant_id=tenant.id)
+        with patch(
+            "tasks.jobs.deletion.get_provider_graph_database_names"
+        ) as mock_get_provider_graph_database_names, patch(
+            "tasks.jobs.deletion.graph_database.drop_database"
+        ) as mock_drop_database:
+            tenant = tenants_fixture[0]
+            providers = list(Provider.objects.filter(tenant_id=tenant.id))
 
-        # Ensure the tenant and related providers exist before deletion
-        assert Tenant.objects.filter(id=tenant.id).exists()
-        assert providers.exists()
+            graph_db_names_per_provider = [
+                [f"graph-db-{provider.id}"] for provider in providers
+            ]
+            mock_get_provider_graph_database_names.side_effect = (
+                graph_db_names_per_provider
+            )
 
-        # Call the function and validate the result
-        deletion_summary = delete_tenant(tenant.id)
+            # Ensure the tenant and related providers exist before deletion
+            assert Tenant.objects.filter(id=tenant.id).exists()
+            assert providers
 
-        assert deletion_summary is not None
-        assert not Tenant.objects.filter(id=tenant.id).exists()
-        assert not Provider.objects.filter(tenant_id=tenant.id).exists()
+            # Call the function and validate the result
+            deletion_summary = delete_tenant(tenant.id)
+
+            assert deletion_summary is not None
+            assert not Tenant.objects.filter(id=tenant.id).exists()
+            assert not Provider.objects.filter(tenant_id=tenant.id).exists()
+
+            expected_calls = [
+                call(provider.tenant_id, provider.id) for provider in providers
+            ]
+            mock_get_provider_graph_database_names.assert_has_calls(
+                expected_calls, any_order=True
+            )
+            assert mock_get_provider_graph_database_names.call_count == len(
+                expected_calls
+            )
+            expected_drop_calls = [
+                call(graph_db_name[0]) for graph_db_name in graph_db_names_per_provider
+            ]
+            mock_drop_database.assert_has_calls(expected_drop_calls, any_order=True)
+            assert mock_drop_database.call_count == len(expected_drop_calls)
 
     def test_delete_tenant_with_no_providers(self, tenants_fixture):
         """
         Test deletion of a tenant with no related providers.
         """
-        tenant = tenants_fixture[1]  # Assume this tenant has no providers
-        providers = Provider.objects.filter(tenant_id=tenant.id)
+        with patch(
+            "tasks.jobs.deletion.get_provider_graph_database_names"
+        ) as mock_get_provider_graph_database_names, patch(
+            "tasks.jobs.deletion.graph_database.drop_database"
+        ) as mock_drop_database:
+            tenant = tenants_fixture[1]  # Assume this tenant has no providers
+            providers = Provider.objects.filter(tenant_id=tenant.id)
 
-        # Ensure the tenant exists but has no related providers
-        assert Tenant.objects.filter(id=tenant.id).exists()
-        assert not providers.exists()
+            # Ensure the tenant exists but has no related providers
+            assert Tenant.objects.filter(id=tenant.id).exists()
+            assert not providers.exists()
 
-        # Call the function and validate the result
-        deletion_summary = delete_tenant(tenant.id)
+            # Call the function and validate the result
+            deletion_summary = delete_tenant(tenant.id)
 
-        assert deletion_summary == {}  # No providers, so empty summary
-        assert not Tenant.objects.filter(id=tenant.id).exists()
+            assert deletion_summary == {}  # No providers, so empty summary
+            assert not Tenant.objects.filter(id=tenant.id).exists()
+
+            mock_get_provider_graph_database_names.assert_not_called()
+            mock_drop_database.assert_not_called()

api/src/backend/tasks/tests/test_tasks.py

@@ -1,10 +1,21 @@
 import uuid
+
+from contextlib import contextmanager
 from unittest.mock import MagicMock, patch
 
 import openai
 import pytest
+
 from botocore.exceptions import ClientError
 from django_celery_beat.models import IntervalSchedule, PeriodicTask
+
+from api.models import (
+    Integration,
+    LighthouseProviderConfiguration,
+    LighthouseProviderModels,
+    Scan,
+    StateChoices,
+)
 from tasks.jobs.lighthouse_providers import (
     _create_bedrock_client,
     _extract_bedrock_credentials,
@@ -15,19 +26,12 @@ from tasks.tasks import (
     check_integrations_task,
     check_lighthouse_provider_connection_task,
     generate_outputs_task,
+    perform_attack_paths_scan_task,
     refresh_lighthouse_provider_models_task,
     s3_integration_task,
     security_hub_integration_task,
 )
 
-from api.models import (
-    Integration,
-    LighthouseProviderConfiguration,
-    LighthouseProviderModels,
-    Scan,
-    StateChoices,
-)
-
 
 @pytest.mark.django_db
 class TestExtractBedrockCredentials:
@@ -737,8 +741,12 @@ class TestScanCompleteTasks:
     @patch("tasks.tasks.generate_outputs_task.si")
     @patch("tasks.tasks.generate_compliance_reports_task.si")
     @patch("tasks.tasks.check_integrations_task.si")
+    @patch("tasks.tasks.perform_attack_paths_scan_task.apply_async")
+    @patch("tasks.tasks.can_provider_run_attack_paths_scan", return_value=False)
     def test_scan_complete_tasks(
         self,
+        mock_can_run_attack_paths,
+        mock_attack_paths_task,
         mock_check_integrations_task,
         mock_compliance_reports_task,
         mock_outputs_task,
@@ -793,6 +801,67 @@ class TestScanCompleteTasks:
             scan_id="scan-id",
         )
 
+        # Attack Paths task should be skipped when provider cannot run it
+        mock_attack_paths_task.assert_not_called()
+
+
+class TestAttackPathsTasks:
+    @staticmethod
+    @contextmanager
+    def _override_task_request(task, **attrs):
+        request = task.request
+        sentinel = object()
+        previous = {key: getattr(request, key, sentinel) for key in attrs}
+        for key, value in attrs.items():
+            setattr(request, key, value)
+
+        try:
+            yield
+        finally:
+            for key, prev in previous.items():
+                if prev is sentinel:
+                    if hasattr(request, key):
+                        delattr(request, key)
+                else:
+                    setattr(request, key, prev)
+
+    def test_perform_attack_paths_scan_task_calls_runner(self):
+        with (
+            patch("tasks.tasks.attack_paths_scan") as mock_attack_paths_scan,
+            self._override_task_request(
+                perform_attack_paths_scan_task, id="celery-task-id"
+            ),
+        ):
+            mock_attack_paths_scan.return_value = {"status": "ok"}
+
+            result = perform_attack_paths_scan_task.run(
+                tenant_id="tenant-id", scan_id="scan-id"
+            )
+
+        mock_attack_paths_scan.assert_called_once_with(
+            tenant_id="tenant-id", scan_id="scan-id", task_id="celery-task-id"
+        )
+        assert result == {"status": "ok"}
+
+    def test_perform_attack_paths_scan_task_propagates_exception(self):
+        with (
+            patch(
+                "tasks.tasks.attack_paths_scan",
+                side_effect=RuntimeError("Exception to propagate"),
+            ) as mock_attack_paths_scan,
+            self._override_task_request(
+                perform_attack_paths_scan_task, id="celery-task-error"
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="Exception to propagate"):
+                perform_attack_paths_scan_task.run(
+                    tenant_id="tenant-id", scan_id="scan-id"
+                )
+
+        mock_attack_paths_scan.assert_called_once_with(
+            tenant_id="tenant-id", scan_id="scan-id", task_id="celery-task-error"
+        )
+
 
 @pytest.mark.django_db
 class TestCheckIntegrationsTask: