fix(c-009): add make cypher-lint and close live interpolation holes

.pre-commit-config.yaml

@@ -78,6 +78,12 @@ repos:
           fi; exit 0'
         language: system
         pass_filenames: false
+      - id: cypher-lint
+        name: make cypher-lint (C-009)
+        entry: python tools/cypher_lint.py
+        language: python
+        pass_filenames: false
+        types: [python]
 
   # L9 contract enforcement (24 invariants, 27 docs)
   - repo: local

Makefile

@@ -12,7 +12,7 @@
 # ─────────────────────────────────────────────────────────────
 
 .PHONY: dev dev-build dev-down dev-logs dev-restart health
-.PHONY: test test-unit test-integration seed shell neo4j-shell
+.PHONY: test test-unit test-integration seed shell neo4j-shell cypher-lint
 
 # ── Governance ─────────────────────────────────────────────
 
@@ -209,13 +209,16 @@ clean:	## Remove volumes + containers
 
 # ── Quality Gates (local, no Docker) ───────────────────────
 
-.PHONY: lint lint-fix typecheck check
+.PHONY: lint lint-fix typecheck check cypher-lint
 
 lint:	## Ruff lint + format check (no mutation) + MyPy — matches CI's blocking gate
 	ruff check .
 	ruff format --check .
 	mypy engine/
 
+cypher-lint:	## C-009: scan generated Cypher for injection vectors
+	python3 tools/cypher_lint.py
+
 lint-fix:	## Autofix: ruff check --fix + ruff format . (run this when `make lint` fails)
 	ruff check . --fix
 	ruff format .

engine/gates/types/all_gates.py

@@ -17,6 +17,7 @@
 from abc import ABC, abstractmethod
 
 from engine.config.schema import DomainSpec, GateSpec
+from engine.utils.security import cypher_quoted_ident
 
 logger = logging.getLogger(__name__)
 
@@ -192,8 +193,8 @@ def compile(self) -> str:
             # Build CASE WHEN for complex mapping
             cases = []
             for query_val, candidate_vals in self.spec.mapping.items():
-                val_list = ", ".join([f"'{v}'" for v in candidate_vals])
-                cases.append(f"WHEN {param} = '{query_val}' THEN {prop} IN [{val_list}]")
+                val_list = ", ".join(cypher_quoted_ident(str(v)) for v in candidate_vals)
+                cases.append(f"WHEN {param} = {cypher_quoted_ident(str(query_val))} THEN {prop} IN [{val_list}]")
 
             case_expr = " ".join(cases)
             return f"CASE {case_expr} ELSE false END"

engine/gds/scheduler.py

@@ -31,7 +31,7 @@
 
 from engine.config.schema import AggregationStrategy, DomainSpec, EdgeCategory, GDSJobSpec
 from engine.graph.driver import GraphDriver
-from engine.utils.security import sanitize_label
+from engine.utils.security import cypher_quoted_ident, sanitize_label
 
 # S2-10: EdgeCategory → recommended GDS configuration mapping.
 # Paper: different relation types benefit from different algorithms/aggregation.
@@ -305,14 +305,15 @@ async def _run_louvain(self, job_spec: GDSJobSpec) -> dict[str, Any]:
         graph_name = f"{safe_job_name}_graph"
 
         # Pre-cleanup: drop stale projection if it exists (fixes crash on re-run)
-        pre_drop = f"""
-        CALL gds.graph.exists('{graph_name}') YIELD exists
+        gds_params = {"graph_name": graph_name}
+        pre_drop = """
+        CALL gds.graph.exists($graph_name) YIELD exists
         WITH exists WHERE exists
-        CALL gds.graph.drop('{graph_name}') YIELD graphName
+        CALL gds.graph.drop($graph_name) YIELD graphName
         RETURN graphName
         """
         try:
-            await self.graph_driver.execute_query(pre_drop, database=db)
+            await self.graph_driver.execute_query(pre_drop, parameters=gds_params, database=db)
         except Exception as exc:
             exc_msg = str(exc).lower()
             if "not found" in exc_msg or "does not exist" in exc_msg:
@@ -327,27 +328,28 @@ async def _run_louvain(self, job_spec: GDSJobSpec) -> dict[str, Any]:
         # Sanitize write property name
         write_prop = sanitize_label(job_spec.writeproperty or "communityId")
 
+        gds_params["write_prop"] = write_prop
         project_cypher = f"""
