Fix Neo4j Cypher injection in BaseBot.write_to_graph

startupintel/bots/base.py

@@ -1,10 +1,18 @@
+import re
 from abc import ABC, abstractmethod
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime
 from uuid import UUID
 
 from startupintel.db.models import StartupScore
 
+# Neo4j has no parameterized syntax for property *names* (only values), so the
+# `{bot_name}_score` property name below is interpolated directly into the
+# Cypher string. Restrict it to a strict identifier allowlist first so a bot
+# name can never break out of the `s.<property>` position (e.g. via spaces,
+# backticks, or braces).
+_VALID_PROPERTY_NAME = re.compile(r"^[a-z][a-z0-9_]*$")
+
 
 @dataclass
 class BotResult:
@@ -87,6 +95,8 @@ async def write_to_graph(self, result: BotResult) -> None:
         if self.neo4j is None:
             return
         property_name = f"{self.name}_score"
+        if not _VALID_PROPERTY_NAME.fullmatch(property_name):
+            raise ValueError(f"Unsafe Neo4j property name derived from bot name: {self.name!r}")
         async with self.neo4j.session() as session:
             await session.run(
                 f"MATCH (s:Startup {{id: $startup_id}}) SET s.{property_name} = $score",

tests/test_bots/test_base.py

@@ -0,0 +1,112 @@
+from datetime import UTC, datetime
+from uuid import uuid4
+
+import pytest
+
+from startupintel.bots.base import BaseBot, BotResult
+
+
+class _StubBot(BaseBot):
+    """Minimal concrete BaseBot for exercising write_to_graph directly."""
+
+    def __init__(self, name: str, **kwargs):
+        super().__init__(**kwargs)
+        self.name = name
+
+    async def fetch_signals(self, startup_id):
+        return {}
+
+    async def compute_score(self, raw):
+        return {}
+
+    def get_weights(self):
+        return {}
+
+    def build_rag_query(self, raw):
+        return ""
+
+    def diagnosis_prompt_template(self):
+        return ""
+
+    async def maybe_emit_event(self, result):
+        return None
+
+
+class _FakeSession:
+    def __init__(self, calls: list):
+        self._calls = calls
+
+    async def run(self, query, **params):
+        self._calls.append((query, params))
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc_info):
+        return False
+
+
+class _FakeNeo4jDriver:
+    def __init__(self):
+        self.calls: list = []
+
+    def session(self):
+        return _FakeSession(self.calls)
+
+
+def _make_result(bot_name: str) -> BotResult:
+    return BotResult(
+        startup_id=uuid4(),
+        bot_name=bot_name,
+        score=42.0,
+        signal_breakdown={},
+        raw_signals={},
+        similar_cases=[],
+        llm_diagnosis="",
+        computed_at=datetime.now(UTC),
+    )
+
+
+@pytest.mark.asyncio
+async def test_write_to_graph_uses_expected_property_name():
+    driver = _FakeNeo4jDriver()
+    bot = _StubBot("runway", neo4j=driver)
+    result = _make_result("runway")
+
+    await bot.write_to_graph(result)
+
+    assert len(driver.calls) == 1
+    query, params = driver.calls[0]
+    assert "s.runway_score" in query
+    assert params["score"] == 42.0
+
+
+@pytest.mark.asyncio
+async def test_write_to_graph_skips_when_neo4j_not_configured():
+    bot = _StubBot("runway", neo4j=None)
+    result = _make_result("runway")
+
+    await bot.write_to_graph(result)  # should not raise
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "malicious_name",
+    [
+        "runway`}) DETACH DELETE s //",
+        "runway} SET s.pwned = true MATCH (x",
+        "runway; DROP",
+        "runway score",
+        "Runway",
+        "",
+    ],
+)
+async def test_write_to_graph_rejects_unsafe_bot_names(malicious_name):
+    driver = _FakeNeo4jDriver()
+    bot = _StubBot(malicious_name, neo4j=driver)
+    result = _make_result(malicious_name)
+
+    with pytest.raises(ValueError, match="Unsafe Neo4j property name"):
+        await bot.write_to_graph(result)
+
+    assert driver.calls == []