mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
Mohd KaifandSameer6305
b846ff88d4
* security: sanitize Cypher labels/relationship types/property keys (GHSA-482h-hw99-h62p) Node labels and property keys passed to create_node/create_relationship were interpolated directly into Cypher strings in the Neptune, Neo4j, and FalkorDB graph stores. Property values are parameterized, but labels and keys can't be bound as parameters, and nothing validated them, so a document-derived entity type or property name could close the current Cypher token early and append arbitrary statements (e.g. DETACH DELETE), running with the application's database credentials. - New shared semantica/graph_store/query_sanitize.py: sanitize_identifier() generalizes age_store.py's existing _sanitize_label/_sanitize_rel_type (the only backend that already validated this) into a helper the other backends can import without an import cycle with graph_store.py/methods.py. - Applied at every label/relationship-type/property-key interpolation site in amazon_neptune.py, neo4j_store.py, falkordb_store.py, graph_store.py (degree_centrality's own query builder), and methods.py (update_relationship's own query builder) — create_node, create_nodes, create_relationship, get_nodes, get_relationships, get_neighbors, shortest_path, update_node, create_index, and all relationship-type filters. - depth/max_depth path-length parameters are also cast to int before interpolation as defense-in-depth (they're already typed int, but Python doesn't enforce that at runtime). Added tests/graph_store/test_cypher_injection.py (12 tests covering the sanitizer directly and reproducing the advisory's injection payload against Neptune/Neo4j/FalkorDB create_node/create_relationship — asserts the malicious query is never built or sent), plus regression tests for graph_store.py's degree_centrality and methods.py's update_relationship. Full graph_store test suite (224 tests) passes with no regressions. * fix(graph-store): prevent depth-based Cypher injection * test(graph-store): tighten injection regression assertions * docs(changelog): add PR #910 (GHSA-482h Cypher injection) entry --------- Co-authored-by: Sameer6305 <sskadam6305@gmail.com>
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
from semantica.graph_store import methods
|
|
from semantica.graph_store.registry import method_registry
|
|
|
|
class TestGraphStoreMethods(unittest.TestCase):
|
|
def setUp(self):
|
|
# Reset global store
|
|
methods._reset_store()
|
|
|
|
# Mock GraphStore
|
|
self.mock_store = MagicMock()
|
|
self.mock_store_patcher = patch('semantica.graph_store.methods.GraphStore', return_value=self.mock_store)
|
|
self.MockGraphStore = self.mock_store_patcher.start()
|
|
|
|
def tearDown(self):
|
|
self.mock_store_patcher.stop()
|
|
methods._reset_store()
|
|
|
|
# Unregister custom methods if any
|
|
method_registry.unregister("node", "custom_create")
|
|
|
|
def test_create_node_default(self):
|
|
# Setup
|
|
labels = ["Person"]
|
|
props = {"name": "Alice"}
|
|
self.mock_store.create_node.return_value = {"id": 1, "labels": labels, "properties": props}
|
|
|
|
# Execute
|
|
result = methods.create_node(labels, props)
|
|
|
|
# Verify
|
|
self.mock_store.create_node.assert_called_once_with(labels, props)
|
|
self.assertEqual(result["id"], 1)
|
|
|
|
def test_create_node_custom(self):
|
|
# Register custom method
|
|
mock_custom = MagicMock(return_value={"id": 99, "custom": True})
|
|
method_registry.register("node", "custom_create", mock_custom)
|
|
|
|
# Execute
|
|
result = methods.create_node(["Person"], {"name": "Bob"}, method="custom_create")
|
|
|
|
# Verify
|
|
mock_custom.assert_called_once()
|
|
self.mock_store.create_node.assert_not_called()
|
|
self.assertEqual(result["id"], 99)
|
|
|
|
def test_execute_query_default(self):
|
|
# Setup
|
|
query = "MATCH (n) RETURN n"
|
|
self.mock_store.execute_query.return_value = {"records": []}
|
|
|
|
# Execute
|
|
result = methods.execute_query(query)
|
|
|
|
# Verify
|
|
self.mock_store.execute_query.assert_called_once_with(query, None)
|
|
|
|
def test_update_relationship_rejects_malicious_property_key(self):
|
|
"""Regression test for GHSA-482h-hw99-h62p: update_relationship()
|
|
interpolates property keys directly into a Cypher SET clause
|
|
(methods.py's own query builder, not delegated to the backend
|
|
store), so an unvalidated key was a direct injection point."""
|
|
from semantica.utils.exceptions import ValidationError
|
|
|
|
evil_key = "x} MATCH (victim) DETACH DELETE victim //"
|
|
with self.assertRaises(ValidationError):
|
|
methods.update_relationship(1, {evil_key: "value"})
|
|
self.mock_store.execute_query.assert_not_called()
|
|
|
|
def test_update_relationship_with_legitimate_keys_still_works(self):
|
|
self.mock_store.execute_query.return_value = {
|
|
"records": [{"id": 1, "type": "KNOWS"}]
|
|
}
|
|
result = methods.update_relationship(1, {"weight": 0.5})
|
|
query = self.mock_store.execute_query.call_args[0][0]
|
|
self.assertIn("r.weight = $weight", query)
|
|
self.assertNotIn("DETACH DELETE", query)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |