fix(api): add security hardening for Attack Paths custom query endpoint (#10238)

Author: josema-xyz

api/CHANGELOG.md

@@ -15,6 +15,7 @@ All notable changes to the **Prowler API** are documented in this file.
 
 ### 🐞 Fixed
 
+- Attack Paths: Security hardening for custom query endpoint (Cypher blocklist, input validation, rate limiting, Helm lockdown) [(#10238)](https://github.com/prowler-cloud/prowler/pull/10238)
 - Attack Paths: Add missing logging for query execution and exception details in scan error handling [(#10269)](https://github.com/prowler-cloud/prowler/pull/10269)
 - Attack Paths: Upgrade Cartography from 0.129.0 to 0.132.0, fixing `exposed_internet` not set on ELB/ELBv2 nodes [(#10272)](https://github.com/prowler-cloud/prowler/pull/10272)
 

api/src/backend/api/attack_paths/database.py

@@ -35,6 +35,7 @@
     "Neo.ClientError.Statement.AccessMode",
     "Neo.ClientError.Procedure.ProcedureNotFound",
 ]
+CLIENT_STATEMENT_EXCEPTION_PREFIX = "Neo.ClientError.Statement."
 
 # Module-level process-wide driver singleton
 _driver: neo4j.Driver | None = None
@@ -108,13 +109,18 @@ def get_session(
     except neo4j.exceptions.Neo4jError as exc:
         if (
             default_access_mode == neo4j.READ_ACCESS
+            and exc.code
             and exc.code in READ_EXCEPTION_CODES
         ):
             message = "Read query not allowed"
             code = READ_EXCEPTION_CODES[0]
             raise WriteQueryNotAllowedException(message=message, code=code)
 
         message = exc.message if exc.message is not None else str(exc)
+
+        if exc.code and exc.code.startswith(CLIENT_STATEMENT_EXCEPTION_PREFIX):
+            raise ClientStatementException(message=message, code=exc.code)
+
         raise GraphDatabaseQueryException(message=message, code=exc.code)
 
     finally:
@@ -227,3 +233,7 @@ def __str__(self) -> str:
 
 class WriteQueryNotAllowedException(GraphDatabaseQueryException):
     pass
+
+
+class ClientStatementException(GraphDatabaseQueryException):
+    pass

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

@@ -1,4 +1,5 @@
 import logging
+import re
 
 from typing import Any, Iterable
 
@@ -117,6 +118,38 @@ def execute_query(
 
 # Custom query helpers
 
+# Patterns that indicate SSRF or dangerous procedure calls
+# Defense-in-depth layer - the primary control is `neo4j.READ_ACCESS`
+_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),
+]
+
+# Strip string literals so patterns inside quotes don't cause false positives
+# Handles escaped quotes (\' and \") inside strings
+_STRING_LITERALS = re.compile(r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"")
+
+
+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 are stripped before matching to avoid false positives.
+    """
+    stripped = _STRING_LITERALS.sub("", cypher)
+    for pattern in _BLOCKED_PATTERNS:
+        if pattern.search(stripped):
+            raise ValidationError({"query": "Query contains a blocked operation"})
+
 
 def normalize_custom_query_payload(raw_data):
     if not isinstance(raw_data, dict):
@@ -135,6 +168,8 @@ def execute_custom_query(
     cypher: str,
     provider_id: str,
 ) -> dict[str, Any]:
+    validate_custom_query(cypher)
+
     try:
         graph = graph_database.execute_read_query(
             database=database_name,
@@ -143,6 +178,9 @@ def execute_custom_query(
         serialized = _serialize_graph(graph, provider_id)
         return _truncate_graph(serialized)
 
+    except graph_database.ClientStatementException as exc:
+        raise ValidationError({"query": exc.message})
+
     except graph_database.WriteQueryNotAllowedException:
         raise PermissionDenied(
             "Attack Paths query execution failed: read-only queries are enforced"
@@ -227,6 +265,12 @@ 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 matching provider_id={provider_id}"
+        )
+
     relationships = []
     for relationship in graph.relationships:
         if relationship._properties.get("provider_id") != provider_id:

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

@@ -501,6 +501,72 @@ def test_execute_custom_query_wraps_graph_errors():
     mock_logger.error.assert_called_once()
 
 
+# -- validate_custom_query ------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "cypher",
+    [
+        "LOAD CSV FROM 'http://169.254.169.254/' AS x RETURN x",
+        "load csv from 'http://evil.com' as row return row",
+        "CALL apoc.load.json('http://evil.com/') YIELD value RETURN value",
+        "CALL apoc.load.csvParams('http://evil.com/', {}, null) YIELD list RETURN list",
+        "CALL apoc.import.csv([{fileName: 'f'}], [], {}) YIELD node RETURN node",
+        "CALL apoc.export.csv.all('file.csv', {})",
+        "CALL apoc.cypher.run('CREATE (n)', {}) YIELD value RETURN value",
+        "CALL apoc.systemdb.graph() YIELD nodes RETURN nodes",
+        "CALL apoc.config.list() YIELD key, value RETURN key, value",
+        "CALL apoc.periodic.iterate('MATCH (n) RETURN n', 'DELETE n', {batchSize: 100})",
+        "CALL apoc.do.when(true, 'CREATE (n) RETURN n', '', {}) YIELD value RETURN value",
+        "CALL apoc.trigger.add('t', 'RETURN 1', {phase: 'before'})",
+        "CALL apoc.custom.asProcedure('myProc', 'RETURN 1')",
+    ],
+    ids=[
+        "LOAD_CSV",
+        "LOAD_CSV_lowercase",
+        "apoc.load.json",
+        "apoc.load.csvParams",
+        "apoc.import.csv",
+        "apoc.export.csv",
+        "apoc.cypher.run",
+        "apoc.systemdb.graph",
+        "apoc.config.list",
+        "apoc.periodic.iterate",
+        "apoc.do.when",
+        "apoc.trigger.add",
+        "apoc.custom.asProcedure",
+    ],
+)
+def test_validate_custom_query_rejects_blocked_patterns(cypher):
+    with pytest.raises(ValidationError) as exc:
+        views_helpers.validate_custom_query(cypher)
+
+    assert "blocked operation" in str(exc.value.detail)
+
+
+@pytest.mark.parametrize(
+    "cypher",
+    [
+        "MATCH (n:AWSAccount) RETURN n LIMIT 10",
+        "MATCH (a)-[r]->(b) RETURN a, r, b",
+        "MATCH (n) WHERE n.name CONTAINS 'load' RETURN n",
+        "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',
+    ],
+    ids=[
+        "simple_match",
+        "traversal",
+        "contains_load_substring",
+        "apoc_virtual_node",
+        "apoc_load_inside_single_quotes",
+        "load_csv_inside_double_quotes",
+    ],
+)
+def test_validate_custom_query_allows_clean_queries(cypher):
+    views_helpers.validate_custom_query(cypher)
+
+
 # -- _truncate_graph ----------------------------------------------------------