T09 · Insecure Skill Coding Practices
Warning
- Location
scripts/csv_loader.py:310- Finding
Unescaped Identifiers and Values Permit Cypher and RDF/Turtle Injection
- Content
View full analysis
List[str]: """Generate Neo4j Cypher statements.""" statements = [] for entity_type, entity_def in self.entities.items(): # Find sample data sample_nodes = [n for n in self.nodes.values() if n.node_type == entity_type] if sample_nodes: sample = sample_nodes[0] stmt = f"MERGE (n:{entity_type} {{id: row.{entity_def.id_column}}})" for prop, value in sample.properties.items(): stmt += f"\nSET n.{prop} = row.{prop}" statements.append(stmt) for rel in self.relationships: stmt = f"MERGE (s:{rel.source_entity})-[:{rel.relationship_type}]->(t:{rel.target_entity})" statements.append(stmt) return statements def generate_triples(self, namespace: str = "http://example.org/") -> List[str]: """Generate RDF triple statements.""" triples = [] for node in self.nodes.values(): # Type triple node_uri = f"<{namespace}{node.node_id}>" type_uri = f"<{namespace}{node.node_type}>" triples.append(f"{node_uri} rdf:type {type_uri} .") # Property triples for prop, value in node.properties.items(): prop_uri = f"<{namespace}{prop}>" if isinstance(value, str): triples.append(f'{node_uri} {prop_uri} "{value}" .') else: triples.append(f"{node_uri} {prop_uri} {value} .") for edge in self.edges: source_uri = f"<{namespace}{edge.source_id}>" target_uri = f"<{namespace}{edge.target_id}>" rel_uri = f"<{namespace}{edge.edge_type}>" triples.append(f"{source_uri} {rel_uri} {target_uri} .") return triples ``` ### Technical Analysis The generator directly ...[truncated 2523 chars]- Remediation
View remediation