Merge remote-tracking branch 'origin/master' into feat/PROWLER-940-stage-2-a-image-provider-api-v3

# Conflicts:
#	poetry.lock
#	prowler/providers/image/exceptions/exceptions.py
#	prowler/providers/image/image_provider.py
#	prowler/providers/image/lib/arguments/arguments.py
#	tests/providers/image/image_fixtures.py
#	tests/providers/image/image_provider_test.py

api/src/backend/api/attack_paths/queries/types.py

@@ -1,6 +1,14 @@
 from dataclasses import dataclass, field
 
 
+@dataclass
+class AttackPathsQueryAttribution:
+    """Source attribution for an Attack Path query."""
+
+    text: str
+    link: str
+
+
 @dataclass
 class AttackPathsQueryParameterDefinition:
     """
@@ -23,7 +31,9 @@ class AttackPathsQueryDefinition:
 
     id: str
     name: str
+    short_description: str
     description: str
     provider: str
     cypher: str
+    attribution: AttackPathsQueryAttribution | None = None
     parameters: list[AttackPathsQueryParameterDefinition] = field(default_factory=list)

api/src/backend/api/attack_paths/views_helpers.py

@@ -5,7 +5,6 @@ from typing import Any, Iterable
 from rest_framework.exceptions import APIException, ValidationError
 
 from api.attack_paths import database as graph_database, AttackPathsQueryDefinition
-from api.models import AttackPathsScan
 from config.custom_logging import BackendLogger
 from tasks.jobs.attack_paths.config import INTERNAL_LABELS
 
@@ -80,12 +79,12 @@ def prepare_query_parameters(
 
 
 def execute_attack_paths_query(
-    attack_paths_scan: AttackPathsScan,
+    database_name: str,
     definition: AttackPathsQueryDefinition,
     parameters: dict[str, Any],
 ) -> dict[str, Any]:
     try:
-        with graph_database.get_session(attack_paths_scan.graph_database) as session:
+        with graph_database.get_session(database_name) as session:
             result = session.run(definition.cypher, parameters)
             return _serialize_graph(result.graph())
 

api/src/backend/api/fixtures/dev/8_dev_attack_paths_scans.json

@@ -7,10 +7,9 @@
       "provider": "b85601a8-4b45-4194-8135-03fb980ef428",
       "scan": "01920573-aa9c-73c9-bcda-f2e35c9b19d2",
       "state": "completed",
+      "graph_data_ready": true,
       "progress": 100,
       "update_tag": 1693586667,
-      "graph_database": "db-a7f0f6de-6f8e-4b3a-8cbe-3f6dd9012345",
-      "is_graph_database_deleted": false,
       "task": null,
       "inserted_at": "2024-09-01T17:24:37Z",
       "updated_at": "2024-09-01T17:44:37Z",
@@ -30,8 +29,6 @@
       "state": "executing",
       "progress": 48,
       "update_tag": 1697625000,
-      "graph_database": "db-4a2fb2af-8a60-4d7d-9cae-4ca65e098765",
-      "is_graph_database_deleted": false,
       "task": null,
       "inserted_at": "2024-10-18T10:55:57Z",
       "updated_at": "2024-10-18T10:56:15Z",

api/src/backend/api/migrations/0076_openstack_provider.py

@@ -0,0 +1,39 @@
+# Generated by Django migration for OpenStack provider support
+
+from django.db import migrations
+
+import api.db_utils
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0075_cloudflare_provider"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="provider",
+            name="provider",
+            field=api.db_utils.ProviderEnumField(
+                choices=[
+                    ("aws", "AWS"),
+                    ("azure", "Azure"),
+                    ("gcp", "GCP"),
+                    ("kubernetes", "Kubernetes"),
+                    ("m365", "M365"),
+                    ("github", "GitHub"),
+                    ("mongodbatlas", "MongoDB Atlas"),
+                    ("iac", "IaC"),
+                    ("oraclecloud", "Oracle Cloud Infrastructure"),
+                    ("alibabacloud", "Alibaba Cloud"),
+                    ("cloudflare", "Cloudflare"),
+                    ("openstack", "OpenStack"),
+                ],
+                default="aws",
+            ),
+        ),
+        migrations.RunSQL(
+            "ALTER TYPE provider ADD VALUE IF NOT EXISTS 'openstack';",
+            reverse_sql=migrations.RunSQL.noop,
+        ),
+    ]

api/src/backend/api/migrations/0077_remove_attackpathsscan_graph_database_indexes.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.1.15 on 2026-02-16 09:24
+
+from django.contrib.postgres.operations import RemoveIndexConcurrently
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    atomic = False
+
+    dependencies = [
+        ("api", "0076_openstack_provider"),
+    ]
+
+    operations = [
+        RemoveIndexConcurrently(
+            model_name="attackpathsscan",
+            name="aps_active_graph_idx",
+        ),
+        RemoveIndexConcurrently(
+            model_name="attackpathsscan",
+            name="aps_completed_graph_idx",
+        ),
+    ]

api/src/backend/api/migrations/0078_remove_attackpathsscan_graph_database_fields.py

@@ -0,0 +1,20 @@
+# Generated by Django 5.1.15 on 2026-02-16 09:24
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0077_remove_attackpathsscan_graph_database_indexes"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="attackpathsscan",
+            name="graph_database",
+        ),
+        migrations.RemoveField(
+            model_name="attackpathsscan",
+            name="is_graph_database_deleted",
+        ),
+    ]

api/src/backend/api/migrations/0079_attackpathsscan_graph_data_ready.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.1.15 on 2026-02-16 13:55
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0078_remove_attackpathsscan_graph_database_fields"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="attackpathsscan",
+            name="graph_data_ready",
+            field=models.BooleanField(default=False),
+        ),
+    ]

api/src/backend/api/migrations/0080_backfill_attack_paths_graph_data_ready.py

@@ -0,0 +1,26 @@
+# Separate from 0079 because psqlextra's schema editor runs AddField DDL and DML
+# on different database connections, causing a deadlock when combined with RunPython
+# in the same migration.
+
+from django.db import migrations
+
+from api.db_router import MainRouter
+
+
+def backfill_graph_data_ready(apps, schema_editor):
+    """Set graph_data_ready=True for all completed AttackPathsScan rows."""
+    AttackPathsScan = apps.get_model("api", "AttackPathsScan")
+    AttackPathsScan.objects.using(MainRouter.admin_db).filter(
+        state="completed",
+        graph_data_ready=False,
+    ).update(graph_data_ready=True)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0079_attackpathsscan_graph_data_ready"),
+    ]
+
+    operations = [
+        migrations.RunPython(backfill_graph_data_ready, migrations.RunPython.noop),
+    ]

api/src/backend/api/models.py

@@ -288,6 +288,7 @@ class Provider(RowLevelSecurityProtectedModel):
         ORACLECLOUD = "oraclecloud", _("Oracle Cloud Infrastructure")
         ALIBABACLOUD = "alibabacloud", _("Alibaba Cloud")
         CLOUDFLARE = "cloudflare", _("Cloudflare")
+        OPENSTACK = "openstack", _("OpenStack")
 
     @staticmethod
     def validate_aws_uid(value):
@@ -326,10 +327,13 @@ class Provider(RowLevelSecurityProtectedModel):
 
     @staticmethod
     def validate_gcp_uid(value):
-        if not re.match(r"^[a-z][a-z0-9-]{5,29}$", value):
+        # Standard format: 6-30 chars, starts with letter, lowercase + digits + hyphens
+        # Legacy App Engine format: domain.com:project-id
+        if not re.match(r"^([a-z][a-z0-9.-]*:)?[a-z][a-z0-9-]{5,29}$", value):
             raise ModelValidationError(
-                detail="GCP provider ID must be 6 to 30 characters, start with a letter, and contain only lowercase "
-                "letters, numbers, and hyphens.",
+                detail="GCP provider ID must be a valid project ID: 6 to 30 characters, start with a letter, "
+                "and contain only lowercase letters, numbers, and hyphens. "
+                "Legacy App Engine project IDs with a domain prefix (e.g., example.com:my-project) are also accepted.",
                 code="gcp-uid",
                 pointer="/data/attributes/uid",
             )
@@ -410,6 +414,15 @@ class Provider(RowLevelSecurityProtectedModel):
                 pointer="/data/attributes/uid",
             )
 
+    @staticmethod
+    def validate_openstack_uid(value):
+        if not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9._-]{0,254}$", value):
+            raise ModelValidationError(
+                detail="OpenStack provider ID must be a valid project ID (UUID or project name).",
+                code="openstack-uid",
+                pointer="/data/attributes/uid",
+            )
+
     id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
     inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
     updated_at = models.DateTimeField(auto_now=True, editable=False)
@@ -645,6 +658,7 @@ class AttackPathsScan(RowLevelSecurityProtectedModel):
 
     state = StateEnumField(choices=StateChoices.choices, default=StateChoices.AVAILABLE)
     progress = models.IntegerField(default=0)
+    graph_data_ready = models.BooleanField(default=False)
 
     # Timing
     started_at = models.DateTimeField(null=True, blank=True)
@@ -681,8 +695,6 @@ class AttackPathsScan(RowLevelSecurityProtectedModel):
     update_tag = models.BigIntegerField(
         null=True, blank=True, help_text="Cartography update tag (epoch)"
     )
-    graph_database = models.CharField(max_length=63, null=True, blank=True)
-    is_graph_database_deleted = models.BooleanField(default=False)
     ingestion_exceptions = models.JSONField(default=dict, null=True, blank=True)
 
     class Meta(RowLevelSecurityProtectedModel.Meta):
@@ -709,21 +721,6 @@ class AttackPathsScan(RowLevelSecurityProtectedModel):
                 fields=["tenant_id", "scan_id"],
                 name="aps_scan_lookup_idx",
             ),
-            models.Index(
-                fields=["tenant_id", "provider_id"],
-                name="aps_active_graph_idx",
-                include=["graph_database", "id"],
-                condition=Q(is_graph_database_deleted=False),
-            ),
-            models.Index(
-                fields=["tenant_id", "provider_id", "-completed_at"],
-                name="aps_completed_graph_idx",
-                include=["graph_database", "id"],
-                condition=Q(
-                    state=StateChoices.COMPLETED,
-                    is_graph_database_deleted=False,
-                ),
-            ),
         ]
 
     class JSONAPIMeta:

api/src/backend/api/tests/test_attack_paths.py

@@ -83,12 +83,12 @@ def test_execute_attack_paths_query_serializes_graph(
     definition = attack_paths_query_definition_factory(
         id="aws-rds",
         name="RDS",
+        short_description="Short desc",
         description="",
         cypher="MATCH (n) RETURN n",
         parameters=[],
     )
     parameters = {"provider_uid": "123"}
-    attack_paths_scan = SimpleNamespace(graph_database="tenant-db")
 
     node = attack_paths_graph_stub_classes.Node(
         element_id="node-1",
@@ -122,15 +122,17 @@ def test_execute_attack_paths_query_serializes_graph(
     session_ctx.__enter__.return_value = session
     session_ctx.__exit__.return_value = False
 
+    database_name = "db-tenant-test-tenant-id"
+
     with patch(
         "api.attack_paths.views_helpers.graph_database.get_session",
         return_value=session_ctx,
     ) as mock_get_session:
         result = views_helpers.execute_attack_paths_query(
-            attack_paths_scan, definition, parameters
+            database_name, definition, parameters
         )
 
-    mock_get_session.assert_called_once_with("tenant-db")
+    mock_get_session.assert_called_once_with(database_name)
     session.run.assert_called_once_with(definition.cypher, parameters)
     assert result["nodes"][0]["id"] == "node-1"
     assert result["nodes"][0]["properties"]["complex"]["items"][0] == "value"
@@ -143,11 +145,12 @@ def test_execute_attack_paths_query_wraps_graph_errors(
     definition = attack_paths_query_definition_factory(
         id="aws-rds",
         name="RDS",
+        short_description="Short desc",
         description="",
         cypher="MATCH (n) RETURN n",
         parameters=[],
     )
-    attack_paths_scan = SimpleNamespace(graph_database="tenant-db")
+    database_name = "db-tenant-test-tenant-id"
     parameters = {"provider_uid": "123"}
 
     class ExplodingContext:
@@ -166,7 +169,7 @@ def test_execute_attack_paths_query_wraps_graph_errors(
     ):
         with pytest.raises(APIException):
             views_helpers.execute_attack_paths_query(
-                attack_paths_scan, definition, parameters
+                database_name, definition, parameters
             )
 
     mock_logger.error.assert_called_once()

api/src/backend/api/tests/test_utils.py

@@ -27,6 +27,7 @@ from prowler.providers.iac.iac_provider import IacProvider
 from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
 from prowler.providers.m365.m365_provider import M365Provider
 from prowler.providers.mongodbatlas.mongodbatlas_provider import MongodbatlasProvider
+from prowler.providers.openstack.openstack_provider import OpenstackProvider
 from prowler.providers.oraclecloud.oraclecloud_provider import OraclecloudProvider
 
 
@@ -120,6 +121,7 @@ class TestReturnProwlerProvider:
             (Provider.ProviderChoices.IAC.value, IacProvider),
             (Provider.ProviderChoices.ALIBABACLOUD.value, AlibabacloudProvider),
             (Provider.ProviderChoices.CLOUDFLARE.value, CloudflareProvider),
+            (Provider.ProviderChoices.OPENSTACK.value, OpenstackProvider),
         ],
     )
     def test_return_prowler_provider(self, provider_type, expected_provider):
@@ -227,6 +229,10 @@ class TestGetProwlerProviderKwargs:
                 Provider.ProviderChoices.CLOUDFLARE.value,
                 {"filter_accounts": ["provider_uid"]},
             ),
+            (
+                Provider.ProviderChoices.OPENSTACK.value,
+                {},
+            ),
         ],
     )
     def test_get_prowler_provider_kwargs(self, provider_type, expected_extra_kwargs):

api/src/backend/api/tests/test_views.py

@@ -1079,6 +1079,11 @@ class TestProviderViewSet:
             [
                 {"provider": "aws", "uid": "111111111111", "alias": "test"},
                 {"provider": "gcp", "uid": "a12322-test54321", "alias": "test"},
+                {
+                    "provider": "gcp",
+                    "uid": "example.com:my-project-123456",
+                    "alias": "legacy-gcp",
+                },
                 {
                     "provider": "kubernetes",
                     "uid": "kubernetes-test-123456789",
@@ -1179,6 +1184,11 @@ class TestProviderViewSet:
                     "uid": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
                     "alias": "Cloudflare Account",
                 },
+                {
+                    "provider": "openstack",
+                    "uid": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
+                    "alias": "OpenStack Project",
+                },
             ]
         ),
     )
@@ -1198,6 +1208,11 @@ class TestProviderViewSet:
             [
                 {"provider": "aws", "uid": "111111111111", "alias": "test"},
                 {"provider": "gcp", "uid": "a12322-test54321", "alias": "test"},
+                {
+                    "provider": "gcp",
+                    "uid": "example.com:my-project-123456",
+                    "alias": "legacy-gcp",
+                },
                 {
                     "provider": "kubernetes",
                     "uid": "kubernetes-test-123456789",
@@ -1598,6 +1613,26 @@ class TestProviderViewSet:
                     "cloudflare-uid",
                     "uid",
                 ),
+                # OpenStack UID validation - starts with special character
+                (
+                    {
+                        "provider": "openstack",
+                        "uid": "-invalid-project",
+                        "alias": "test",
+                    },
+                    "openstack-uid",
+                    "uid",
+                ),
+                # OpenStack UID validation - too short (below min_length)
+                (
+                    {
+                        "provider": "openstack",
+                        "uid": "ab",
+                        "alias": "test",
+                    },
+                    "min_length",
+                    "uid",
+                ),
             ]
         ),
     )
@@ -1771,21 +1806,21 @@ class TestProviderViewSet:
                 (
                     "uid.icontains",
                     "1",
-                    9,
+                    10,
                 ),
                 ("alias", "aws_testing_1", 1),
                 ("alias.icontains", "aws", 2),
-                ("inserted_at", TODAY, 10),
+                ("inserted_at", TODAY, 11),
                 (
                     "inserted_at.gte",
                     "2024-01-01",
-                    10,
+                    11,
                 ),
                 ("inserted_at.lte", "2024-01-01", 0),
                 (
                     "updated_at.gte",
                     "2024-01-01",
-                    10,
+                    11,
                 ),
                 ("updated_at.lte", "2024-01-01", 0),
             ]
@@ -2392,6 +2427,15 @@ class TestProviderSecretViewSet:
                     "api_email": "user@example.com",
                 },
             ),
+            # OpenStack with clouds.yaml content
+            (
+                Provider.ProviderChoices.OPENSTACK.value,
+                ProviderSecret.TypeChoices.STATIC,
+                {
+                    "clouds_yaml_content": "clouds:\n  mycloud:\n    auth:\n      auth_url: https://openstack.example.com:5000/v3\n",
+                    "clouds_yaml_cloud": "mycloud",
+                },
+            ),
         ],
     )
     def test_provider_secrets_create_valid(
@@ -3830,6 +3874,7 @@ class TestAttackPathsScanViewSet:
             AttackPathsQueryDefinition(
                 id="aws-rds",
                 name="RDS inventory",
+                short_description="List account RDS assets.",
                 description="List account RDS assets",
                 provider=provider.provider,
                 cypher="MATCH (n) RETURN n",
@@ -3887,11 +3932,12 @@ class TestAttackPathsScanViewSet:
         attack_paths_scan = create_attack_paths_scan(
             provider,
             scan=scans_fixture[0],
-            graph_database="tenant-db",
+            graph_data_ready=True,
         )
         query_definition = AttackPathsQueryDefinition(
             id="aws-rds",
             name="RDS inventory",
+            short_description="List account RDS assets.",
             description="List account RDS assets",
             provider=provider.provider,
             cypher="MATCH (n) RETURN n",
@@ -3917,10 +3963,16 @@ class TestAttackPathsScanViewSet:
             ],
         }
 
+        expected_db_name = f"db-tenant-{attack_paths_scan.provider.tenant_id}"
+
         with (
             patch(
                 "api.v1.views.get_query_by_id", return_value=query_definition
             ) as mock_get_query,
+            patch(
+                "api.v1.views.graph_database.get_database_name",
+                return_value=expected_db_name,
+            ) as mock_get_db_name,
             patch(
                 "api.v1.views.attack_paths_views_helpers.prepare_query_parameters",
                 return_value=prepared_parameters,
@@ -3942,23 +3994,24 @@ class TestAttackPathsScanViewSet:
 
         assert response.status_code == status.HTTP_200_OK
         mock_get_query.assert_called_once_with("aws-rds")
+        mock_get_db_name.assert_called_once_with(attack_paths_scan.provider.tenant_id)
         mock_prepare.assert_called_once_with(
             query_definition,
             {},
             attack_paths_scan.provider.uid,
         )
         mock_execute.assert_called_once_with(
-            attack_paths_scan,
+            expected_db_name,
             query_definition,
             prepared_parameters,
         )
-        mock_clear_cache.assert_called_once_with(attack_paths_scan.graph_database)
+        mock_clear_cache.assert_called_once_with(expected_db_name)
         result = response.json()["data"]
         attributes = result["attributes"]
         assert attributes["nodes"] == graph_payload["nodes"]
         assert attributes["relationships"] == graph_payload["relationships"]
 
-    def test_run_attack_paths_query_requires_completed_scan(
+    def test_run_attack_paths_query_blocks_when_graph_data_not_ready(
         self,
         authenticated_client,
         providers_fixture,
@@ -3970,6 +4023,7 @@ class TestAttackPathsScanViewSet:
             provider,
             scan=scans_fixture[0],
             state=StateChoices.EXECUTING,
+            graph_data_ready=False,
         )
 
         response = authenticated_client.post(
@@ -3981,9 +4035,9 @@ class TestAttackPathsScanViewSet:
         )
 
         assert response.status_code == status.HTTP_400_BAD_REQUEST
-        assert "must be completed" in response.json()["errors"][0]["detail"]
+        assert "not available" in response.json()["errors"][0]["detail"]
 
-    def test_run_attack_paths_query_requires_graph_database(
+    def test_run_attack_paths_query_allows_executing_scan_when_graph_data_ready(
         self,
         authenticated_client,
         providers_fixture,
@@ -3994,19 +4048,100 @@ class TestAttackPathsScanViewSet:
         attack_paths_scan = create_attack_paths_scan(
             provider,
             scan=scans_fixture[0],
-            graph_database=None,
+            state=StateChoices.EXECUTING,
+            graph_data_ready=True,
+        )
+        query_definition = AttackPathsQueryDefinition(
+            id="aws-test",
+            name="Test",
+            short_description="Test query.",
+            description="Test query",
+            provider=provider.provider,
+            cypher="MATCH (n) RETURN n",
+            parameters=[],
         )
 
-        response = authenticated_client.post(
-            reverse(
-                "attack-paths-scans-queries-run", kwargs={"pk": attack_paths_scan.id}
+        with (
+            patch("api.v1.views.get_query_by_id", return_value=query_definition),
+            patch(
+                "api.v1.views.attack_paths_views_helpers.prepare_query_parameters",
+                return_value={"provider_uid": provider.uid},
             ),
-            data=self._run_payload(),
-            content_type=API_JSON_CONTENT_TYPE,
+            patch(
+                "api.v1.views.attack_paths_views_helpers.execute_attack_paths_query",
+                return_value={
+                    "nodes": [{"id": "n1", "labels": ["AWSAccount"], "properties": {}}],
+                    "relationships": [],
+                },
+            ),
+            patch("api.v1.views.graph_database.clear_cache"),
+            patch(
+                "api.v1.views.graph_database.get_database_name", return_value="db-test"
+            ),
+        ):
+            response = authenticated_client.post(
+                reverse(
+                    "attack-paths-scans-queries-run",
+                    kwargs={"pk": attack_paths_scan.id},
+                ),
+                data=self._run_payload("aws-test"),
+                content_type=API_JSON_CONTENT_TYPE,
+            )
+
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_run_attack_paths_query_allows_failed_scan_when_graph_data_ready(
+        self,
+        authenticated_client,
+        providers_fixture,
+        scans_fixture,
+        create_attack_paths_scan,
+    ):
+        provider = providers_fixture[0]
+        attack_paths_scan = create_attack_paths_scan(
+            provider,
+            scan=scans_fixture[0],
+            state=StateChoices.FAILED,
+            graph_data_ready=True,
+        )
+        query_definition = AttackPathsQueryDefinition(
+            id="aws-test",
+            name="Test",
+            short_description="Test query.",
+            description="Test query",
+            provider=provider.provider,
+            cypher="MATCH (n) RETURN n",
+            parameters=[],
         )
 
-        assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
-        assert "does not reference a graph database" in str(response.json())
+        with (
+            patch("api.v1.views.get_query_by_id", return_value=query_definition),
+            patch(
+                "api.v1.views.attack_paths_views_helpers.prepare_query_parameters",
+                return_value={"provider_uid": provider.uid},
+            ),
+            patch(
+                "api.v1.views.attack_paths_views_helpers.execute_attack_paths_query",
+                return_value={
+                    "nodes": [{"id": "n1", "labels": ["AWSAccount"], "properties": {}}],
+                    "relationships": [],
+                },
+            ),
+            patch("api.v1.views.graph_database.clear_cache"),
+            patch(
+                "api.v1.views.graph_database.get_database_name", return_value="db-test"
+            ),
+        ):
+            response = authenticated_client.post(
+                reverse(
+                    "attack-paths-scans-queries-run",
+                    kwargs={"pk": attack_paths_scan.id},
+                ),
+                data=self._run_payload("aws-test"),
+                content_type=API_JSON_CONTENT_TYPE,
+            )
+
+        assert response.status_code == status.HTTP_200_OK
 
     def test_run_attack_paths_query_unknown_query(
         self,
@@ -4019,6 +4154,7 @@ class TestAttackPathsScanViewSet:
         attack_paths_scan = create_attack_paths_scan(
             provider,
             scan=scans_fixture[0],
+            graph_data_ready=True,
         )
 
         with patch("api.v1.views.get_query_by_id", return_value=None):
@@ -4045,10 +4181,12 @@ class TestAttackPathsScanViewSet:
         attack_paths_scan = create_attack_paths_scan(
             provider,
             scan=scans_fixture[0],
+            graph_data_ready=True,
         )
         query_definition = AttackPathsQueryDefinition(
             id="aws-empty",
             name="empty",
+            short_description="",
             description="",
             provider=provider.provider,
             cypher="MATCH (n) RETURN n",
@@ -10841,25 +10979,20 @@ class TestTenantFinishACSView:
             assert "sso_saml_failed=true" in response.url
 
     def test_dispatch_skips_role_mapping_when_single_manage_account_user(
-        self, create_test_user, tenants_fixture, saml_setup, settings, monkeypatch
+        self,
+        create_test_user,
+        tenants_fixture,
+        admin_role_fixture,
+        saml_setup,
+        settings,
+        monkeypatch,
     ):
         """Test that role mapping is skipped when tenant has only one user with MANAGE_ACCOUNT role"""
         monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
         user = create_test_user
         tenant = tenants_fixture[0]
 
-        # Create a single role with manage_account=True for the user
-        admin_role = Role.objects.using(MainRouter.admin_db).create(
-            name="admin",
-            tenant=tenant,
-            manage_account=True,
-            manage_users=True,
-            manage_billing=True,
-            manage_providers=True,
-            manage_integrations=True,
-            manage_scans=True,
-            unlimited_visibility=True,
-        )
+        admin_role = admin_role_fixture
         UserRoleRelationship.objects.using(MainRouter.admin_db).create(
             user=user, role=admin_role, tenant_id=tenant.id
         )
@@ -10930,35 +11063,26 @@ class TestTenantFinishACSView:
             .exists()
         )
 
-    def test_dispatch_applies_role_mapping_when_multiple_manage_account_users(
-        self, create_test_user, tenants_fixture, saml_setup, settings, monkeypatch
+    def test_dispatch_skips_role_mapping_when_last_manage_account_user_maps_to_existing_role(
+        self,
+        create_test_user,
+        tenants_fixture,
+        admin_role_fixture,
+        roles_fixture,
+        saml_setup,
+        settings,
+        monkeypatch,
     ):
-        """Test that role mapping is applied when tenant has multiple users with MANAGE_ACCOUNT role"""
+        """Test that role mapping is skipped when it would remove the last MANAGE_ACCOUNT user"""
         monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
         user = create_test_user
         tenant = tenants_fixture[0]
 
-        # Create a second user with manage_account=True
-        second_admin = User.objects.using(MainRouter.admin_db).create(
-            email="admin2@prowler.com", name="Second Admin"
-        )
-        admin_role = Role.objects.using(MainRouter.admin_db).create(
-            name="admin",
-            tenant=tenant,
-            manage_account=True,
-            manage_users=True,
-            manage_billing=True,
-            manage_providers=True,
-            manage_integrations=True,
-            manage_scans=True,
-            unlimited_visibility=True,
-        )
+        admin_role = admin_role_fixture
+        viewer_role = roles_fixture[3]
         UserRoleRelationship.objects.using(MainRouter.admin_db).create(
             user=user, role=admin_role, tenant_id=tenant.id
         )
-        UserRoleRelationship.objects.using(MainRouter.admin_db).create(
-            user=second_admin, role=admin_role, tenant_id=tenant.id
-        )
 
         social_account = SocialAccount(
             user=user,
@@ -10967,7 +11091,7 @@ class TestTenantFinishACSView:
                 "firstName": ["John"],
                 "lastName": ["Doe"],
                 "organization": ["testing_company"],
-                "userType": ["viewer"],  # This SHOULD be applied
+                "userType": [viewer_role.name],
             },
         )
 
@@ -11005,10 +11129,91 @@ class TestTenantFinishACSView:
 
         assert response.status_code == 302
 
-        # Verify the viewer role was created and assigned (role mapping was applied)
-        viewer_role = Role.objects.using(MainRouter.admin_db).get(
-            name="viewer", tenant=tenant
+        assert (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(user=user, role=admin_role, tenant_id=tenant.id)
+            .exists()
         )
+        assert not (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(user=user, role=viewer_role, tenant_id=tenant.id)
+            .exists()
+        )
+
+    def test_dispatch_applies_role_mapping_when_multiple_manage_account_users(
+        self,
+        create_test_user,
+        tenants_fixture,
+        admin_role_fixture,
+        roles_fixture,
+        saml_setup,
+        settings,
+        monkeypatch,
+    ):
+        """Test that role mapping is applied when tenant has multiple users with MANAGE_ACCOUNT role"""
+        monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
+        user = create_test_user
+        tenant = tenants_fixture[0]
+
+        # Create a second user with manage_account=True
+        second_admin = User.objects.using(MainRouter.admin_db).create(
+            email="admin2@prowler.com", name="Second Admin"
+        )
+        admin_role = admin_role_fixture
+        viewer_role = roles_fixture[3]
+        UserRoleRelationship.objects.using(MainRouter.admin_db).create(
+            user=user, role=admin_role, tenant_id=tenant.id
+        )
+        UserRoleRelationship.objects.using(MainRouter.admin_db).create(
+            user=second_admin, role=admin_role, tenant_id=tenant.id
+        )
+
+        social_account = SocialAccount(
+            user=user,
+            provider="saml",
+            extra_data={
+                "firstName": ["John"],
+                "lastName": ["Doe"],
+                "organization": ["testing_company"],
+                "userType": [viewer_role.name],  # This SHOULD be applied
+            },
+        )
+
+        request = RequestFactory().get(
+            reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
+        )
+        request.user = user
+        request.session = {}
+
+        with (
+            patch(
+                "allauth.socialaccount.providers.saml.views.get_app_or_404"
+            ) as mock_get_app_or_404,
+            patch(
+                "allauth.socialaccount.models.SocialApp.objects.get"
+            ) as mock_socialapp_get,
+            patch(
+                "allauth.socialaccount.models.SocialAccount.objects.get"
+            ) as mock_sa_get,
+            patch("api.models.SAMLDomainIndex.objects.get") as mock_saml_domain_get,
+            patch("api.models.SAMLConfiguration.objects.get") as mock_saml_config_get,
+            patch("api.models.User.objects.get") as mock_user_get,
+        ):
+            mock_get_app_or_404.return_value = MagicMock(
+                provider="saml", client_id="testtenant", name="Test App", settings={}
+            )
+            mock_sa_get.return_value = social_account
+            mock_socialapp_get.return_value = MagicMock(provider_id="saml")
+            mock_saml_domain_get.return_value = SimpleNamespace(tenant_id=tenant.id)
+            mock_saml_config_get.return_value = MagicMock()
+            mock_user_get.return_value = user
+
+            view = TenantFinishACSView.as_view()
+            response = view(request, organization_slug="testtenant")
+
+        assert response.status_code == 302
+
+        # Verify the viewer role was assigned (role mapping was applied)
         assert (
             UserRoleRelationship.objects.using(MainRouter.admin_db)
             .filter(user=user, role=viewer_role, tenant_id=tenant.id)
@@ -11022,6 +11227,86 @@ class TestTenantFinishACSView:
             .exists()
         )
 
+    def test_dispatch_applies_role_mapping_for_non_admin_user_with_single_admin(
+        self,
+        create_test_user,
+        tenants_fixture,
+        admin_role_fixture,
+        roles_fixture,
+        saml_setup,
+        settings,
+        monkeypatch,
+    ):
+        """Test that role mapping is applied for a non-admin user when a single admin exists"""
+        monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
+        admin_user = create_test_user
+        tenant = tenants_fixture[0]
+        non_admin_user = User.objects.using(MainRouter.admin_db).create(
+            email="viewer@prowler.com", name="Viewer"
+        )
+
+        admin_role = admin_role_fixture
+        viewer_role = roles_fixture[3]
+        UserRoleRelationship.objects.using(MainRouter.admin_db).create(
+            user=admin_user, role=admin_role, tenant_id=tenant.id
+        )
+
+        social_account = SocialAccount(
+            user=non_admin_user,
+            provider="saml",
+            extra_data={
+                "firstName": ["Jane"],
+                "lastName": ["Doe"],
+                "organization": ["testing_company"],
+                "userType": [viewer_role.name],
+            },
+        )
+
+        request = RequestFactory().get(
+            reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
+        )
+        request.user = non_admin_user
+        request.session = {}
+
+        with (
+            patch(
+                "allauth.socialaccount.providers.saml.views.get_app_or_404"
+            ) as mock_get_app_or_404,
+            patch(
+                "allauth.socialaccount.models.SocialApp.objects.get"
+            ) as mock_socialapp_get,
+            patch(
+                "allauth.socialaccount.models.SocialAccount.objects.get"
+            ) as mock_sa_get,
+            patch("api.models.SAMLDomainIndex.objects.get") as mock_saml_domain_get,
+            patch("api.models.SAMLConfiguration.objects.get") as mock_saml_config_get,
+            patch("api.models.User.objects.get") as mock_user_get,
+        ):
+            mock_get_app_or_404.return_value = MagicMock(
+                provider="saml", client_id="testtenant", name="Test App", settings={}
+            )
+            mock_sa_get.return_value = social_account
+            mock_socialapp_get.return_value = MagicMock(provider_id="saml")
+            mock_saml_domain_get.return_value = SimpleNamespace(tenant_id=tenant.id)
+            mock_saml_config_get.return_value = MagicMock()
+            mock_user_get.return_value = non_admin_user
+
+            view = TenantFinishACSView.as_view()
+            response = view(request, organization_slug="testtenant")
+
+        assert response.status_code == 302
+
+        assert (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(user=non_admin_user, role=viewer_role, tenant_id=tenant.id)
+            .exists()
+        )
+        assert (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(user=admin_user, role=admin_role, tenant_id=tenant.id)
+            .exists()
+        )
+
 
 @pytest.mark.django_db
 class TestLighthouseConfigViewSet:

api/src/backend/api/utils.py

@@ -33,6 +33,7 @@ if TYPE_CHECKING:
     from prowler.providers.mongodbatlas.mongodbatlas_provider import (
         MongodbatlasProvider,
     )
+    from prowler.providers.openstack.openstack_provider import OpenstackProvider
     from prowler.providers.oraclecloud.oraclecloud_provider import OraclecloudProvider
 
 
@@ -78,12 +79,14 @@ def return_prowler_provider(
     AlibabacloudProvider
     | AwsProvider
     | AzureProvider
+    | CloudflareProvider
     | GcpProvider
     | GithubProvider
     | IacProvider
     | KubernetesProvider
     | M365Provider
     | MongodbatlasProvider
+    | OpenstackProvider
     | OraclecloudProvider
 ):
     """Return the Prowler provider class based on the given provider type.