-        CALL gds.graph.project('{graph_name}', {node_labels}, {edge_types})
+        CALL gds.graph.project($graph_name, {node_labels}, {edge_types})
         YIELD graphName, nodeCount, relationshipCount
         RETURN graphName, nodeCount, relationshipCount
         """
         try:
-            await self.graph_driver.execute_query(project_cypher, database=db)
+            await self.graph_driver.execute_query(project_cypher, parameters=gds_params, database=db)
 
-            louvain_cypher = f"""
-            CALL gds.louvain.write('{graph_name}', {{writeProperty: '{write_prop}'}})
+            louvain_cypher = """
+            CALL gds.louvain.write($graph_name, {writeProperty: $write_prop})
             YIELD communityCount, modularity
             RETURN communityCount, modularity
             """
-            result = await self.graph_driver.execute_query(louvain_cypher, database=db)
+            result = await self.graph_driver.execute_query(louvain_cypher, parameters=gds_params, database=db)
             data = result[0] if result else {}
             logger.info(f"Louvain: {data}")
             return {"communities": data.get("communityCount"), "modularity": data.get("modularity")}
         finally:
-            drop_cypher = f"CALL gds.graph.drop('{graph_name}') YIELD graphName RETURN graphName"
+            drop_cypher = "CALL gds.graph.drop($graph_name) YIELD graphName RETURN graphName"
             try:
-                await self.graph_driver.execute_query(drop_cypher, database=db)
+                await self.graph_driver.execute_query(drop_cypher, parameters=gds_params, database=db)
             except Exception:
                 logger.exception(f"Failed to drop projected graph '{graph_name}'")
 
@@ -588,7 +590,9 @@ async def _run_equipment_sync(self, job_spec: GDSJobSpec) -> dict[str, Any]:
         equipment_props = self._get_equipment_properties(job_spec)
 
         # Build dynamic CASE statements for equipment detection
-        case_statements = [f"CASE WHEN f.{prop} = true THEN '{name}' END" for prop, name in equipment_props]
+        case_statements = [
+            f"CASE WHEN f.{prop} = true THEN {cypher_quoted_ident(name)} END" for prop, name in equipment_props
+        ]
         case_list = ",\n            ".join(case_statements)
 
         cypher = f"""

engine/handlers.py

@@ -481,7 +481,7 @@ async def handle_match(tenant: str, payload: dict[str, Any]) -> dict[str, Any]:
     )
 
     try:
-        parameters = {**resolved_query, "top_n": top_n}
+        parameters = {**resolved_query, "top_n": top_n, **scoring_assembler.last_query_params}
         results = await graph_driver.execute_query(
             cypher=cypher,
             parameters=parameters,

engine/scoring/assembler.py

@@ -91,6 +91,7 @@ def __init__(
         self.domain_spec = domain_spec
         self.scoring_spec = domain_spec.scoring
         self._last_active_dims: list[str] = []
+        self._query_params: dict[str, Any] = {}
         self._graph_driver = graph_driver
         self._learned_weights: dict[str, float] | None = None
         self._population_means: dict[str, float] = {}  # S2-01: cached population means
@@ -152,6 +153,7 @@ def assemble_scoring_clause(
         """
         from engine.config.settings import settings
 
+        self._query_params = {}
         pareto_metadata: dict[str, Any] | None = None
 
         # Pareto pre-filter (lazy import to avoid circular deps)
@@ -214,6 +216,11 @@ def last_active_dimension_names(self) -> list[str]:
         """Return dimension names from the most recent assemble_scoring_clause call."""
         return list(self._last_active_dims)
 
