T09 · Insecure Skill Coding Practices
- Location
scripts/neo4j_client.py:19- Finding
Cypher Injection Through Unvalidated Neo4j Identifiers
- Content
View full analysis
Vulnerability Details
File Location:
scripts/neo4j_client.py:19-58
Vulnerability Type: Cypher injection through direct interpolation of labels, property keys, and relationship types
Risk Level: HighVulnerable Code
python def create_node(self, label: str, properties: Dict[str, Any]) -> str: cypher = f"CREATE (n:{label} $props) RETURN id(n) as node_id" with self.driver.session() as session: result = session.run(cypher, props=properties) record = result.single() return record["node_id"] if record else None def create_relationship(self, from_name: str, from_label: str, rel_type: str, to_name: str, to_label: str, properties: Optional[Dict] = None, match_key: str = "name") -> bool: props = properties or {} cypher = f""" MATCH (a:{from_label} {{{match_key}: $from_name}}) MATCH (b:{to_label} {{{match_key}: $to_name}}) CREATE (a)-[r:{rel_type} $props]->(b) RETURN r """ with self.driver.session() as session: result = session.run(cypher, from_name=from_name, to_name=to_name, props=props) return result.single() is not None def get_node(self, label: str, key: str, value: Any) -> Optional[Dict]: cypher = f"MATCH (n:{label} {{{key}: $value}}) RETURN n" results = self.query(cypher, {"value": value}) return results[0] if results else None def update_node(self, label: str, key: str, value: Any, new_props: Dict[str, Any]) -> bool: cypher = f"MATCH (n:{label} {{{key}: $value}}) SET n += $new_props RETURN n" results = self.query(cypher, {"value": value, "new_props": new_props}) return len(results) > 0Technical Analysis
Property values are correctly supplied as Neo4j parameters, but structural query elements—including node labels, relationship types, and property keys—are inserted directly into Cypher strings.
...[truncated 1761 chars]
- Remediation
View remediation
Remediation Suggestions
- Validate every label, relationship type, property key, and variable identifier with a restrictive rule such as
^[A-Za-z_][A-Za-z0-9_]*$. - Prefer application-controlled allowlists for expected labels, relationship types, and keys.
- Reject identifiers containing whitespace, quotes, backticks, delimiters, comments, or other Cypher syntax.
- Where dynamic identifiers are unavoidable, use the Neo4j driver's supported safe identifier-quoting strategy after validation.
- Keep all property values parameterized.
- Clearly document
query()as a trusted-input-only interface and do not pass HTTP parameters, model output, or other untrusted strings directly to it. - Use a least-privileged Neo4j account that cannot perform schema administration or unrestricted deletion.
- Add adversarial unit tests for every structural argument.
- Validate every label, relationship type, property key, and variable identifier with a restrictive rule such as