@@ -92,7 +95,7 @@ def return_prowler_provider(
         provider (Provider): The provider object containing the provider type and associated secrets.
 
     Returns:
-        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OraclecloudProvider: The corresponding provider class.
+        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OpenstackProvider | OraclecloudProvider: The corresponding provider class.
 
     Raises:
         ValueError: If the provider type specified in `provider.provider` is not supported.
@@ -152,6 +155,10 @@ def return_prowler_provider(
             )
 
             prowler_provider = CloudflareProvider
+        case Provider.ProviderChoices.OPENSTACK.value:
+            from prowler.providers.openstack.openstack_provider import OpenstackProvider
+
+            prowler_provider = OpenstackProvider
         case _:
             raise ValueError(f"Provider type {provider.provider} not supported")
     return prowler_provider
@@ -208,6 +215,12 @@ def get_prowler_provider_kwargs(
             **prowler_provider_kwargs,
             "filter_accounts": [provider.uid],
         }
+    elif provider.provider == Provider.ProviderChoices.OPENSTACK.value:
+        # No extra kwargs needed: clouds_yaml_content and clouds_yaml_cloud from the
+        # secret are sufficient. Validating project_id (provider.uid) against the
+        # clouds.yaml is not feasible because not all auth methods include it and the
+        # Keystone API is unavailable on public clouds.
+        pass
 
     if mutelist_processor:
         mutelist_content = mutelist_processor.configuration.get("Mutelist", {})
@@ -232,6 +245,7 @@ def initialize_prowler_provider(
     | KubernetesProvider
     | M365Provider
     | MongodbatlasProvider
+    | OpenstackProvider
     | OraclecloudProvider
 ):
     """Initialize a Prowler provider instance based on the given provider type.
@@ -241,7 +255,7 @@ def initialize_prowler_provider(
         mutelist_processor (Processor): The mutelist processor object containing the mutelist configuration.
 
     Returns:
-        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OraclecloudProvider: An instance of the corresponding provider class
+        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OpenstackProvider | OraclecloudProvider: An instance of the corresponding provider class
             initialized with the provider's secrets.
     """
     prowler_provider = return_prowler_provider(provider)
@@ -276,6 +290,13 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
         if "access_token" in prowler_provider_kwargs:
             iac_test_kwargs["access_token"] = prowler_provider_kwargs["access_token"]
         return prowler_provider.test_connection(**iac_test_kwargs)
+    elif provider.provider == Provider.ProviderChoices.OPENSTACK.value:
+        openstack_kwargs = {
+            "clouds_yaml_content": prowler_provider_kwargs["clouds_yaml_content"],
+            "clouds_yaml_cloud": prowler_provider_kwargs["clouds_yaml_cloud"],
+            "raise_on_exception": False,
+        }
+        return prowler_provider.test_connection(**openstack_kwargs)
     else:
         return prowler_provider.test_connection(
             **prowler_provider_kwargs,

api/src/backend/api/v1/serializer_utils/providers.py

@@ -373,6 +373,21 @@ from rest_framework_json_api import serializers
                 },
                 "required": ["api_key", "api_email"],
             },
