fix(security)!: make raw Cypher endpoints read-only with write-clause denylist

/query/graph scoping was a documented no-op (graph_service returned the
query unscoped) and neo4j_client permitted writes; a live probe showed a
nonexistent user could read the whole graph.

- Add Neo4jClient.execute_read() that opens the session with
  default_access_mode=READ_ACCESS so the database refuses writes even if
  validation is bypassed.
- GraphService.execute_query() now rejects queries containing
  CREATE/MERGE/DELETE/DETACH/SET/REMOVE/DROP/FOREACH/LOAD or any CALL
  (conservative word-boundary denylist on the uppercased query) and
  executes through the read-only session; the no-op _scope_query_to_user
  is removed.
- Remove the false user-scoping claims from /query/graph (main.py) and
  /graph/query docs and the CypherQueryRequest model: the endpoints are
  documented as admin/debug, unscoped read-only (per-tenant label
  injection for arbitrary Cypher would need a real parser; /graph/nodes
  remains the tenant-scoped path).
- Offline unit tests: denylist coverage (incl. lowercase/multiline/CALL),
  word-boundary false-positive check, and READ_ACCESS session assertion.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-14 11:08:40 +02:00
co-authored by Claude Fable 5
parent 84e9185371
commit a4c299bf9b
6 changed files with 216 additions and 35 deletions
+33 -1
View File
@@ -8,7 +8,7 @@ Provides async Neo4j operations with:
- Automatic retry on transient failures
"""
from neo4j import AsyncGraphDatabase, AsyncDriver, AsyncSession
from neo4j import AsyncGraphDatabase, AsyncDriver, AsyncSession, READ_ACCESS
from typing import Optional, List, Dict, Any
import logging
@@ -97,6 +97,38 @@ class Neo4jClient:
records = await result.data()
return records
async def execute_read(
self,
cypher: str,
parameters: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""
Execute Cypher in a READ-ONLY session.
The session is opened with default_access_mode=READ_ACCESS, so the
database rejects any write attempt (CREATE/MERGE/DELETE/SET/...)
even if it slips past caller-side validation. Use this for any
query built from untrusted input (e.g. the /query/graph endpoint).
Args:
cypher: Cypher query string
parameters: Query parameters
Returns:
List of result records as dictionaries
Raises:
Exception: If driver not initialized, query fails, or the
query attempts a write (rejected by the read session)
"""
if not self._driver:
await self.connect()
async with self._driver.session(default_access_mode=READ_ACCESS) as session:
result = await session.run(cypher, parameters or {})
records = await result.data()
return records
async def execute_write(
self,
cypher: str,