cypher_validator

langroid/agent/special/neo4j/cypher_validator.py

Security validation for LLM-generated Cypher queries (GHSA-2pq5-3q89-j7cc).

Neo4jChatAgent executes Cypher produced by an LLM, whose text is influenceable by prompt injection -- directly, or indirectly via content the agent reads back through RAG. Executing such Cypher without restriction lets an attacker who can influence the prompt read or destroy all graph data and, when APOC or dbms.security procedures are enabled on the database role, reach the filesystem, network, or OS (config-conditional RCE). This mirrors the prompt-to-SQL-to-RCE issue fixed for SQLChatAgent in CVE-2026-25879.

Unless the operator opts in with allow_dangerous_operations=True:

• the retrieval (read) path is restricted to read-only Cypher -- any write or schema-mutating clause is rejected;
• both paths reject the code-execution / file / network primitives (LOAD CSV, apoc.*, dbms.*, CALL db.*).

Cypher has no lightweight AST parser in our dependency set, so the checks are keyword/pattern based, applied after stripping comments and string/identifier literals to reduce false positives. A blocklist is inherently bypassable; the real guarantee is the default-off gate plus running the agent against a least-privilege Neo4j role. Treat this as defense in depth, not a parser-grade allowlist.

validate_cypher_query(query, *, is_write, allow_dangerous)

Check an LLM-generated Cypher query against the agent's safety policy.

Parameters:

Name	Type	Description	Default
query	str	The raw Cypher string from the LLM.	required
is_write	bool	True for the creation/write path (ordinary write clauses are permitted), False for the retrieval/read path (write clauses are rejected so the path is read-only).	required
allow_dangerous	bool	When True, skip all checks (operator opt-in).	required

Returns:

Type	Description
Optional[str]	None if the query may be executed, otherwise a human-readable rejection
Optional[str]	message to relay back to the LLM.

Source code in langroid/agent/special/neo4j/cypher_validator.py

def validate_cypher_query(
    query: str, *, is_write: bool, allow_dangerous: bool
) -> Optional[str]:
    """Check an LLM-generated Cypher query against the agent's safety policy.

    Args:
        query: The raw Cypher string from the LLM.
        is_write: True for the creation/write path (ordinary write clauses are
            permitted), False for the retrieval/read path (write clauses are
            rejected so the path is read-only).
        allow_dangerous: When True, skip all checks (operator opt-in).

    Returns:
        None if the query may be executed, otherwise a human-readable rejection
        message to relay back to the LLM.
    """
    if allow_dangerous:
        return None

    scrubbed = _strip_literals_and_comments(query)

    for pat, label in _DANGEROUS_PATTERNS:
        if pat.search(scrubbed):
            logger.warning("Neo4jChatAgent rejected Cypher using %s: %r", label, query)
            return (
                f"Cypher query REJECTED for safety: it uses {label}, which can "
                f"execute code or access the filesystem/network. Rewrite the "
                f"query without it, or {_CONFIG_HINT}."
            )

    if not is_write:
        for pat, label in _WRITE_PATTERNS:
            if pat.search(scrubbed):
                logger.warning(
                    "Neo4jChatAgent rejected write clause %s on read path: %r",
                    label,
                    query,
                )
                return (
                    f"Cypher query REJECTED for safety: the retrieval tool runs "
                    f"READ-ONLY Cypher, but this query uses `{label}`. Use the "
                    f"creation tool for writes, rewrite this as a read-only "
                    f"(MATCH/RETURN) query, or {_CONFIG_HINT}."
                )

    return None