+            {
+                "type": "object",
+                "title": "OpenStack clouds.yaml Credentials",
+                "properties": {
+                    "clouds_yaml_content": {
+                        "type": "string",
+                        "description": "The full content of a clouds.yaml configuration file.",
+                    },
+                    "clouds_yaml_cloud": {
+                        "type": "string",
+                        "description": "The name of the cloud to use from the clouds.yaml file.",
+                    },
+                },
+                "required": ["clouds_yaml_content", "clouds_yaml_cloud"],
+            },
         ]
     }
 )

api/src/backend/api/v1/serializers.py

@@ -1145,6 +1145,7 @@ class AttackPathsScanSerializer(RLSSerializer):
             "id",
             "state",
             "progress",
+            "graph_data_ready",
             "provider",
             "provider_alias",
             "provider_type",
@@ -1176,6 +1177,14 @@ class AttackPathsScanSerializer(RLSSerializer):
         return provider.uid if provider else None
 
 
+class AttackPathsQueryAttributionSerializer(BaseSerializerV1):
+    text = serializers.CharField()
+    link = serializers.CharField()
+
+    class JSONAPIMeta:
+        resource_name = "attack-paths-query-attributions"
+
+
 class AttackPathsQueryParameterSerializer(BaseSerializerV1):
     name = serializers.CharField()
     label = serializers.CharField()
@@ -1190,7 +1199,9 @@ class AttackPathsQueryParameterSerializer(BaseSerializerV1):
 class AttackPathsQuerySerializer(BaseSerializerV1):
     id = serializers.CharField()
     name = serializers.CharField()
+    short_description = serializers.CharField()
     description = serializers.CharField()
+    attribution = AttackPathsQueryAttributionSerializer(allow_null=True, required=False)
     provider = serializers.CharField()
     parameters = AttackPathsQueryParameterSerializer(many=True)
 
@@ -1515,6 +1526,8 @@ class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
                             "or both 'api_key' and 'api_email'."
                         }
                     )
+            elif provider_type == Provider.ProviderChoices.OPENSTACK.value:
+                serializer = OpenStackCloudsYamlProviderSecret(data=secret)
             else:
                 raise serializers.ValidationError(
                     {"provider": f"Provider type not supported {provider_type}"}
@@ -1681,6 +1694,14 @@ class CloudflareApiKeyProviderSecret(serializers.Serializer):
         resource_name = "provider-secrets"
 
 
+class OpenStackCloudsYamlProviderSecret(serializers.Serializer):
+    clouds_yaml_content = serializers.CharField()
+    clouds_yaml_cloud = serializers.CharField()
+
+    class Meta:
+        resource_name = "provider-secrets"
+
+
 class AlibabaCloudProviderSecret(serializers.Serializer):
     access_key_id = serializers.CharField()
     access_key_secret = serializers.CharField()

api/src/backend/api/v1/views.py

@@ -763,27 +763,40 @@ class TenantFinishACSView(FinishACSView):
             .tenant
         )
 
-        # Check if tenant has only one user with MANAGE_ACCOUNT role
-        users_with_manage_account = (
+        role_name = (
+            extra.get("userType", ["no_permissions"])[0].strip()
+            if extra.get("userType")
+            else "no_permissions"
+        )
+        role = (
+            Role.objects.using(MainRouter.admin_db)
+            .filter(name=role_name, tenant=tenant)
+            .first()
+        )
+
+        # Only skip mapping if it would remove the last MANAGE_ACCOUNT user
+        remaining_manage_account_users = (
             UserRoleRelationship.objects.using(MainRouter.admin_db)
             .filter(role__manage_account=True, tenant_id=tenant.id)
+            .exclude(user_id=user_id)
             .values("user")
             .distinct()
             .count()
         )
+        user_has_manage_account = (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(role__manage_account=True, tenant_id=tenant.id, user_id=user_id)
+            .exists()
+        )
+        role_manage_account = role.manage_account if role else False
+        would_remove_last_manage_account = (
+            user_has_manage_account
+            and remaining_manage_account_users == 0
+            and not role_manage_account
+        )
 
-        # Only apply role mapping from userType if tenant does NOT have exactly one user with MANAGE_ACCOUNT
-        if users_with_manage_account != 1:
-            role_name = (
-                extra.get("userType", ["no_permissions"])[0].strip()
-                if extra.get("userType")
-                else "no_permissions"
-            )
-            try:
-                role = Role.objects.using(MainRouter.admin_db).get(
-                    name=role_name, tenant=tenant
-                )
-            except Role.DoesNotExist:
+        if not would_remove_last_manage_account:
+            if role is None:
                 role = Role.objects.using(MainRouter.admin_db).create(
                     name=role_name,
                     tenant=tenant,
@@ -1746,6 +1759,25 @@ class ProviderViewSet(DisablePaginationMixin, BaseRLSViewSet):
             ),
         },
     ),
+    csa=extend_schema(
+        tags=["Scan"],
+        summary="Retrieve CSA CCM compliance report",
+        description="Download CSA Cloud Controls Matrix (CCM) v4.0 compliance report as a PDF file.",
+        request=None,
+        responses={
+            200: OpenApiResponse(
+                description="PDF file containing the CSA CCM compliance report"
+            ),
+            202: OpenApiResponse(description="The task is in progress"),
+            401: OpenApiResponse(
+                description="API key missing or user not Authenticated"
+            ),
+            403: OpenApiResponse(description="There is a problem with credentials"),
+            404: OpenApiResponse(
+                description="The scan has no CSA CCM reports, or the CSA CCM report generation task has not started yet"
+            ),
+        },
+    ),
 )
 @method_decorator(CACHE_DECORATOR, name="list")
 @method_decorator(CACHE_DECORATOR, name="retrieve")
@@ -1811,6 +1843,9 @@ class ScanViewSet(BaseRLSViewSet):
         elif self.action == "nis2":
             if hasattr(self, "response_serializer_class"):
                 return self.response_serializer_class
+        elif self.action == "csa":
+            if hasattr(self, "response_serializer_class"):
+                return self.response_serializer_class
         return super().get_serializer_class()
 
     def partial_update(self, request, *args, **kwargs):
@@ -2172,6 +2207,45 @@ class ScanViewSet(BaseRLSViewSet):
         content, filename = loader
         return self._serve_file(content, filename, "application/pdf")
 
+    @action(
+        detail=True,
+        methods=["get"],
+        url_name="csa",
+    )
+    def csa(self, request, pk=None):
+        scan = self.get_object()
+        running_resp = self._get_task_status(scan)
+        if running_resp:
+            return running_resp
+
+        if not scan.output_location:
+            return Response(
+                {
+                    "detail": "The scan has no reports, or the CSA CCM report generation task has not started yet."
+                },
+                status=status.HTTP_404_NOT_FOUND,
+            )
+
+        if scan.output_location.startswith("s3://"):
+            bucket = env.str("DJANGO_OUTPUT_S3_AWS_OUTPUT_BUCKET", "")
+            key_prefix = scan.output_location.removeprefix(f"s3://{bucket}/")
+            prefix = os.path.join(
+                os.path.dirname(key_prefix),
+                "csa",
+                "*_csa_report.pdf",
+            )
+            loader = self._load_file(prefix, s3=True, bucket=bucket, list_objects=True)
+        else:
+            base = os.path.dirname(scan.output_location)
+            pattern = os.path.join(base, "csa", "*_csa_report.pdf")
+            loader = self._load_file(pattern, s3=False)
+
+        if isinstance(loader, Response):
+            return loader
+
+        content, filename = loader
+        return self._serve_file(content, filename, "application/pdf")
+
     def create(self, request, *args, **kwargs):
         input_serializer = self.get_serializer(data=request.data)
         input_serializer.is_valid(raise_exception=True)
@@ -2408,22 +2482,13 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
     def run_attack_paths_query(self, request, pk=None):
         attack_paths_scan = self.get_object()
 
-        if attack_paths_scan.state != StateChoices.COMPLETED:
+        if not attack_paths_scan.graph_data_ready:
             raise ValidationError(
                 {
-                    "detail": "The Attack Paths scan must be completed before running Attack Paths queries"
+                    "detail": "Attack Paths data is not available for querying - a scan must complete at least once before queries can be run"
                 }
             )
 
-        if not attack_paths_scan.graph_database:
-            logger.error(
-                f"The Attack Paths Scan {attack_paths_scan.id} does not reference a graph database"
-            )
-            return Response(
-                {"detail": "The Attack Paths scan does not reference a graph database"},
-                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            )
-
         payload = attack_paths_views_helpers.normalize_run_payload(request.data)
         serializer = AttackPathsQueryRunRequestSerializer(data=payload)
         serializer.is_valid(raise_exception=True)