+    @property
+    def last_query_params(self) -> dict[str, Any]:
+        """Copy of Cypher parameters collected during the last assemble call."""
+        return dict(self._query_params)
+
     def _compile_dimension(self, dim: ScoringDimensionSpec) -> str:
         """Dispatch to computation-specific compiler.
 
@@ -454,21 +461,31 @@ def _compile_preference_attention(self, dim: ScoringDimensionSpec) -> str:
         outcome_rel = sanitize_label(metadata.get("outcome_relation", "RESULTED_IN"))
         outcome_node = sanitize_label(metadata.get("outcome_node", "TransactionOutcome"))
         success_prop = sanitize_label(metadata.get("success_property", "outcome_type"))
-        success_value = sanitize_label(metadata.get("success_value", "closed_won"))
+        success_value = metadata.get("success_value", "closed_won")
+        if not isinstance(success_value, str):
+            msg = f"Dimension '{dim.name}': success_value must be a string"
+            raise ValueError(msg)
 
         cand_community_prop = sanitize_label(dim.candidateprop or "community_id")
+        safe_dim = sanitize_label(dim.name)
+        success_key = f"pref_success_{safe_dim}"
+        default_key = f"pref_default_{safe_dim}"
+        sample_key = f"pref_sample_k_{safe_dim}"
+        self._query_params[success_key] = success_value
+        self._query_params[default_key] = default
+        self._query_params[sample_key] = int(sample_k)
 
         return (
             f"CASE "
             f"  WHEN size([(qe)-[:{outcome_rel}]->(o:{outcome_node}) "
-            f"    WHERE o.{success_prop} = '{success_value}' | o]) = 0 THEN {default} "
+            f"    WHERE o.{success_prop} = ${success_key} | o]) = 0 THEN ${default_key} "
             f"  ELSE toFloat("
             f"    size([(qe)-[:{outcome_rel}]->(o:{outcome_node}) "
-            f"      WHERE o.{success_prop} = '{success_value}' "
-            f"      AND o.community_id = candidate.{cand_community_prop} | o][0..{sample_k}])"
+            f"      WHERE o.{success_prop} = ${success_key} "
+            f"      AND o.community_id = candidate.{cand_community_prop} | o][0..${sample_key}])"
             f"  ) / toFloat("
             f"    size([(qe)-[:{outcome_rel}]->(o:{outcome_node}) "
-            f"      WHERE o.{success_prop} = '{success_value}' | o][0..{sample_k}])"
+            f"      WHERE o.{success_prop} = ${success_key} | o][0..${sample_key}])"
             f"  ) "
             f"END"
         )

engine/utils/security.py

@@ -38,3 +38,12 @@ def sanitize_label(label: str) -> str:
         msg = f"Invalid label or type: {label!r}"
         raise ValueError(msg)
     return label
+
+
+def cypher_quoted_ident(value: str) -> str:
+    """Return a single-quoted Cypher string literal of a sanitized identifier.
+
+    Compile-time Cypher fragments that cannot take ``$params`` use this instead
+    of interpolating ``'{value}'`` so the source never contains that pattern.
+    """
+    return "'" + sanitize_label(value) + "'"

tests/unit/test_cypher_lint.py

@@ -0,0 +1,41 @@
+"""Unit tests — tools/cypher_lint.py C-009 scanner."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tools.cypher_lint import scan_tree
+
+
+@pytest.mark.unit
+def test_quoted_interpolation_fails(tmp_path: Path) -> None:
+    engine = tmp_path / "engine"
+    engine.mkdir()
+    target = engine / "sync"
+    target.mkdir()
+    (target / "generator.py").write_text(
+        "cypher = f\"SET n.status = '{status}'\"\n",
+        encoding="utf-8",
+    )
+
+    findings = scan_tree(tmp_path)
+
+    assert findings, "quoted value interpolation must fail the scanner"
+    assert findings[0].rel_path == "engine/sync/generator.py"
+    assert "status" in findings[0].pattern
+
+
+@pytest.mark.unit
+def test_sanitize_label_interpolation_passes(tmp_path: Path) -> None:
+    engine = tmp_path / "engine"
+    engine.mkdir()
+    (engine / "ok.py").write_text(
+        'from engine.utils.security import sanitize_label\ncypher = f"MATCH (n:{sanitize_label(label)}) RETURN n"\n',
+        encoding="utf-8",
+    )
+
+    findings = scan_tree(tmp_path)
+
+    assert findings == []

tests/unit/test_cypher_utils.py

@@ -45,3 +45,19 @@ def test_sanitize_label_rejects_too_long():
 
     with pytest.raises((ValueError, Exception)):
         sanitize_label("A" * 200)
+
+
+@pytest.mark.unit
+def test_cypher_quoted_ident_wraps_sanitized_label():
+    from engine.utils.security import cypher_quoted_ident
+
+    assert cypher_quoted_ident("Facility") == "'Facility'"
+    assert cypher_quoted_ident("closed_won") == "'closed_won'"
+
+
+@pytest.mark.unit
+def test_cypher_quoted_ident_rejects_injection():
+    from engine.utils.security import cypher_quoted_ident
+
+    with pytest.raises((ValueError, Exception)):
+        cypher_quoted_ident("'; DROP TABLE")