feat(api): replace _provider_id property with label-based isolation and regex injection for custom queries (#10402)

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

@@ -63,11 +63,9 @@ INTERNAL_LABELS: list[str] = [
 ]
 
 # Provider isolation properties
-PROVIDER_ID_PROPERTY = "_provider_id"
 PROVIDER_ELEMENT_ID_PROPERTY = "_provider_element_id"
 
 PROVIDER_ISOLATION_PROPERTIES: list[str] = [
-    PROVIDER_ID_PROPERTY,
     PROVIDER_ELEMENT_ID_PROPERTY,
 ]
 

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

@@ -22,7 +22,6 @@ from tasks.jobs.attack_paths.config import (
     get_provider_resource_label,
     get_root_node_label,
 )
-from tasks.jobs.attack_paths.indexes import IndexType, create_indexes
 from tasks.jobs.attack_paths.queries import (
     ADD_RESOURCE_LABEL_TEMPLATE,
     CLEANUP_FINDINGS_TEMPLATE,
@@ -84,11 +83,6 @@ def _to_neo4j_dict(record: dict[str, Any], resource_uid: str) -> dict[str, Any]:
 # ----------
 
 
-def create_findings_indexes(neo4j_session: neo4j.Session) -> None:
-    """Create indexes for Prowler findings and resource lookups."""
-    create_indexes(neo4j_session, IndexType.FINDINGS)
-
-
 def analysis(
     neo4j_session: neo4j.Session,
     prowler_api_provider: Provider,
@@ -196,7 +190,6 @@ def cleanup_findings(
 ) -> None:
     """Remove stale findings (classic Cartography behaviour)."""
     parameters = {
-        "provider_uid": str(prowler_api_provider.uid),
         "last_updated": config.update_tag,
         "batch_size": BATCH_SIZE,
     }

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

@@ -1,5 +1,3 @@
-from enum import Enum
-
 import neo4j
 
 from cartography.client.core.tx import run_write_query
@@ -9,20 +7,12 @@ from tasks.jobs.attack_paths.config import (
     INTERNET_NODE_LABEL,
     PROWLER_FINDING_LABEL,
     PROVIDER_ELEMENT_ID_PROPERTY,
-    PROVIDER_ID_PROPERTY,
     PROVIDER_RESOURCE_LABEL,
 )
 
 logger = get_task_logger(__name__)
 
 
-class IndexType(Enum):
-    """Types of indexes that can be created."""
-
-    FINDINGS = "findings"
-    SYNC = "sync"
-
-
 # Indexes for Prowler findings and resource lookups
 FINDINGS_INDEX_STATEMENTS = [
     # Resource indexes for Prowler Finding lookups
@@ -30,7 +20,6 @@ FINDINGS_INDEX_STATEMENTS = [
     "CREATE INDEX aws_resource_id IF NOT EXISTS FOR (n:_AWSResource) ON (n.id);",
     # Prowler Finding indexes
     f"CREATE INDEX prowler_finding_id IF NOT EXISTS FOR (n:{PROWLER_FINDING_LABEL}) ON (n.id);",
-    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
@@ -40,30 +29,18 @@ FINDINGS_INDEX_STATEMENTS = [
 # Indexes for provider resource sync operations
 SYNC_INDEX_STATEMENTS = [
     f"CREATE INDEX provider_resource_element_id IF NOT EXISTS FOR (n:{PROVIDER_RESOURCE_LABEL}) ON (n.{PROVIDER_ELEMENT_ID_PROPERTY});",
-    f"CREATE INDEX provider_resource_provider_id IF NOT EXISTS FOR (n:{PROVIDER_RESOURCE_LABEL}) ON (n.{PROVIDER_ID_PROPERTY});",
 ]
 
 
-def create_indexes(neo4j_session: neo4j.Session, index_type: IndexType) -> None:
-    """
-    Create indexes for the specified type.
-
-    Args:
-        `neo4j_session`: The Neo4j session to use
-        `index_type`: The type of indexes to create (FINDINGS or SYNC)
-    """
-    if index_type == IndexType.FINDINGS:
-        logger.info("Creating indexes for Prowler Findings node types")
-        for statement in FINDINGS_INDEX_STATEMENTS:
-            run_write_query(neo4j_session, statement)
-
-    elif index_type == IndexType.SYNC:
-        logger.info("Ensuring ProviderResource indexes exist")
-        for statement in SYNC_INDEX_STATEMENTS:
-            neo4j_session.run(statement)
+def create_findings_indexes(neo4j_session: neo4j.Session) -> None:
+    """Create indexes for Prowler findings and resource lookups."""
+    logger.info("Creating indexes for Prowler Findings node types")
+    for statement in FINDINGS_INDEX_STATEMENTS:
+        run_write_query(neo4j_session, statement)
 
 
-def create_all_indexes(neo4j_session: neo4j.Session) -> None:
-    """Create all indexes (both findings and sync)."""
-    create_indexes(neo4j_session, IndexType.FINDINGS)
-    create_indexes(neo4j_session, IndexType.SYNC)
+def create_sync_indexes(neo4j_session: neo4j.Session) -> None:
+    """Create indexes for provider resource sync operations."""
+    logger.info("Ensuring ProviderResource indexes exist")
+    for statement in SYNC_INDEX_STATEMENTS:
+        neo4j_session.run(statement)

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

@@ -3,7 +3,6 @@ from tasks.jobs.attack_paths.config import (
     INTERNET_NODE_LABEL,
     PROWLER_FINDING_LABEL,
     PROVIDER_ELEMENT_ID_PROPERTY,
-    PROVIDER_ID_PROPERTY,
     PROVIDER_RESOURCE_LABEL,
 )
 
@@ -62,7 +61,6 @@ INSERT_FINDING_TEMPLATE = f"""
             finding.check_title = finding_data.check_title,
             finding.muted = finding_data.muted,
             finding.muted_reason = finding_data.muted_reason,
-            finding.provider_uid = $provider_uid,
             finding.firstseen = timestamp(),
             finding.lastupdated = $last_updated,
             finding._module_name = 'cartography:prowler',
@@ -74,7 +72,6 @@ INSERT_FINDING_TEMPLATE = f"""
 
     MERGE (resource)-[rel:HAS_FINDING]->(finding)
         ON CREATE SET
-            rel.provider_uid = $provider_uid,
             rel.firstseen = timestamp(),
             rel.lastupdated = $last_updated,
             rel._module_name = 'cartography:prowler',
@@ -84,7 +81,7 @@ INSERT_FINDING_TEMPLATE = f"""
 """
 
 CLEANUP_FINDINGS_TEMPLATE = f"""
-    MATCH (finding:{PROWLER_FINDING_LABEL} {{provider_uid: $provider_uid}})
+    MATCH (finding:{PROWLER_FINDING_LABEL})
         WHERE finding.lastupdated < $last_updated
 
     WITH finding LIMIT $batch_size
@@ -155,7 +152,6 @@ NODE_SYNC_TEMPLATE = f"""
     UNWIND $rows AS row
     MERGE (n:__NODE_LABELS__ {{{PROVIDER_ELEMENT_ID_PROPERTY}: row.provider_element_id}})
     SET n += row.props
-    SET n.{PROVIDER_ID_PROPERTY} = $provider_id
 """
 
 RELATIONSHIP_SYNC_TEMPLATE = f"""
@@ -164,5 +160,4 @@ RELATIONSHIP_SYNC_TEMPLATE = f"""
     MATCH (t:{PROVIDER_RESOURCE_LABEL} {{{PROVIDER_ELEMENT_ID_PROPERTY}: row.end_element_id}})
     MERGE (s)-[r:__REL_TYPE__ {{{PROVIDER_ELEMENT_ID_PROPERTY}: row.provider_element_id}}]->(t)
     SET r += row.props
-    SET r.{PROVIDER_ID_PROPERTY} = $provider_id
 """

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

@@ -38,12 +38,12 @@ Pipeline steps:
    Stale findings from previous scans are cleaned up.
 
 7. Sync the temp database into the tenant database:
-   - Drop the old provider subgraph (matched by _provider_id property).
+   - Drop the old provider subgraph (matched by dynamic _Provider_{uuid} label).
      graph_data_ready is set to False for all scans of this provider while
      the swap happens so the API doesn't serve partial data.
    - Copy nodes and relationships in batches. Every synced node gets a
-     _ProviderResource label and _provider_id / _provider_element_id
-     properties for multi-provider isolation.
+     _ProviderResource label and dynamic _Tenant_{uuid} / _Provider_{uuid}
+     isolation labels, plus a _provider_element_id property for MERGE keys.
    - Set graph_data_ready back to True.
 
 8. Drop the temporary database, mark the AttackPathsScan as COMPLETED.
@@ -62,7 +62,7 @@ from cartography.intel import analysis as cartography_analysis
 from cartography.intel import create_indexes as cartography_create_indexes
 from cartography.intel import ontology as cartography_ontology
 from celery.utils.log import get_task_logger
-from tasks.jobs.attack_paths import db_utils, findings, internet, sync, utils
+from tasks.jobs.attack_paths import db_utils, findings, indexes, internet, sync, utils
 from tasks.jobs.attack_paths.config import get_cartography_ingestion_function
 
 from api.attack_paths import database as graph_database
@@ -165,7 +165,7 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
         ) as tmp_neo4j_session:
             # Indexes creation
             cartography_create_indexes.run(tmp_neo4j_session, tmp_cartography_config)
-            findings.create_findings_indexes(tmp_neo4j_session)
+            indexes.create_findings_indexes(tmp_neo4j_session)
             db_utils.update_attack_paths_scan_progress(attack_paths_scan, 2)
 
             # The real scan, where iterates over cloud services
@@ -221,8 +221,8 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             cartography_create_indexes.run(
                 tenant_neo4j_session, tenant_cartography_config
             )
-            findings.create_findings_indexes(tenant_neo4j_session)
-            sync.create_sync_indexes(tenant_neo4j_session)
+            indexes.create_findings_indexes(tenant_neo4j_session)
+            indexes.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)

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

@@ -10,8 +10,6 @@ from typing import Any
 
 import neo4j
 from celery.utils.log import get_task_logger
-
-from api.attack_paths import database as graph_database
 from tasks.jobs.attack_paths.config import (
     PROVIDER_ISOLATION_PROPERTIES,
     PROVIDER_RESOURCE_LABEL,
@@ -19,7 +17,6 @@ from tasks.jobs.attack_paths.config import (
     get_provider_label,
     get_tenant_label,
 )
-from tasks.jobs.attack_paths.indexes import IndexType, create_indexes
 from tasks.jobs.attack_paths.queries import (
     NODE_FETCH_QUERY,
     NODE_SYNC_TEMPLATE,
@@ -28,14 +25,11 @@ from tasks.jobs.attack_paths.queries import (
     render_cypher_template,
 )
 
+from api.attack_paths import database as graph_database
+
 logger = get_task_logger(__name__)
 
 
-def create_sync_indexes(neo4j_session) -> None:
-    """Create indexes for provider resource sync operations."""
-    create_indexes(neo4j_session, IndexType.SYNC)
-
-
 def sync_graph(
     source_database: str,
     target_database: str,
@@ -81,8 +75,8 @@ def sync_nodes(
     """
     Sync nodes from source to target database.
 
-    Adds `_ProviderResource` label and `_provider_id` property to all nodes.
-    Also adds dynamic `_Tenant_{id}` and `_Provider_{id}` isolation labels.
+    Adds `_ProviderResource` label and dynamic `_Tenant_{id}` and `_Provider_{id}`
+    isolation labels to all nodes.
 
     Source and target sessions are opened sequentially per batch to avoid
     holding two Bolt connections simultaneously for the entire sync duration.
@@ -119,13 +113,7 @@ def sync_nodes(
                 query = render_cypher_template(
                     NODE_SYNC_TEMPLATE, {"__NODE_LABELS__": node_labels}
                 )
-                target_session.run(
-                    query,
-                    {
-                        "rows": batch,
-                        "provider_id": provider_id,
-                    },
-                )
+                target_session.run(query, {"rows": batch})
 
         total_synced += batch_count
         logger.info(
@@ -143,7 +131,7 @@ def sync_relationships(
     """
     Sync relationships from source to target database.
 
-    Adds `_provider_id` property to all relationships.
+    Matches source and target nodes by `_provider_element_id` in the tenant database.
 
     Source and target sessions are opened sequentially per batch to avoid
     holding two Bolt connections simultaneously for the entire sync duration.
@@ -174,13 +162,7 @@ def sync_relationships(
                 query = render_cypher_template(
                     RELATIONSHIP_SYNC_TEMPLATE, {"__REL_TYPE__": rel_type}
                 )
-                target_session.run(
-                    query,
-                    {
-                        "rows": batch,
-                        "provider_id": provider_id,
-                    },
-                )
+                target_session.run(query, {"rows": batch})
 
         total_synced += batch_count
         logger.info(

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

@@ -6,6 +6,7 @@ from unittest.mock import MagicMock, call, patch
 import pytest
 from django_celery_results.models import TaskResult
 from tasks.jobs.attack_paths import findings as findings_module
+from tasks.jobs.attack_paths import indexes as indexes_module
 from tasks.jobs.attack_paths import internet as internet_module
 from tasks.jobs.attack_paths import sync as sync_module
 from tasks.jobs.attack_paths.scan import run as attack_paths_run
@@ -39,10 +40,10 @@ class TestAttackPathsRun:
     @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.indexes.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.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run")
@@ -189,7 +190,7 @@ class TestAttackPathsRun:
     @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.indexes.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")
@@ -288,7 +289,7 @@ class TestAttackPathsRun:
     @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.indexes.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")
@@ -391,7 +392,7 @@ class TestAttackPathsRun:
     @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.indexes.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")
@@ -493,10 +494,10 @@ class TestAttackPathsRun:
         "tasks.jobs.attack_paths.scan.graph_database.drop_subgraph",
         side_effect=RuntimeError("drop failed"),
     )
-    @patch("tasks.jobs.attack_paths.scan.sync.create_sync_indexes")
+    @patch("tasks.jobs.attack_paths.scan.indexes.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.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run")
@@ -606,10 +607,10 @@ class TestAttackPathsRun:
         side_effect=RuntimeError("sync failed"),
     )
     @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.indexes.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.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run")
@@ -719,10 +720,10 @@ class TestAttackPathsRun:
     @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.indexes.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.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run")
@@ -837,10 +838,10 @@ class TestAttackPathsRun:
         "tasks.jobs.attack_paths.scan.graph_database.drop_subgraph",
         side_effect=RuntimeError("drop failed"),
     )
-    @patch("tasks.jobs.attack_paths.scan.sync.create_sync_indexes")
+    @patch("tasks.jobs.attack_paths.scan.indexes.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.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run")
@@ -1250,7 +1251,7 @@ class TestAttackPathsFindingsHelpers:
     def test_create_findings_indexes_executes_all_statements(self):
         mock_session = MagicMock()
         with patch("tasks.jobs.attack_paths.indexes.run_write_query") as mock_run_write:
-            findings_module.create_findings_indexes(mock_session)
+            indexes_module.create_findings_indexes(mock_session)
 
         from tasks.jobs.attack_paths.indexes import FINDINGS_INDEX_STATEMENTS
 
@@ -1312,7 +1313,6 @@ class TestAttackPathsFindingsHelpers:
 
         assert mock_session.run.call_count == 2
         params = mock_session.run.call_args.args[1]
-        assert params["provider_uid"] == str(provider.uid)
         assert params["last_updated"] == config.update_tag
 
     def test_stream_findings_with_resources_returns_latest_scan_data(