@@ -2437,6 +2502,9 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
                 {"id": "Unknown Attack Paths query for the selected provider"}
             )
 
+        database_name = graph_database.get_database_name(
+            attack_paths_scan.provider.tenant_id
+        )
         parameters = attack_paths_views_helpers.prepare_query_parameters(
             query_definition,
             serializer.validated_data.get("parameters", {}),
@@ -2444,9 +2512,9 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
         )
 
         graph = attack_paths_views_helpers.execute_attack_paths_query(
-            attack_paths_scan, query_definition, parameters
+            database_name, query_definition, parameters
         )
-        graph_database.clear_cache(attack_paths_scan.graph_database)
+        graph_database.clear_cache(database_name)
 
         status_code = status.HTTP_200_OK
         if not graph.get("nodes"):

api/src/backend/config/django/testing.py

@@ -18,6 +18,10 @@ DATABASES = {
 
 DATABASE_ROUTERS = []
 TESTING = True
+# Override page size for testing to a value only slightly above the current fixture count.
+# We explicitly set PAGE_SIZE to 15 (round number just above fixture) to avoid masking pagination bugs, while not setting it excessively high.
+# If you add more providers to the fixture, please review that the total value is below the current one and update this value if needed.
+REST_FRAMEWORK["PAGE_SIZE"] = 15  # noqa: F405
 SECRETS_ENCRYPTION_KEY = "ZMiYVo7m4Fbe2eXXPyrwxdJss2WSalXSv3xHBcJkPl0="
 
 # DRF Simple API Key settings

api/src/backend/conftest.py

@@ -537,6 +537,12 @@ def providers_fixture(tenants_fixture):
         alias="cloudflare_testing",
         tenant_id=tenant.id,
     )
+    provider11 = Provider.objects.create(
+        provider="openstack",
+        uid="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
+        alias="openstack_testing",
+        tenant_id=tenant.id,
+    )
 
     return (
         provider1,
@@ -549,6 +555,7 @@ def providers_fixture(tenants_fixture):
         provider8,
         provider9,
         provider10,
+        provider11,
     )
 
 
@@ -1618,7 +1625,6 @@ def create_attack_paths_scan():
         scan=None,
         state=StateChoices.COMPLETED,
         progress=0,
-        graph_database="tenant-db",
         **extra_fields,
     ):
         scan_instance = scan or Scan.objects.create(
@@ -1635,7 +1641,6 @@ def create_attack_paths_scan():
             "scan": scan_instance,
             "state": state,
             "progress": progress,
-            "graph_database": graph_database,
         }
         payload.update(extra_fields)
 
@@ -1663,6 +1668,7 @@ def attack_paths_query_definition_factory():
         definition_payload = {
             "id": "aws-test",
             "name": "Attack Paths Test Query",
+            "short_description": "Synthetic short description for tests.",
             "description": "Synthetic Attack Paths definition for tests.",
             "provider": "aws",
             "cypher": "RETURN 1",

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

@@ -12,8 +12,10 @@ BATCH_SIZE = env.int("ATTACK_PATHS_BATCH_SIZE", 1000)
 # Neo4j internal labels (Prowler-specific, not provider-specific)
 # - `ProwlerFinding`: Label for finding nodes created by Prowler and linked to cloud resources.
 # - `ProviderResource`: Added to ALL synced nodes for provider isolation and drop/query ops.
+# - `Internet`: Singleton node representing external internet access for exposed-resource queries.
 PROWLER_FINDING_LABEL = "ProwlerFinding"
 PROVIDER_RESOURCE_LABEL = "ProviderResource"
+INTERNET_NODE_LABEL = "Internet"
 
 
 @dataclass(frozen=True)

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

@@ -2,7 +2,9 @@ from datetime import datetime, timezone
 from typing import Any
 
 from cartography.config import Config as CartographyConfig
+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 (
     AttackPathsScan as ProwlerAPIAttackPathsScan,
@@ -11,6 +13,8 @@ from api.models import (
 )
 from tasks.jobs.attack_paths.config import is_provider_available
 
+logger = get_task_logger(__name__)
+
 
 def can_provider_run_attack_paths_scan(tenant_id: str, provider_id: int) -> bool:
     with rls_transaction(tenant_id):
@@ -28,12 +32,21 @@ def create_attack_paths_scan(
         return None
 
     with rls_transaction(tenant_id):
+        # Inherit graph_data_ready from the previous scan for this provider,
+        # so queries remain available while the new scan runs.
+        previous_data_ready = ProwlerAPIAttackPathsScan.objects.filter(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+            graph_data_ready=True,
+        ).exists()
+
         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),
+            graph_data_ready=previous_data_ready,
         )
         attack_paths_scan.save()
 
@@ -66,7 +79,6 @@ def starting_attack_paths_scan(
         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=[
@@ -74,7 +86,6 @@ def starting_attack_paths_scan(
                 "state",
                 "started_at",
                 "update_tag",
-                "graph_database",
             ]
         )
 
@@ -86,7 +97,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
@@ -114,33 +129,59 @@ def update_attack_paths_scan_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,
+def set_graph_data_ready(
+    attack_paths_scan: ProwlerAPIAttackPathsScan,
+    ready: bool,
 ) -> 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"])
+    with rls_transaction(attack_paths_scan.tenant_id):
+        attack_paths_scan.graph_data_ready = ready
+        attack_paths_scan.save(update_fields=["graph_data_ready"])
+
+
+def set_provider_graph_data_ready(
+    attack_paths_scan: ProwlerAPIAttackPathsScan,
+    ready: bool,
+) -> None:
+    """
+    Set `graph_data_ready` for ALL scans of the same provider.
+
+    Used before drop/sync so that older scan IDs cannot bypass the query gate while the graph is being replaced.
+    """
+    with rls_transaction(attack_paths_scan.tenant_id):
+        ProwlerAPIAttackPathsScan.objects.filter(
+            tenant_id=attack_paths_scan.tenant_id,
+            provider_id=attack_paths_scan.provider_id,
+        ).update(graph_data_ready=ready)
+        attack_paths_scan.refresh_from_db(fields=["graph_data_ready"])
+
+
+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,
+    ):
+        tmp_db_name = graph_database.get_database_name(
+            attack_paths_scan.id, temporary=True
+        )
+        try:
+            graph_database.drop_database(tmp_db_name)
+
+        except Exception:
+            logger.exception(
+                f"Failed to drop temp database {tmp_db_name} during failure handling"
+            )
+
+        finish_attack_paths_scan(
+            attack_paths_scan,
+            StateChoices.FAILED,
+            {"global_error": error},
+        )

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

@@ -6,6 +6,7 @@ from cartography.client.core.tx import run_write_query
 from celery.utils.log import get_task_logger
 
 from tasks.jobs.attack_paths.config import (
+    INTERNET_NODE_LABEL,
     PROWLER_FINDING_LABEL,
     PROVIDER_RESOURCE_LABEL,
 )
@@ -30,6 +31,8 @@ FINDINGS_INDEX_STATEMENTS = [
     f"CREATE INDEX prowler_finding_provider_uid IF NOT EXISTS FOR (n:{PROWLER_FINDING_LABEL}) ON (n.provider_uid);",
     f"CREATE INDEX prowler_finding_lastupdated IF NOT EXISTS FOR (n:{PROWLER_FINDING_LABEL}) ON (n.lastupdated);",
     f"CREATE INDEX prowler_finding_status IF NOT EXISTS FOR (n:{PROWLER_FINDING_LABEL}) ON (n.status);",
+    # Internet node index for MERGE lookups
+    f"CREATE INDEX internet_id IF NOT EXISTS FOR (n:{INTERNET_NODE_LABEL}) ON (n.id);",
 ]
 
 # Indexes for provider resource sync operations

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

@@ -0,0 +1,67 @@
+"""
+Internet node enrichment for Attack Paths graph.
+
+Creates a real Internet node and CAN_ACCESS relationships to
+internet-exposed resources (EC2Instance, LoadBalancer, LoadBalancerV2)
+in the temporary scan database before sync.
+"""
+
+import neo4j
+
+from cartography.config import Config as CartographyConfig
+from celery.utils.log import get_task_logger
+
+from api.models import Provider
+from prowler.config import config as ProwlerConfig
+from tasks.jobs.attack_paths.config import get_root_node_label
+from tasks.jobs.attack_paths.queries import (
+    CREATE_CAN_ACCESS_RELATIONSHIPS_TEMPLATE,
+    CREATE_INTERNET_NODE,
+    render_cypher_template,
+)
+
+logger = get_task_logger(__name__)
+
+
+def analysis(
+    neo4j_session: neo4j.Session,
+    prowler_api_provider: Provider,
+    config: CartographyConfig,
+) -> int:
+    """
+    Create Internet node and CAN_ACCESS relationships to exposed resources.
+
+    Args:
+        neo4j_session: Active Neo4j session (temp database).
+        prowler_api_provider: The Prowler API provider instance.
+        config: Cartography configuration with update_tag.
+
+    Returns:
+        Number of CAN_ACCESS relationships created.
+    """
+    provider_uid = str(prowler_api_provider.uid)
+
+    parameters = {
+        "provider_uid": provider_uid,
+        "last_updated": config.update_tag,
+        "prowler_version": ProwlerConfig.prowler_version,
+    }
+
+    logger.info(f"Creating Internet node for provider {provider_uid}")
+    neo4j_session.run(CREATE_INTERNET_NODE, parameters)
+
+    query = render_cypher_template(
+        CREATE_CAN_ACCESS_RELATIONSHIPS_TEMPLATE,
+        {"__ROOT_LABEL__": get_root_node_label(prowler_api_provider.provider)},
+    )
+
+    logger.info(
+        f"Creating CAN_ACCESS relationships from Internet to exposed resources for {provider_uid}"
+    )
+    result = neo4j_session.run(query, parameters)
+    relationships_merged = result.single().get("relationships_merged", 0)
+
+    logger.info(
+        f"Created {relationships_merged} CAN_ACCESS relationships for provider {provider_uid}"
+    )
+    return relationships_merged

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

@@ -1,5 +1,6 @@
 # Cypher query templates for Attack Paths operations
 from tasks.jobs.attack_paths.config import (
+    INTERNET_NODE_LABEL,
     PROWLER_FINDING_LABEL,
     PROVIDER_RESOURCE_LABEL,
 )
@@ -91,6 +92,37 @@ CLEANUP_FINDINGS_TEMPLATE = f"""
     RETURN COUNT(finding) AS deleted_findings_count
 """
 
+# Internet queries (used by internet.py)
+# ---------------------------------------
+
+CREATE_INTERNET_NODE = f"""
+    MERGE (internet:{INTERNET_NODE_LABEL} {{id: 'Internet'}})
+    ON CREATE SET
+        internet.name = 'Internet',
+        internet.firstseen = timestamp(),
+        internet.lastupdated = $last_updated,
+        internet._module_name = 'cartography:prowler',
+        internet._module_version = $prowler_version
+    ON MATCH SET
+        internet.lastupdated = $last_updated
+"""
+
+CREATE_CAN_ACCESS_RELATIONSHIPS_TEMPLATE = f"""
+    MATCH (account:__ROOT_LABEL__ {{id: $provider_uid}})-->(resource)
+    WHERE resource.exposed_internet = true
+    WITH resource
+    MATCH (internet:{INTERNET_NODE_LABEL} {{id: 'Internet'}})
+    MERGE (internet)-[r:CAN_ACCESS]->(resource)
+    ON CREATE SET
+        r.firstseen = timestamp(),
+        r.lastupdated = $last_updated,
+        r._module_name = 'cartography:prowler',
+        r._module_version = $prowler_version
+    ON MATCH SET
+        r.lastupdated = $last_updated
+    RETURN COUNT(r) AS relationships_merged
+"""
+
 # Sync queries (used by sync.py)
 # -------------------------------
 

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

@@ -16,7 +16,7 @@ from api.models import (
     StateChoices,
 )
 from api.utils import initialize_prowler_provider
-from tasks.jobs.attack_paths import db_utils, findings, sync, utils
+from tasks.jobs.attack_paths import db_utils, findings, internet, sync, utils
 from tasks.jobs.attack_paths.config import get_cartography_ingestion_function
 
 # Without this Celery goes crazy with Cartography logging
@@ -135,7 +135,15 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             cartography_analysis.run(tmp_neo4j_session, tmp_cartography_config)
             db_utils.update_attack_paths_scan_progress(attack_paths_scan, 96)
 
-            # Adding Prowler nodes and relationships
+            # Creating Internet node and CAN_ACCESS relationships
+            logger.info(
+                f"Creating Internet graph for AWS account {prowler_api_provider.uid}"
+            )
+            internet.analysis(
+                tmp_neo4j_session, prowler_api_provider, tmp_cartography_config
+            )
+
+            # Adding Prowler Finding nodes and relationships
             logger.info(
                 f"Syncing Prowler analysis for AWS account {prowler_api_provider.uid}"
             )
@@ -161,6 +169,7 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             sync.create_sync_indexes(tenant_neo4j_session)
 
         logger.info(f"Deleting existing provider graph in {tenant_database_name}")
+        db_utils.set_provider_graph_data_ready(attack_paths_scan, False)
         graph_database.drop_subgraph(
             database=tenant_database_name,
             provider_id=str(prowler_api_provider.id),
@@ -175,6 +184,7 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             target_database=tenant_database_name,
             provider_id=str(prowler_api_provider.id),
         )
+        db_utils.set_graph_data_ready(attack_paths_scan, True)
         db_utils.update_attack_paths_scan_progress(attack_paths_scan, 99)
 
         logger.info(f"Clearing Neo4j cache for database {tenant_database_name}")
@@ -185,30 +195,6 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             f"{prowler_api_provider.provider.upper()} provider {prowler_api_provider.id}"
         )
 
-        # TODO
-        # This piece of code delete old Neo4j databases for this tenant's provider
-        # When we clean all of these databases we need to:
-        #   - Delete this block
-        #   - Delete function from `db_utils` the functions get_old_attack_paths_scans` & `update_old_attack_paths_scan`
-        #   - Remove `graph_database` & `is_graph_database_deleted` from the AttackPathsScan model:
-        #     - Check indexes
-        #     - Create migration
-        #     - The use of `attack_paths_scan.graph_database` on `views` and `views_helpers`
-        #     - Tests
-        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:
-            old_graph_database = old_attack_paths_scan.graph_database
-            if old_graph_database and old_graph_database != tenant_database_name:
-                logger.info(
-                    f"Dropping old Neo4j database {old_graph_database} for provider {prowler_api_provider.id}"
-                )
-                graph_database.drop_database(old_graph_database)
-            db_utils.update_old_attack_paths_scan(old_attack_paths_scan)
-
         logger.info(f"Dropping temporary Neo4j database {tmp_database_name}")
         graph_database.drop_database(tmp_database_name)
 
@@ -220,10 +206,17 @@ 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/deletion.py

@@ -27,23 +27,24 @@ def delete_provider(tenant_id: str, pk: str):
 
     Returns:
         dict: A dictionary with the count of deleted objects per model,
-              including related models.
-
-    Raises:
-        Provider.DoesNotExist: If no instance with the provided primary key exists.
+              including related models. Returns an empty dict if the provider
+              was already deleted.
     """
-    # Delete the Attack Paths' graph data related to the provider
-    tenant_database_name = graph_database.get_database_name(tenant_id)
-    try:
-        graph_database.drop_subgraph(tenant_database_name, str(pk))
 
-    except graph_database.GraphDatabaseQueryException as gdb_error:
-        logger.error(f"Error deleting Provider graph data: {gdb_error}")
-        raise
-
-    # Get all provider related data and delete them in batches
+    # Get all provider related data to delete them in batches
     with rls_transaction(tenant_id):
-        instance = Provider.all_objects.get(pk=pk)
+        try:
+            instance = Provider.all_objects.get(pk=pk)
+        except Provider.DoesNotExist:
+            logger.info(f"Provider `{pk}` already deleted, skipping")
+            return {}
+
+        attack_paths_scan_ids = list(
+            AttackPathsScan.all_objects.filter(provider=instance).values_list(
+                "id", flat=True
+            )
+        )
+
         deletion_steps = [
             ("Scan Summaries", ScanSummary.all_objects.filter(scan__provider=instance)),
             ("Findings", Finding.all_objects.filter(scan__provider=instance)),
@@ -52,6 +53,25 @@ def delete_provider(tenant_id: str, pk: str):
             ("AttackPathsScans", AttackPathsScan.all_objects.filter(provider=instance)),
         ]
 
+    # Drop orphaned temporary Neo4j databases
+    for aps_id in attack_paths_scan_ids:
+        tmp_db_name = graph_database.get_database_name(aps_id, temporary=True)
+        try:
+            graph_database.drop_database(tmp_db_name)
+
+        except graph_database.GraphDatabaseQueryException:
+            logger.warning(f"Failed to drop temp database {tmp_db_name}, continuing")
+
+    # Delete the Attack Paths' graph data related to the provider from the tenant database
+    tenant_database_name = graph_database.get_database_name(tenant_id)
+    try:
+        graph_database.drop_subgraph(tenant_database_name, str(pk))
+
+    except graph_database.GraphDatabaseQueryException as gdb_error:
+        logger.error(f"Error deleting Provider graph data: {gdb_error}")
+        raise
+
+    # Delete related data in batches
     deletion_summary = {}
     for step_name, queryset in deletion_steps:
         try:
@@ -61,6 +81,7 @@ def delete_provider(tenant_id: str, pk: str):
             logger.error(f"Error deleting {step_name}: {db_error}")
             raise
 
+    # Delete the provider instance itself
     try:
         with rls_transaction(tenant_id):
             _, provider_summary = instance.delete()
@@ -85,7 +106,9 @@ def delete_tenant(pk: str):
     """
     deletion_summary = {}
 
-    for provider in Provider.objects.using(MainRouter.admin_db).filter(tenant_id=pk):
+    for provider in Provider.all_objects.using(MainRouter.admin_db).filter(
+        tenant_id=pk
+    ):
         summary = delete_provider(pk, provider.id)
         deletion_summary.update(summary)
 

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

@@ -35,6 +35,11 @@ from prowler.lib.outputs.compliance.cis.cis_github import GithubCIS
 from prowler.lib.outputs.compliance.cis.cis_kubernetes import KubernetesCIS
 from prowler.lib.outputs.compliance.cis.cis_m365 import M365CIS
 from prowler.lib.outputs.compliance.cis.cis_oraclecloud import OracleCloudCIS
+from prowler.lib.outputs.compliance.csa.csa_alibabacloud import AlibabaCloudCSA
+from prowler.lib.outputs.compliance.csa.csa_aws import AWSCSA
+from prowler.lib.outputs.compliance.csa.csa_azure import AzureCSA
+from prowler.lib.outputs.compliance.csa.csa_gcp import GCPCSA
+from prowler.lib.outputs.compliance.csa.csa_oraclecloud import OracleCloudCSA
 from prowler.lib.outputs.compliance.ens.ens_aws import AWSENS
 from prowler.lib.outputs.compliance.ens.ens_azure import AzureENS
 from prowler.lib.outputs.compliance.ens.ens_gcp import GCPENS
@@ -90,6 +95,7 @@ COMPLIANCE_CLASS_MAP = {
         (lambda name: name == "prowler_threatscore_aws", ProwlerThreatScoreAWS),
         (lambda name: name == "ccc_aws", CCC_AWS),
         (lambda name: name.startswith("c5_"), AWSC5),
+        (lambda name: name.startswith("csa_"), AWSCSA),
     ],
     "azure": [
         (lambda name: name.startswith("cis_"), AzureCIS),
@@ -99,6 +105,7 @@ COMPLIANCE_CLASS_MAP = {
         (lambda name: name == "ccc_azure", CCC_Azure),
         (lambda name: name == "prowler_threatscore_azure", ProwlerThreatScoreAzure),
         (lambda name: name == "c5_azure", AzureC5),
+        (lambda name: name.startswith("csa_"), AzureCSA),
     ],
     "gcp": [
         (lambda name: name.startswith("cis_"), GCPCIS),
@@ -108,6 +115,7 @@ COMPLIANCE_CLASS_MAP = {
         (lambda name: name == "prowler_threatscore_gcp", ProwlerThreatScoreGCP),
         (lambda name: name == "ccc_gcp", CCC_GCP),
         (lambda name: name == "c5_gcp", GCPC5),
+        (lambda name: name.startswith("csa_"), GCPCSA),
     ],
     "kubernetes": [
         (lambda name: name.startswith("cis_"), KubernetesCIS),
@@ -131,9 +139,11 @@ COMPLIANCE_CLASS_MAP = {
     ],
     "oraclecloud": [
         (lambda name: name.startswith("cis_"), OracleCloudCIS),
+        (lambda name: name.startswith("csa_"), OracleCloudCSA),
     ],
     "alibabacloud": [
         (lambda name: name.startswith("cis_"), AlibabaCloudCIS),
+        (lambda name: name.startswith("csa_"), AlibabaCloudCSA),
         (
             lambda name: name == "prowler_threatscore_alibabacloud",
             ProwlerThreatScoreAlibaba,

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

@@ -6,6 +6,7 @@ from config.django.base import DJANGO_TMP_OUTPUT_DIRECTORY
 from tasks.jobs.export import _generate_compliance_output_directory, _upload_to_s3
 from tasks.jobs.reports import (
     FRAMEWORK_REGISTRY,
+    CSAReportGenerator,
     ENSReportGenerator,
     NIS2ReportGenerator,
     ThreatScoreReportGenerator,
@@ -147,6 +148,49 @@ def generate_nis2_report(
     )
 
 
+def generate_csa_report(
+    tenant_id: str,
+    scan_id: str,
+    compliance_id: str,
+    output_path: str,
+    provider_id: str,
+    only_failed: bool = True,
+    include_manual: bool = False,
+    provider_obj: Provider | None = None,
+    requirement_statistics: dict[str, dict[str, int]] | None = None,
+    findings_cache: dict[str, list[FindingOutput]] | None = None,
+) -> None:
+    """
+    Generate a PDF compliance report for CSA Cloud Controls Matrix (CCM) v4.0.
+
+    Args:
+        tenant_id: The tenant ID for Row-Level Security context.
+        scan_id: ID of the scan executed by Prowler.
+        compliance_id: ID of the compliance framework (e.g., "csa_ccm_4.0_aws").
+        output_path: Output PDF file path.
+        provider_id: Provider ID for the scan.
+        only_failed: If True, only include failed requirements in detailed section.
+        include_manual: If True, include manual requirements in detailed section.
+        provider_obj: Pre-fetched Provider object to avoid duplicate queries.
+        requirement_statistics: Pre-aggregated requirement statistics.
+        findings_cache: Cache of already loaded findings to avoid duplicate queries.
+    """
+    generator = CSAReportGenerator(FRAMEWORK_REGISTRY["csa_ccm"])
+
+    generator.generate(
+        tenant_id=tenant_id,
+        scan_id=scan_id,
+        compliance_id=compliance_id,
+        output_path=output_path,
+        provider_id=provider_id,
+        provider_obj=provider_obj,
+        requirement_statistics=requirement_statistics,
+        findings_cache=findings_cache,
+        only_failed=only_failed,
+        include_manual=include_manual,
+    )
+
+
 def generate_compliance_reports(
     tenant_id: str,
     scan_id: str,
@@ -154,11 +198,14 @@ def generate_compliance_reports(
     generate_threatscore: bool = True,
     generate_ens: bool = True,
     generate_nis2: bool = True,
+    generate_csa: bool = True,
     only_failed_threatscore: bool = True,
     min_risk_level_threatscore: int = 4,
     include_manual_ens: bool = True,
     include_manual_nis2: bool = False,
     only_failed_nis2: bool = True,
+    only_failed_csa: bool = True,
+    include_manual_csa: bool = False,
 ) -> dict[str, dict[str, bool | str]]:
     """
     Generate multiple compliance reports with shared database queries.
@@ -175,23 +222,27 @@ def generate_compliance_reports(
         generate_threatscore: Whether to generate ThreatScore report.
         generate_ens: Whether to generate ENS report.
         generate_nis2: Whether to generate NIS2 report.
+        generate_csa: Whether to generate CSA CCM report.
         only_failed_threatscore: For ThreatScore, only include failed requirements.
         min_risk_level_threatscore: Minimum risk level for ThreatScore critical requirements.
         include_manual_ens: For ENS, include manual requirements.
         include_manual_nis2: For NIS2, include manual requirements.
         only_failed_nis2: For NIS2, only include failed requirements.
+        only_failed_csa: For CSA CCM, only include failed requirements.
+        include_manual_csa: For CSA CCM, include manual requirements.
 
     Returns:
         Dictionary with results for each report type.
     """
     logger.info(
         "Generating compliance reports for scan %s with provider %s"
-        " (ThreatScore: %s, ENS: %s, NIS2: %s)",
+        " (ThreatScore: %s, ENS: %s, NIS2: %s, CSA: %s)",
         scan_id,
         provider_id,
         generate_threatscore,
         generate_ens,
         generate_nis2,
+        generate_csa,
     )
 
     results = {}
@@ -206,6 +257,8 @@ def generate_compliance_reports(
                 results["ens"] = {"upload": False, "path": ""}
             if generate_nis2:
                 results["nis2"] = {"upload": False, "path": ""}
+            if generate_csa:
+                results["csa"] = {"upload": False, "path": ""}
             return results
 
         provider_obj = Provider.objects.get(id=provider_id)
@@ -235,7 +288,23 @@ def generate_compliance_reports(
         results["nis2"] = {"upload": False, "path": ""}
         generate_nis2 = False
 
-    if not generate_threatscore and not generate_ens and not generate_nis2:
+    if generate_csa and provider_type not in [
+        "aws",
+        "azure",
+        "gcp",
+        "oraclecloud",
+        "alibabacloud",
+    ]:
+        logger.info("Provider %s not supported for CSA CCM report", provider_type)
+        results["csa"] = {"upload": False, "path": ""}
+        generate_csa = False
+
+    if (
+        not generate_threatscore
+        and not generate_ens
+        and not generate_nis2
+        and not generate_csa
+    ):
         return results
 
     # Aggregate requirement statistics once
@@ -274,6 +343,13 @@ def generate_compliance_reports(
             scan_id,
             compliance_framework="nis2",
         )
+        csa_path = _generate_compliance_output_directory(
+            DJANGO_TMP_OUTPUT_DIRECTORY,
+            provider_uid,
+            tenant_id,
+            scan_id,
+            compliance_framework="csa",
+        )
         out_dir = str(Path(threatscore_path).parent.parent)
     except Exception as e:
         logger.error("Error generating output directory: %s", e)
@@ -284,6 +360,8 @@ def generate_compliance_reports(
             results["ens"] = error_dict.copy()
         if generate_nis2:
             results["nis2"] = error_dict.copy()
+        if generate_csa:
+            results["csa"] = error_dict.copy()
         return results
 
     # Generate ThreatScore report
@@ -456,6 +534,41 @@ def generate_compliance_reports(
             logger.error("Error generating NIS2 report: %s", e)
             results["nis2"] = {"upload": False, "path": "", "error": str(e)}
 
+    # Generate CSA CCM report
+    if generate_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)
+
+        try:
+            generate_csa_report(
+                tenant_id=tenant_id,
+                scan_id=scan_id,
+                compliance_id=compliance_id_csa,
+                output_path=pdf_path_csa,
+                provider_id=provider_id,
+                only_failed=only_failed_csa,
+                include_manual=include_manual_csa,
+                provider_obj=provider_obj,
+                requirement_statistics=requirement_statistics,
+                findings_cache=findings_cache,
+            )
+
+            upload_uri_csa = _upload_to_s3(
+                tenant_id, scan_id, pdf_path_csa, f"csa/{Path(pdf_path_csa).name}"
+            )
+
+            if upload_uri_csa:
+                results["csa"] = {"upload": True, "path": upload_uri_csa}
+                logger.info("CSA CCM report uploaded to %s", upload_uri_csa)
+            else:
+                results["csa"] = {"upload": False, "path": out_dir}
+                logger.warning("CSA CCM report saved locally at %s", out_dir)
+
+        except Exception as e:
+            logger.error("Error generating CSA CCM report: %s", e)
+            results["csa"] = {"upload": False, "path": "", "error": str(e)}
+
     # Clean up temporary files if all reports were uploaded successfully
     all_uploaded = all(
         result.get("upload", False)
@@ -481,6 +594,7 @@ def generate_compliance_reports_job(
     generate_threatscore: bool = True,
     generate_ens: bool = True,
     generate_nis2: bool = True,
+    generate_csa: bool = True,
 ) -> dict[str, dict[str, bool | str]]:
     """
     Celery task wrapper for generate_compliance_reports.
@@ -492,6 +606,7 @@ def generate_compliance_reports_job(
         generate_threatscore: Whether to generate ThreatScore report.
         generate_ens: Whether to generate ENS report.
         generate_nis2: Whether to generate NIS2 report.
+        generate_csa: Whether to generate CSA CCM report.
 
     Returns:
         Dictionary with results for each report type.
@@ -503,4 +618,5 @@ def generate_compliance_reports_job(
         generate_threatscore=generate_threatscore,
         generate_ens=generate_ens,
         generate_nis2=generate_nis2,
+        generate_csa=generate_csa,
     )

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

@@ -71,6 +71,8 @@ from .config import (
     COLOR_PROWLER_DARK_GREEN,
     COLOR_SAFE,
     COLOR_WHITE,
+    CSA_CCM_SECTION_SHORT_NAMES,
+    CSA_CCM_SECTIONS,
     DIMENSION_KEYS,
     DIMENSION_MAPPING,
     DIMENSION_NAMES,
@@ -90,6 +92,7 @@ from .config import (
 )
 
 # Framework-specific generators
+from .csa import CSAReportGenerator
 from .ens import ENSReportGenerator
 from .nis2 import NIS2ReportGenerator
 from .threatscore import ThreatScoreReportGenerator
@@ -105,6 +108,7 @@ __all__ = [
     "ThreatScoreReportGenerator",
     "ENSReportGenerator",
     "NIS2ReportGenerator",
+    "CSAReportGenerator",
     # Configuration
     "FrameworkConfig",
     "FRAMEWORK_REGISTRY",
@@ -147,6 +151,8 @@ __all__ = [
     "THREATSCORE_SECTIONS",
     "NIS2_SECTIONS",
     "NIS2_SECTION_TITLES",
+    "CSA_CCM_SECTIONS",
+    "CSA_CCM_SECTION_SHORT_NAMES",
     # Layout constants
     "COL_WIDTH_SMALL",
     "COL_WIDTH_MEDIUM",

api/src/backend/tasks/jobs/reports/base.py

@@ -662,6 +662,9 @@ class BaseComplianceReportGenerator(ABC):
             elements.append(create_status_badge(req.status))
             elements.append(Spacer(1, 0.1 * inch))
 
+            # Hook for subclasses to add extra detail (e.g., CSA attributes)
+            elements.extend(self._render_requirement_detail_extras(req, data))
+
             # Findings for this requirement
             for check_id in req.checks:
                 elements.append(Paragraph(f"Check: {check_id}", self.styles["h2"]))
@@ -701,6 +704,24 @@ class BaseComplianceReportGenerator(ABC):
 
         return page_text, "Powered by Prowler"
 
+    def _render_requirement_detail_extras(
+        self, req: RequirementData, data: ComplianceData
+    ) -> list:
+        """Hook for subclasses to render extra content in detailed findings.
+
+        Called after the status badge for each requirement in the detailed
+        findings section. Override in subclasses to add framework-specific
+        metadata (e.g., CSA CCM attributes).
+
+        Args:
+            req: The requirement being rendered.
+            data: Aggregated compliance data.
+
+        Returns:
+            List of ReportLab elements (empty by default).
+        """
+        return []
+
     # =========================================================================
     # Private Helper Methods
     # =========================================================================

api/src/backend/tasks/jobs/reports/config.py

@@ -143,6 +143,36 @@ NIS2_SECTION_TITLES = {
     "12": "12. Asset Management",
 }
 
+# CSA CCM sections (Cloud Controls Matrix v4.0 domains)
+CSA_CCM_SECTIONS = [
+    "Application & Interface Security",
+    "Audit & Assurance",
+    "Business Continuity Management and Operational Resilience",
+    "Change Control and Configuration Management",
+    "Cryptography, Encryption & Key Management",
+    "Data Security and Privacy Lifecycle Management",
+    "Datacenter Security",
+    "Governance, Risk and Compliance",
+    "Identity & Access Management",
+    "Infrastructure & Virtualization Security",
+    "Interoperability & Portability",
+    "Logging and Monitoring",
+    "Security Incident Management, E-Discovery, & Cloud Forensics",
+    "Threat & Vulnerability Management",
+    "Universal Endpoint Management",
+]
+
+# Short names for CSA CCM sections (used in chart labels)
+CSA_CCM_SECTION_SHORT_NAMES = {
+    "Application & Interface Security": "App & Interface Security",
+    "Business Continuity Management and Operational Resilience": "Business Continuity",
+    "Change Control and Configuration Management": "Change Control & Config",
+    "Cryptography, Encryption & Key Management": "Cryptography & Encryption",
+    "Data Security and Privacy Lifecycle Management": "Data Security & Privacy",
+    "Security Incident Management, E-Discovery, & Cloud Forensics": "Incident Mgmt & Forensics",
+    "Infrastructure & Virtualization Security": "Infrastructure & Virtualization",
+}
+
 # Table column widths
 COL_WIDTH_SMALL = 0.4 * inch
 COL_WIDTH_MEDIUM = 0.9 * inch
@@ -261,6 +291,28 @@ FRAMEWORK_REGISTRY: dict[str, FrameworkConfig] = {
         has_niveles=False,
         has_weight=False,
     ),
+    "csa_ccm": FrameworkConfig(
+        name="csa_ccm",
+        display_name="CSA Cloud Controls Matrix (CCM)",
+        logo_filename=None,
+        primary_color=COLOR_BLUE,
+        secondary_color=COLOR_LIGHT_BLUE,
+        bg_color=COLOR_BG_BLUE,
+        attribute_fields=[
+            "Section",
+            "CCMLite",
+            "IaaS",
+            "PaaS",
+            "SaaS",
+            "ScopeApplicability",
+        ],
+        sections=CSA_CCM_SECTIONS,
+        language="en",
+        has_risk_levels=False,
+        has_dimensions=False,
+        has_niveles=False,
+        has_weight=False,
+    ),
 }
 
 
@@ -282,5 +334,7 @@ def get_framework_config(compliance_id: str) -> FrameworkConfig | None:
         return FRAMEWORK_REGISTRY["ens"]
     if "nis2" in compliance_lower:
         return FRAMEWORK_REGISTRY["nis2"]
+    if "csa" in compliance_lower or "ccm" in compliance_lower:
+        return FRAMEWORK_REGISTRY["csa_ccm"]
 
     return None

api/src/backend/tasks/jobs/reports/csa.py

@@ -0,0 +1,474 @@
+from collections import defaultdict
+
+from celery.utils.log import get_task_logger
+from reportlab.lib.units import inch
+from reportlab.platypus import Image, PageBreak, Paragraph, Spacer, Table, TableStyle
+
+from api.models import StatusChoices
+
+from .base import (
+    BaseComplianceReportGenerator,
+    ComplianceData,
+    get_requirement_metadata,
+)
+from .charts import create_horizontal_bar_chart, get_chart_color_for_percentage
+from .config import (
+    COLOR_BG_BLUE,
+    COLOR_BLUE,
+    COLOR_BORDER_GRAY,
+    COLOR_DARK_GRAY,
+    COLOR_GRID_GRAY,
+    COLOR_HIGH_RISK,
+    COLOR_SAFE,
+    COLOR_WHITE,
+    CSA_CCM_SECTION_SHORT_NAMES,
+    CSA_CCM_SECTIONS,
+)
+
+logger = get_task_logger(__name__)
+
+
+class CSAReportGenerator(BaseComplianceReportGenerator):
+    """
+    PDF report generator for CSA Cloud Controls Matrix (CCM) v4.0.
+
+    This generator creates comprehensive PDF reports containing:
+    - Cover page with Prowler logo
+    - Executive summary with overall compliance score
+    - Section analysis with horizontal bar chart
+    - Section breakdown table
+    - Requirements index organized by section
+    - Detailed findings for failed requirements
+    """
+
+    def create_executive_summary(self, data: ComplianceData) -> list:
+        """
+        Create the executive summary with compliance metrics.
+
+        Args:
+            data: Aggregated compliance data.
+
+        Returns:
+            List of ReportLab elements.
+        """
+        elements = []
+
+        elements.append(Paragraph("Executive Summary", self.styles["h1"]))
+        elements.append(Spacer(1, 0.1 * inch))
+
+        # Calculate statistics
+        total = len(data.requirements)
+        passed = sum(1 for r in data.requirements if r.status == StatusChoices.PASS)
+        failed = sum(1 for r in data.requirements if r.status == StatusChoices.FAIL)
+        manual = sum(1 for r in data.requirements if r.status == StatusChoices.MANUAL)
+
+        logger.info(
+            "CSA CCM Executive Summary: total=%d, passed=%d, failed=%d, manual=%d",
+            total,
+            passed,
+            failed,
+            manual,
+        )
+
+        # Log sample of requirements for debugging
+        for req in data.requirements[:5]:
+            logger.info(
+                "  Requirement %s: status=%s, passed_findings=%d, total_findings=%d",
+                req.id,
+                req.status,
+                req.passed_findings,
+                req.total_findings,
+            )
+
+        # Calculate compliance excluding manual
+        evaluated = passed + failed
+        overall_compliance = (passed / evaluated * 100) if evaluated > 0 else 100
+
+        # Summary statistics table
+        summary_data = [
+            ["Metric", "Value"],
+            ["Total Requirements", str(total)],
+            ["Passed \u2713", str(passed)],
+            ["Failed \u2717", str(failed)],
+            ["Manual \u2299", str(manual)],
+            ["Overall Compliance", f"{overall_compliance:.1f}%"],
+        ]
+
+        summary_table = Table(summary_data, colWidths=[3 * inch, 2 * inch])
+        summary_table.setStyle(
+            TableStyle(
+                [
+                    ("BACKGROUND", (0, 0), (-1, 0), COLOR_BLUE),
+                    ("TEXTCOLOR", (0, 0), (-1, 0), COLOR_WHITE),
+                    ("BACKGROUND", (0, 2), (0, 2), COLOR_SAFE),
+                    ("TEXTCOLOR", (0, 2), (0, 2), COLOR_WHITE),
+                    ("BACKGROUND", (0, 3), (0, 3), COLOR_HIGH_RISK),
+                    ("TEXTCOLOR", (0, 3), (0, 3), COLOR_WHITE),
+                    ("BACKGROUND", (0, 4), (0, 4), COLOR_DARK_GRAY),
+                    ("TEXTCOLOR", (0, 4), (0, 4), COLOR_WHITE),
+                    ("ALIGN", (0, 0), (-1, -1), "CENTER"),
+                    ("FONTNAME", (0, 0), (-1, 0), "PlusJakartaSans"),
+                    ("FONTSIZE", (0, 0), (-1, 0), 12),
+                    ("FONTSIZE", (0, 1), (-1, -1), 10),
+                    ("BOTTOMPADDING", (0, 0), (-1, 0), 10),
+                    ("GRID", (0, 0), (-1, -1), 0.5, COLOR_BORDER_GRAY),
+                    (
+                        "ROWBACKGROUNDS",
+                        (1, 1),
+                        (1, -1),
+                        [COLOR_WHITE, COLOR_BG_BLUE],
+                    ),
+                ]
+            )
+        )
+        elements.append(summary_table)
+
+        return elements
+
+    def create_charts_section(self, data: ComplianceData) -> list:
+        """
+        Create the charts section with section analysis.
+
+        Args:
+            data: Aggregated compliance data.
+
+        Returns:
+            List of ReportLab elements.
+        """
+        elements = []
+
+        # Section chart
+        elements.append(Paragraph("Compliance by Section", self.styles["h1"]))
+        elements.append(Spacer(1, 0.1 * inch))
+        elements.append(
+            Paragraph(
+                "The following chart shows compliance percentage for each domain "
+                "of the CSA Cloud Controls Matrix:",
+                self.styles["normal_center"],
+            )
+        )
+        elements.append(Spacer(1, 0.1 * inch))
+
+        chart_buffer = self._create_section_chart(data)
+        chart_buffer.seek(0)
+        chart_image = Image(chart_buffer, width=6.5 * inch, height=5 * inch)
+        elements.append(chart_image)
+        elements.append(PageBreak())
+
+        # Section breakdown table
+        elements.append(Paragraph("Section Breakdown", self.styles["h1"]))
+        elements.append(Spacer(1, 0.1 * inch))
+
+        section_table = self._create_section_table(data)
+        elements.append(section_table)
+
+        return elements
+
+    def create_requirements_index(self, data: ComplianceData) -> list:
+        """
+        Create the requirements index organized by section.
+
+        Args:
+            data: Aggregated compliance data.
+
+        Returns:
+            List of ReportLab elements.
+        """
+        elements = []
+
+        elements.append(Paragraph("Requirements Index", self.styles["h1"]))
+        elements.append(Spacer(1, 0.1 * inch))
+
+        # Organize by section
+        sections = {}
+        for req in data.requirements:
+            m = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
+            if m:
+                section = getattr(m, "Section", "Other")
+
+                if section not in sections:
+                    sections[section] = []
+
+                sections[section].append(
+                    {
+                        "id": req.id,
+                        "description": req.description,
+                        "status": req.status,
+                    }
+                )
+
+        # Sort by CSA CCM section order
+        for section in CSA_CCM_SECTIONS:
+            if section not in sections:
+                continue
+
+            elements.append(Paragraph(section, self.styles["h2"]))
+
+            for req in sections[section]:
+                status_indicator = (
+                    "\u2713" if req["status"] == StatusChoices.PASS else "\u2717"
+                )
+                if req["status"] == StatusChoices.MANUAL:
+                    status_indicator = "\u2299"
+
+                desc = (
+                    req["description"][:80] + "..."
+                    if len(req["description"]) > 80
+                    else req["description"]
+                )
+                elements.append(
+                    Paragraph(
+                        f"{status_indicator} <b>{req['id']}</b>: {desc}",
+                        self.styles["normal"],
+                    )
+                )
+
+            elements.append(Spacer(1, 0.1 * inch))
+
+        return elements
+
+    def _render_requirement_detail_extras(self, req, data: ComplianceData) -> list:
+        """
+        Render CSA CCM attributes in the detailed findings view.
+
+        Shows CCMLite flag, IaaS/PaaS/SaaS applicability, and
+        cross-framework references after the status badge for each requirement.
+
+        Args:
+            req: The requirement being rendered.
+            data: Aggregated compliance data.
+
+        Returns:
+            List of ReportLab elements.
+        """
+        m = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
+        if not m:
+            return []
+        return self._format_requirement_attributes(m)
+
+    def _format_requirement_attributes(self, m) -> list:
+        """
+        Format CSA CCM requirement attributes as compact PDF elements.
+
+        Displays CCMLite flag, IaaS/PaaS/SaaS applicability, and
+        cross-framework references from ScopeApplicability.
+
+        Args:
+            m: Requirement metadata (CSA_CCM_Requirement_Attribute).
+
+        Returns:
+            List of ReportLab elements.
+        """
+        elements = []
+
+        # Applicability line: CCMLite | IaaS | PaaS | SaaS
+        ccm_lite = getattr(m, "CCMLite", "")
+        iaas = getattr(m, "IaaS", "")
+        paas = getattr(m, "PaaS", "")
+        saas = getattr(m, "SaaS", "")
+
+        applicability_parts = []
+        if ccm_lite:
+            applicability_parts.append(f"CCMLite: {ccm_lite}")
+        if iaas:
+            applicability_parts.append(f"IaaS: {iaas}")
+        if paas:
+            applicability_parts.append(f"PaaS: {paas}")
+        if saas:
+            applicability_parts.append(f"SaaS: {saas}")
+
+        if applicability_parts:
+            elements.append(
+                Paragraph(
+                    f"<font color='#4A5568' size='10'>"
+                    f"{'&nbsp;&nbsp;|&nbsp;&nbsp;'.join(applicability_parts)}"
+                    f"</font>",
+                    self._attr_style(),
+                )
+            )
+
+        # ScopeApplicability references (compact)
+        scope_list = getattr(m, "ScopeApplicability", [])
+        if scope_list:
+            refs = []
+            for scope in scope_list:
+                ref_id = scope.get("ReferenceId", "") if isinstance(scope, dict) else ""
+                identifiers = (
+                    scope.get("Identifiers", []) if isinstance(scope, dict) else []
+                )
+                if ref_id and identifiers:
+                    ids_str = ", ".join(str(i) for i in identifiers[:4])
+                    if len(identifiers) > 4:
+                        ids_str += "..."
+                    refs.append(f"{ref_id}: {ids_str}")
+
+            if refs:
+                refs_text = "&nbsp;&nbsp;|&nbsp;&nbsp;".join(refs)
+                elements.append(
+                    Paragraph(
+                        f"<font color='#718096' size='9'>{refs_text}</font>",
+                        self._attr_style(),
+                    )
+                )
+
+        return elements
+
+    def _attr_style(self):
+        """
+        Return a compact style for attribute text lines.
+
+        Returns:
+            ParagraphStyle for attribute display.
+        """
+        from reportlab.lib.styles import ParagraphStyle
+
+        return ParagraphStyle(
+            "AttrLine",
+            parent=self.styles["normal"],
+            fontSize=10,
+            spaceBefore=2,
+            spaceAfter=2,
+            leftIndent=30,
+            leading=13,
+        )
+
+    def _create_section_chart(self, data: ComplianceData):
+        """
+        Create the section compliance chart.
+
+        Args:
+            data: Aggregated compliance data.
+
+        Returns:
+            BytesIO buffer containing the chart image.
+        """
+        section_scores = defaultdict(lambda: {"passed": 0, "total": 0})
+
+        no_metadata_count = 0
+        for req in data.requirements:
+            if req.status == StatusChoices.MANUAL:
+                continue
+
+            m = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
+            if m:
+                section = getattr(m, "Section", "Other")
+                section_scores[section]["total"] += 1
+                if req.status == StatusChoices.PASS:
+                    section_scores[section]["passed"] += 1
+            else:
+                no_metadata_count += 1
+
+        if no_metadata_count > 0:
+            logger.warning(
+                "CSA CCM chart: %d requirements had no metadata", no_metadata_count
+            )
+
+        logger.info("CSA CCM section scores:")
+        for section in CSA_CCM_SECTIONS:
+            if section in section_scores:
+                scores = section_scores[section]
+                pct = (
+                    (scores["passed"] / scores["total"] * 100)
+                    if scores["total"] > 0
+                    else 0
+                )
+                logger.info(
+                    "  %s: %d/%d (%.1f%%)",
+                    section,
+                    scores["passed"],
+                    scores["total"],
+                    pct,
+                )
+
+        # Build labels and values in CSA CCM section order
+        labels = []
+        values = []
+        for section in CSA_CCM_SECTIONS:
+            if section in section_scores and section_scores[section]["total"] > 0:
+                scores = section_scores[section]
+                pct = (scores["passed"] / scores["total"]) * 100
+                # Use short name if available
+                label = CSA_CCM_SECTION_SHORT_NAMES.get(section, section)
+                labels.append(label)
+                values.append(pct)
+
+        return create_horizontal_bar_chart(
+            labels=labels,
+            values=values,
+            xlabel="Compliance (%)",
+            color_func=get_chart_color_for_percentage,
+        )
+
+    def _create_section_table(self, data: ComplianceData) -> Table:
+        """
+        Create the section breakdown table.
+
+        Args:
+            data: Aggregated compliance data.
+
+        Returns:
+            ReportLab Table element.
+        """
+        section_scores = defaultdict(lambda: {"passed": 0, "failed": 0, "manual": 0})
+
+        for req in data.requirements:
+            m = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
+            if m:
+                section = getattr(m, "Section", "Other")
+
+                if req.status == StatusChoices.PASS:
+                    section_scores[section]["passed"] += 1
+                elif req.status == StatusChoices.FAIL:
+                    section_scores[section]["failed"] += 1
+                else:
+                    section_scores[section]["manual"] += 1
+
+        table_data = [["Section", "Passed", "Failed", "Manual", "Compliance"]]
+        for section in CSA_CCM_SECTIONS:
+            if section not in section_scores:
+                continue
+            scores = section_scores[section]
+            total = scores["passed"] + scores["failed"]
+            pct = (scores["passed"] / total * 100) if total > 0 else 100
+            # Use short name if available
+            label = CSA_CCM_SECTION_SHORT_NAMES.get(section, section)
+            table_data.append(
+                [
+                    label,
+                    str(scores["passed"]),
+                    str(scores["failed"]),
+                    str(scores["manual"]),
+                    f"{pct:.1f}%",
+                ]
+            )
+
+        table = Table(
+            table_data,
+            colWidths=[2.4 * inch, 0.9 * inch, 0.9 * inch, 0.9 * inch, 1.2 * inch],
+        )
+        table.setStyle(
+            TableStyle(
+                [
+                    ("BACKGROUND", (0, 0), (-1, 0), COLOR_BLUE),
+                    ("TEXTCOLOR", (0, 0), (-1, 0), COLOR_WHITE),
+                    ("FONTNAME", (0, 0), (-1, 0), "FiraCode"),
+                    ("FONTSIZE", (0, 0), (-1, 0), 10),
+                    ("ALIGN", (0, 0), (-1, -1), "CENTER"),
+                    ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
+                    ("FONTSIZE", (0, 1), (-1, -1), 9),
+                    ("GRID", (0, 0), (-1, -1), 0.5, COLOR_GRID_GRAY),
+                    ("LEFTPADDING", (0, 0), (-1, -1), 6),
+                    ("RIGHTPADDING", (0, 0), (-1, -1), 6),
+                    ("TOPPADDING", (0, 0), (-1, -1), 4),
+                    ("BOTTOMPADDING", (0, 0), (-1, -1), 4),
+                    (
+                        "ROWBACKGROUNDS",
+                        (0, 1),
+                        (-1, -1),
+                        [COLOR_WHITE, COLOR_BG_BLUE],
+                    ),
+                ]
+            )
+        )
+
+        return table

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

@@ -114,6 +114,11 @@ def _calculate_requirements_data_from_statistics(
                 requirement_status = StatusChoices.PASS
             else:
                 requirement_status = StatusChoices.FAIL
+        elif requirement_checks:
+            # Requirement has checks but none produced findings — consistent
+            # with the dashboard's scan processing which treats this as PASS
+            # (no failed checks means the requirement is considered compliant).
+            requirement_status = StatusChoices.PASS
         else:
             requirement_status = StatusChoices.MANUAL
 

api/src/backend/tasks/tasks.py

@@ -11,6 +11,7 @@ from django_celery_beat.models import PeriodicTask
 from tasks.jobs.attack_paths import (
     attack_paths_scan,
     can_provider_run_attack_paths_scan,
+    db_utils as attack_paths_db_utils,
 )
 from tasks.jobs.backfill import (
     backfill_compliance_summaries,
@@ -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",
@@ -888,11 +906,11 @@ def jira_integration_task(
 @handle_provider_deletion
 def generate_compliance_reports_task(tenant_id: str, scan_id: str, provider_id: str):
     """
-    Optimized task to generate ThreatScore, ENS, and NIS2 reports with shared queries.
+    Optimized task to generate ThreatScore, ENS, NIS2, and CSA CCM reports with shared queries.
 
     This task is more efficient than running separate report tasks because it reuses database queries:
-    - Provider object fetched once (instead of three times)
-    - Requirement statistics aggregated once (instead of three times)
+    - Provider object fetched once (instead of multiple times)
+    - Requirement statistics aggregated once (instead of multiple times)
     - Can reduce database load by up to 50-70%
 
     Args:
@@ -910,6 +928,7 @@ def generate_compliance_reports_task(tenant_id: str, scan_id: str, provider_id:
         generate_threatscore=True,
         generate_ens=True,
         generate_nis2=True,
+        generate_csa=True,
     )
 
 

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

@@ -4,6 +4,7 @@ from unittest.mock import MagicMock, call, patch
 
 import pytest
 from tasks.jobs.attack_paths import findings as findings_module
+from tasks.jobs.attack_paths import internet as internet_module
 from tasks.jobs.attack_paths.scan import run as attack_paths_run
 
 from api.models import (
@@ -27,16 +28,15 @@ class TestAttackPathsRun:
         "tasks.jobs.attack_paths.scan.utils.call_within_event_loop",
         side_effect=lambda fn, *a, **kw: fn(*a, **kw),
     )
-    @patch(
-        "tasks.jobs.attack_paths.scan.db_utils.get_old_attack_paths_scans",
-        return_value=[],
-    )
+    @patch("tasks.jobs.attack_paths.scan.db_utils.set_graph_data_ready")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
     @patch("tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
     @patch("tasks.jobs.attack_paths.scan.sync.sync_graph")
     @patch("tasks.jobs.attack_paths.scan.graph_database.drop_subgraph")
     @patch("tasks.jobs.attack_paths.scan.sync.create_sync_indexes")
+    @patch("tasks.jobs.attack_paths.scan.internet.analysis")
     @patch("tasks.jobs.attack_paths.scan.findings.analysis")
     @patch("tasks.jobs.attack_paths.scan.findings.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
@@ -67,13 +67,15 @@ class TestAttackPathsRun:
         mock_cartography_ontology,
         mock_findings_indexes,
         mock_findings_analysis,
+        mock_internet_analysis,
         mock_sync_indexes,
         mock_drop_subgraph,
         mock_sync,
         mock_starting,
         mock_update_progress,
         mock_finish,
-        mock_get_old_scans,
+        mock_set_provider_graph_data_ready,
+        mock_set_graph_data_ready,
         mock_event_loop,
         mock_drop_db,
         tenants_fixture,
@@ -139,6 +141,7 @@ class TestAttackPathsRun:
         # These use tmp_cartography_config (neo4j_database="db-scan-id")
         mock_cartography_analysis.assert_called_once()
         mock_cartography_ontology.assert_called_once()
+        mock_internet_analysis.assert_called_once()
         mock_findings_analysis.assert_called_once()
         mock_drop_subgraph.assert_called_once_with(
             database="tenant-db",
@@ -160,9 +163,66 @@ class TestAttackPathsRun:
         mock_finish.assert_called_once_with(
             attack_paths_scan, StateChoices.COMPLETED, ingestion_result
         )
+        mock_set_provider_graph_data_ready.assert_called_once_with(
+            attack_paths_scan, False
+        )
+        mock_set_graph_data_ready.assert_called_once_with(attack_paths_scan, True)
 
+    @patch(
+        "tasks.jobs.attack_paths.scan.utils.stringify_exception",
+        return_value="Cartography failed: ingestion boom",
+    )
+    @patch(
+        "tasks.jobs.attack_paths.scan.utils.call_within_event_loop",
+        side_effect=lambda fn, *a, **kw: fn(*a, **kw),
+    )
+    @patch("tasks.jobs.attack_paths.scan.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.set_graph_data_ready")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.internet.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.create_findings_indexes")
+    @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
+    @patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run")
+    @patch("tasks.jobs.attack_paths.scan.graph_database.create_database")
+    @patch(
+        "tasks.jobs.attack_paths.scan.graph_database.get_database_name",
+        return_value="db-scan-id",
+    )
+    @patch("tasks.jobs.attack_paths.scan.graph_database.get_uri")
+    @patch(
+        "tasks.jobs.attack_paths.scan.initialize_prowler_provider",
+        return_value=MagicMock(_enabled_regions=["us-east-1"]),
+    )
+    @patch(
+        "tasks.jobs.attack_paths.scan.rls_transaction",
+        new=lambda *args, **kwargs: nullcontext(),
+    )
     def test_run_failure_marks_scan_failed(
-        self, tenants_fixture, providers_fixture, scans_fixture
+        self,
+        mock_init_provider,
+        mock_get_uri,
+        mock_get_db_name,
+        mock_create_db,
+        mock_cartography_indexes,
+        mock_cartography_analysis,
+        mock_findings_indexes,
+        mock_internet_analysis,
+        mock_findings_analysis,
+        mock_starting,
+        mock_update_progress,
+        mock_set_provider_graph_data_ready,
+        mock_set_graph_data_ready,
+        mock_finish,
+        mock_drop_db,
+        mock_event_loop,
+        mock_stringify,
+        tenants_fixture,
+        providers_fixture,
+        scans_fixture,
     ):
         tenant = tenants_fixture[0]
         provider = providers_fixture[0]
@@ -186,52 +246,18 @@ class TestAttackPathsRun:
         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.findings.create_findings_indexes"),
-            patch("tasks.jobs.attack_paths.scan.findings.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.graph_database.drop_database"),
             patch(
                 "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function",
                 return_value=ingestion_fn,
             ),
-            patch(
-                "tasks.jobs.attack_paths.scan.utils.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")
@@ -239,9 +265,109 @@ class TestAttackPathsRun:
         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"
-        }
+        assert failure_args[2] == {"global_error": "Cartography failed: ingestion boom"}
+
+    @patch(
+        "tasks.jobs.attack_paths.scan.utils.stringify_exception",
+        return_value="Cartography failed: ingestion boom",
+    )
+    @patch(
+        "tasks.jobs.attack_paths.scan.utils.call_within_event_loop",
+        side_effect=lambda fn, *a, **kw: fn(*a, **kw),
+    )
+    @patch(
+        "tasks.jobs.attack_paths.scan.graph_database.drop_database",
+        side_effect=ConnectionError("neo4j down"),
+    )
+    @patch("tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.set_graph_data_ready")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
+    @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.internet.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.create_findings_indexes")
+    @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
+    @patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run")
+    @patch("tasks.jobs.attack_paths.scan.graph_database.create_database")
+    @patch(
+        "tasks.jobs.attack_paths.scan.graph_database.get_database_name",
+        return_value="db-scan-id",
+    )
+    @patch("tasks.jobs.attack_paths.scan.graph_database.get_uri")
+    @patch(
+        "tasks.jobs.attack_paths.scan.initialize_prowler_provider",
+        return_value=MagicMock(_enabled_regions=["us-east-1"]),
+    )
+    @patch(
+        "tasks.jobs.attack_paths.scan.rls_transaction",
+        new=lambda *args, **kwargs: nullcontext(),
+    )
+    def test_run_failure_marks_scan_failed_even_when_drop_database_fails(
+        self,
+        mock_init_provider,
+        mock_get_uri,
+        mock_get_db_name,
+        mock_create_db,
+        mock_cartography_indexes,
+        mock_cartography_analysis,
+        mock_findings_indexes,
+        mock_internet_analysis,
+        mock_findings_analysis,
+        mock_starting,
+        mock_update_progress,
+        mock_set_provider_graph_data_ready,
+        mock_set_graph_data_ready,
+        mock_finish,
+        mock_drop_db,
+        mock_event_loop,
+        mock_stringify,
+        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.graph_database.get_session",
+                return_value=session_ctx,
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.db_utils.retrieve_attack_paths_scan",
+                return_value=attack_paths_scan,
+            ),
+            patch(
+                "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function",
+                return_value=ingestion_fn,
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="ingestion boom"):
+                attack_paths_run(str(tenant.id), str(scan.id), "task-789")
+
+        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_error": "Cartography failed: ingestion boom"}
 
     def test_run_returns_early_for_unsupported_provider(self, tenants_fixture):
         tenant = tenants_fixture[0]
@@ -286,6 +412,194 @@ class TestAttackPathsRun:
         mock_retrieve.assert_called_once_with(str(tenant.id), str(scan.id))
 
 
+@pytest.mark.django_db
+class TestFailAttackPathsScan:
+    def test_marks_executing_scan_as_failed(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import (
+            fail_attack_paths_scan,
+        )
+
+        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.EXECUTING,
+        )
+
+        with (
+            patch(
+                "tasks.jobs.attack_paths.db_utils.retrieve_attack_paths_scan",
+                return_value=attack_paths_scan,
+            ) as mock_retrieve,
+            patch(
+                "tasks.jobs.attack_paths.db_utils.graph_database.drop_database"
+            ) as mock_drop_db,
+            patch(
+                "tasks.jobs.attack_paths.db_utils.finish_attack_paths_scan"
+            ) as mock_finish,
+        ):
+            fail_attack_paths_scan(str(tenant.id), str(scan.id), "setup exploded")
+
+        mock_retrieve.assert_called_once_with(str(tenant.id), str(scan.id))
+        expected_tmp_db = f"db-tmp-scan-{str(attack_paths_scan.id).lower()}"
+        mock_drop_db.assert_called_once_with(expected_tmp_db)
+        mock_finish.assert_called_once_with(
+            attack_paths_scan,
+            StateChoices.FAILED,
+            {"global_error": "setup exploded"},
+        )
+
+    def test_drops_temp_database_even_when_drop_fails(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import (
+            fail_attack_paths_scan,
+        )
+
+        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.EXECUTING,
+        )
+
+        with (
+            patch(
+                "tasks.jobs.attack_paths.db_utils.retrieve_attack_paths_scan",
+                return_value=attack_paths_scan,
+            ),
+            patch(
+                "tasks.jobs.attack_paths.db_utils.graph_database.drop_database",
+                side_effect=Exception("Neo4j unreachable"),
+            ),
+            patch(
+                "tasks.jobs.attack_paths.db_utils.finish_attack_paths_scan"
+            ) as mock_finish,
+        ):
+            fail_attack_paths_scan(str(tenant.id), str(scan.id), "setup exploded")
+
+        mock_finish.assert_called_once_with(
+            attack_paths_scan,
+            StateChoices.FAILED,
+            {"global_error": "setup exploded"},
+        )
+
+    def test_skips_already_failed_scan(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import (
+            fail_attack_paths_scan,
+        )
+
+        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.FAILED,
+        )
+
+        with (
+            patch(
+                "tasks.jobs.attack_paths.db_utils.retrieve_attack_paths_scan",
+                return_value=attack_paths_scan,
+            ),
+            patch(
+                "tasks.jobs.attack_paths.db_utils.graph_database.drop_database"
+            ) as mock_drop_db,
+            patch(
+                "tasks.jobs.attack_paths.db_utils.finish_attack_paths_scan"
+            ) as mock_finish,
+        ):
+            fail_attack_paths_scan(str(tenant.id), str(scan.id), "setup exploded")
+
+        mock_drop_db.assert_not_called()
+        mock_finish.assert_not_called()
+
+    def test_skips_when_no_scan_found(self, tenants_fixture):
+        from tasks.jobs.attack_paths.db_utils import (
+            fail_attack_paths_scan,
+        )
+
+        tenant = tenants_fixture[0]
+
+        with (
+            patch(
+                "tasks.jobs.attack_paths.db_utils.retrieve_attack_paths_scan",
+                return_value=None,
+            ),
+            patch(
+                "tasks.jobs.attack_paths.db_utils.finish_attack_paths_scan"
+            ) as mock_finish,
+        ):
+            fail_attack_paths_scan(str(tenant.id), "nonexistent", "setup exploded")
+
+        mock_finish.assert_not_called()
+
+
+class TestAttackPathsScanRLSTaskOnFailure:
+    def test_on_failure_delegates_to_fail_attack_paths_scan(self):
+        from tasks.tasks import AttackPathsScanRLSTask
+
+        task = AttackPathsScanRLSTask()
+
+        with patch(
+            "tasks.tasks.attack_paths_db_utils.fail_attack_paths_scan"
+        ) as mock_fail:
+            task.on_failure(
+                exc=RuntimeError("boom"),
+                task_id="task-abc",
+                args=(),
+                kwargs={"tenant_id": "t-1", "scan_id": "s-1"},
+                _einfo=None,
+            )
+
+        mock_fail.assert_called_once_with("t-1", "s-1", "boom")
+
+    def test_on_failure_skips_when_missing_kwargs(self):
+        from tasks.tasks import AttackPathsScanRLSTask
+
+        task = AttackPathsScanRLSTask()
+
+        with patch(
+            "tasks.tasks.attack_paths_db_utils.fail_attack_paths_scan"
+        ) as mock_fail:
+            task.on_failure(
+                exc=RuntimeError("boom"),
+                task_id="task-abc",
+                args=(),
+                kwargs={},
+                _einfo=None,
+            )
+
+        mock_fail.assert_not_called()
+
+
 @pytest.mark.django_db
 class TestAttackPathsFindingsHelpers:
     def test_create_findings_indexes_executes_all_statements(self):
@@ -757,3 +1071,359 @@ class TestAttackPathsFindingsHelpers:
             findings_module.load_findings(mock_session, empty_gen(), provider, config)
 
         mock_session.run.assert_not_called()
+
+
+class TestInternetAnalysis:
+    def _make_provider_and_config(self):
+        provider = MagicMock()
+        provider.provider = "aws"
+        provider.uid = "123456789012"
+        config = SimpleNamespace(update_tag=1234567890)
+        return provider, config
+
+    def test_analysis_creates_node_and_relationships(self):
+        """Verify both Cypher statements are executed and relationship count returned."""
+        mock_session = MagicMock()
+        mock_result = MagicMock()
+        mock_result.single.return_value = {"relationships_merged": 3}
+        mock_session.run.side_effect = [None, mock_result]
+        provider, config = self._make_provider_and_config()
+
+        with patch(
+            "tasks.jobs.attack_paths.internet.get_root_node_label",
+            return_value="AWSAccount",
+        ):
+            result = internet_module.analysis(mock_session, provider, config)
+
+        assert mock_session.run.call_count == 2
+        assert result == 3
+
+    def test_analysis_zero_exposed_resources(self):
+        """When no resources are exposed, zero relationships are created."""
+        mock_session = MagicMock()
+        mock_result = MagicMock()
+        mock_result.single.return_value = {"relationships_merged": 0}
+        mock_session.run.side_effect = [None, mock_result]
+        provider, config = self._make_provider_and_config()
+
+        with patch(
+            "tasks.jobs.attack_paths.internet.get_root_node_label",
+            return_value="AWSAccount",
+        ):
+            result = internet_module.analysis(mock_session, provider, config)
+
+        assert result == 0
+
+
+@pytest.mark.django_db
+class TestAttackPathsDbUtilsGraphDataReady:
+    """Tests for db_utils functions related to graph_data_ready lifecycle."""
+
+    def test_create_attack_paths_scan_first_scan_defaults_to_false(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import create_attack_paths_scan
+
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+        scan = scans_fixture[0]
+        scan.provider = provider
+        scan.save()
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            attack_paths_scan = create_attack_paths_scan(
+                str(tenant.id), str(scan.id), provider.id
+            )
+
+        assert attack_paths_scan is not None
+        assert attack_paths_scan.graph_data_ready is False
+
+    def test_create_attack_paths_scan_inherits_true_from_previous(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import create_attack_paths_scan
+
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+        scan = scans_fixture[0]
+        scan.provider = provider
+        scan.save()
+
+        AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider,
+            scan=scan,
+            state=StateChoices.COMPLETED,
+            graph_data_ready=True,
+        )
+
+        new_scan = Scan.objects.create(
+            name="New Scan",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.AVAILABLE,
+            tenant_id=tenant.id,
+        )
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            attack_paths_scan = create_attack_paths_scan(
+                str(tenant.id), str(new_scan.id), provider.id
+            )
+
+        assert attack_paths_scan is not None
+        assert attack_paths_scan.graph_data_ready is True
+
+    def test_create_attack_paths_scan_inherits_false_when_no_previous_ready(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import create_attack_paths_scan
+
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+        scan = scans_fixture[0]
+        scan.provider = provider
+        scan.save()
+
+        AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider,
+            scan=scan,
+            state=StateChoices.FAILED,
+            graph_data_ready=False,
+        )
+
+        new_scan = Scan.objects.create(
+            name="New Scan",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.AVAILABLE,
+            tenant_id=tenant.id,
+        )
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            attack_paths_scan = create_attack_paths_scan(
+                str(tenant.id), str(new_scan.id), provider.id
+            )
+
+        assert attack_paths_scan is not None
+        assert attack_paths_scan.graph_data_ready is False
+
+    def test_set_graph_data_ready_updates_field(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import set_graph_data_ready
+
+        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.EXECUTING,
+            graph_data_ready=True,
+        )
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            set_graph_data_ready(attack_paths_scan, False)
+
+        attack_paths_scan.refresh_from_db()
+        assert attack_paths_scan.graph_data_ready is False
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            set_graph_data_ready(attack_paths_scan, True)
+
+        attack_paths_scan.refresh_from_db()
+        assert attack_paths_scan.graph_data_ready is True
+
+    def test_finish_attack_paths_scan_does_not_modify_graph_data_ready(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import finish_attack_paths_scan
+
+        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.EXECUTING,
+            graph_data_ready=True,
+        )
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            finish_attack_paths_scan(attack_paths_scan, StateChoices.COMPLETED, {})
+
+        attack_paths_scan.refresh_from_db()
+        assert attack_paths_scan.state == StateChoices.COMPLETED
+        assert attack_paths_scan.graph_data_ready is True
+
+    def test_finish_attack_paths_scan_preserves_graph_data_ready_on_failure(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import finish_attack_paths_scan
+
+        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.EXECUTING,
+            graph_data_ready=True,
+        )
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            finish_attack_paths_scan(
+                attack_paths_scan,
+                StateChoices.FAILED,
+                {"global_error": "boom"},
+            )
+
+        attack_paths_scan.refresh_from_db()
+        assert attack_paths_scan.state == StateChoices.FAILED
+        assert attack_paths_scan.graph_data_ready is True
+
+    def test_set_provider_graph_data_ready_updates_all_scans_for_provider(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import set_provider_graph_data_ready
+
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+
+        scan_a = scans_fixture[0]
+        scan_a.provider = provider
+        scan_a.save()
+
+        scan_b = Scan.objects.create(
+            name="Second Scan",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.AVAILABLE,
+            tenant_id=tenant.id,
+        )
+
+        old_ap_scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider,
+            scan=scan_a,
+            state=StateChoices.COMPLETED,
+            graph_data_ready=True,
+        )
+        new_ap_scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider,
+            scan=scan_b,
+            state=StateChoices.EXECUTING,
+            graph_data_ready=True,
+        )
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            set_provider_graph_data_ready(new_ap_scan, False)
+
+        old_ap_scan.refresh_from_db()
+        new_ap_scan.refresh_from_db()
+        assert old_ap_scan.graph_data_ready is False
+        assert new_ap_scan.graph_data_ready is False
+
+    def test_set_provider_graph_data_ready_does_not_affect_other_providers(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        from tasks.jobs.attack_paths.db_utils import set_provider_graph_data_ready
+
+        tenant = tenants_fixture[0]
+        provider_a = providers_fixture[0]
+        provider_a.provider = Provider.ProviderChoices.AWS
+        provider_a.save()
+
+        provider_b = providers_fixture[1]
+        provider_b.provider = Provider.ProviderChoices.AWS
+        provider_b.save()
+
+        scan_a = scans_fixture[0]
+        scan_a.provider = provider_a
+        scan_a.save()
+
+        scan_b = Scan.objects.create(
+            name="Scan for provider B",
+            provider=provider_b,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant_id=tenant.id,
+        )
+
+        ap_scan_a = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider_a,
+            scan=scan_a,
+            state=StateChoices.EXECUTING,
+            graph_data_ready=True,
+        )
+        ap_scan_b = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=provider_b,
+            scan=scan_b,
+            state=StateChoices.COMPLETED,
+            graph_data_ready=True,
+        )
+
+        with patch(
+            "tasks.jobs.attack_paths.db_utils.rls_transaction",
+            new=lambda *args, **kwargs: nullcontext(),
+        ):
+            set_provider_graph_data_ready(ap_scan_a, False)
+
+        ap_scan_a.refresh_from_db()
+        ap_scan_b.refresh_from_db()
+        assert ap_scan_a.graph_data_ready is False
+        assert ap_scan_b.graph_data_ready is True

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

@@ -4,6 +4,7 @@ import pytest
 
 from django.core.exceptions import ObjectDoesNotExist
 
+from api.attack_paths import database as graph_database
 from api.models import Provider, Tenant
 from tasks.jobs.deletion import delete_provider, delete_tenant
 
@@ -47,14 +48,61 @@ class TestDeleteProvider:
             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)
+            result = delete_provider(tenant_id, non_existent_pk)
 
-            mock_get_database_name.assert_called_once_with(tenant_id)
-            mock_drop_subgraph.assert_called_once_with(
-                "tenant-db",
-                non_existent_pk,
-            )
+            assert result == {}
+            mock_get_database_name.assert_not_called()
+            mock_drop_subgraph.assert_not_called()
+
+    def test_delete_provider_drops_temp_attack_paths_databases(
+        self, providers_fixture, create_attack_paths_scan
+    ):
+        instance = providers_fixture[0]
+        tenant_id = str(instance.tenant_id)
+
+        aps1 = create_attack_paths_scan(instance)
+        aps2 = create_attack_paths_scan(instance)
+
+        with (
+            patch(
+                "tasks.jobs.deletion.graph_database.drop_subgraph",
+            ),
+            patch(
+                "tasks.jobs.deletion.graph_database.drop_database",
+            ) as mock_drop_database,
+        ):
+            result = delete_provider(tenant_id, instance.id)
+
+        assert result
+        expected_tmp_calls = [
+            call(f"db-tmp-scan-{str(aps1.id).lower()}"),
+            call(f"db-tmp-scan-{str(aps2.id).lower()}"),
+        ]
+        mock_drop_database.assert_has_calls(expected_tmp_calls, any_order=True)
+
+    def test_delete_provider_continues_when_temp_db_drop_fails(
+        self, providers_fixture, create_attack_paths_scan
+    ):
+        instance = providers_fixture[0]
+        tenant_id = str(instance.tenant_id)
+
+        create_attack_paths_scan(instance)
+
+        with (
+            patch(
+                "tasks.jobs.deletion.graph_database.drop_subgraph",
+            ),
+            patch(
+                "tasks.jobs.deletion.graph_database.drop_database",
+                side_effect=graph_database.GraphDatabaseQueryException(
+                    "Neo4j unreachable"
+                ),
+            ),
+        ):
+            result = delete_provider(tenant_id, instance.id)
+
+        assert result
+        assert not Provider.all_objects.filter(pk=instance.id).exists()
 
 
 @pytest.mark.django_db
@@ -142,3 +190,56 @@ class TestDeleteTenant:
             mock_get_database_name.assert_called_once_with(tenant.id)
             mock_drop_subgraph.assert_not_called()
             mock_drop_database.assert_called_once_with("tenant-db")
+
+    def test_delete_tenant_includes_soft_deleted_providers(self, tenants_fixture):
+        tenant = tenants_fixture[0]
+        provider = Provider.objects.create(
+            provider="aws",
+            uid="999999999999",
+            alias="soft_deleted_provider",
+            tenant_id=tenant.id,
+        )
+        # Soft-delete the provider so ActiveProviderManager would skip it
+        Provider.all_objects.filter(pk=provider.id).update(is_deleted=True)
+
+        with (
+            patch(
+                "tasks.jobs.deletion.graph_database.get_database_name",
+                return_value="tenant-db",
+            ),
+            patch(
+                "tasks.jobs.deletion.graph_database.drop_subgraph"
+            ) as mock_drop_subgraph,
+            patch("tasks.jobs.deletion.graph_database.drop_database"),
+        ):
+            delete_tenant(tenant.id)
+
+            mock_drop_subgraph.assert_any_call("tenant-db", str(provider.id))
+
+    def test_delete_tenant_handles_concurrently_deleted_provider(self, tenants_fixture):
+        tenant = tenants_fixture[0]
+        Provider.objects.create(
+            provider="aws",
+            uid="111111111111",
+            alias="vanishing_provider",
+            tenant_id=tenant.id,
+        )
+
+        def drop_subgraph_side_effect(_db_name, provider_id):
+            # Simulate concurrent deletion by another process
+            Provider.all_objects.filter(pk=provider_id).delete()
+
+        with (
+            patch(
+                "tasks.jobs.deletion.graph_database.get_database_name",
+                return_value="tenant-db",
+            ),
+            patch(
+                "tasks.jobs.deletion.graph_database.drop_subgraph",
+                side_effect=drop_subgraph_side_effect,
+            ),
+            patch("tasks.jobs.deletion.graph_database.drop_database"),
+        ):
+            deletion_summary = delete_tenant(tenant.id)
+
+            assert deletion_summary is not None