fix(csv_sql): prevent SQL injection via DuckDB parameter binding

tools/src/aden_tools/tools/csv_tool/csv_tool.py

@@ -330,9 +330,9 @@ def csv_sql(
             if not query or not query.strip():
                 return {"error": "query cannot be empty"}
 
-            # Security: only allow SELECT statements
-            query_upper = query.strip().upper()
-            if not query_upper.startswith("SELECT"):
+            # Security: allow SELECT/WITH only
+            query_upper = query.lstrip().upper()
+            if not (query_upper.startswith("SELECT") or query_upper.startswith("WITH")):
                 return {"error": "Only SELECT queries are allowed for security reasons"}
 
             # Disallowed keywords for security
@@ -351,18 +351,24 @@ def csv_sql(
                 if keyword in query_upper:
                     return {"error": f"'{keyword}' is not allowed in queries"}
 
-            # Execute query using in-memory DuckDB
+            # Block obvious multi-statement / injection attempts
+            q_lower = query.lower()
+            for token in [";", "--", "/*", "*/"]:
+                if token in q_lower:
+                    return {"error": "Multiple statements or comments are not allowed"}
+
             con = duckdb.connect(":memory:")
             try:
-                # Load CSV as 'data' table
-                con.execute(f"CREATE TABLE data AS SELECT * FROM read_csv_auto('{secure_path}')")
+                # SAFE: parameter binding (no string interpolation)
+                con.execute(
+                    "CREATE TABLE data AS SELECT * FROM read_csv_auto(?)",
+                    [str(secure_path)],
+                )
 
-                # Execute user query
                 result = con.execute(query)
                 columns = [desc[0] for desc in result.description]
                 rows = result.fetchall()
 
-                # Convert to list of dicts
                 rows_as_dicts = [dict(zip(columns, row, strict=False)) for row in rows]
 
                 return {
@@ -374,12 +380,12 @@ def csv_sql(
                     "rows": rows_as_dicts,
                     "row_count": len(rows_as_dicts),
                 }
+
             finally:
                 con.close()
 
         except Exception as e:
             error_msg = str(e)
-            # Make DuckDB errors more readable
             if "Catalog Error" in error_msg:
                 return {"error": f"SQL error: {error_msg}. Remember the table is named 'data'."}
             return {"error": f"Query failed: {error_msg}"}

tools/tests/tools/test_csv_tool.py

@@ -732,6 +732,53 @@ def test_basic_select(self, csv_tools, products_csv, tmp_path):
         assert "id" in result["columns"]
         assert "name" in result["columns"]
 
+    def test_path_with_single_quote(self, csv_tools, session_dir, tmp_path):
+        """Regression: CSV paths containing single quotes should work (parameter binding)."""
+        csv_file = session_dir / "O'Reilly.csv"
+        csv_file.write_text("name,age\nAlice,21\nBob,22\n", encoding="utf-8")
+
+        with patch("aden_tools.tools.file_system_toolkits.security.WORKSPACES_DIR", str(tmp_path)):
+            result = csv_tools["csv_sql"](
+                path="O'Reilly.csv",
+                workspace_id=TEST_WORKSPACE_ID,
+                agent_id=TEST_AGENT_ID,
+                session_id=TEST_SESSION_ID,
+                query="SELECT * FROM data",
+            )
+
+        assert "error" not in result, result
+        assert result["success"] is True
+        assert result["row_count"] == 2
+        names = [row["name"] for row in result["rows"]]
+        assert "Alice" in names
+        assert "Bob" in names
+
+    # --- NEW: security regression tests required by Issue #1256 ---
+
+    def test_reject_non_select(self, csv_tools, products_csv, tmp_path):
+        """Reject any non-SELECT / non-WITH query."""
+        with patch("aden_tools.tools.file_system_toolkits.security.WORKSPACES_DIR", str(tmp_path)):
+            result = csv_tools["csv_sql"](
+                path=products_csv.name,
+                workspace_id=TEST_WORKSPACE_ID,
+                agent_id=TEST_AGENT_ID,
+                session_id=TEST_SESSION_ID,
+                query="DROP TABLE data",
+            )
+        assert "error" in result
+
+    def test_reject_multi_statement(self, csv_tools, products_csv, tmp_path):
+        """Reject multi-statement queries with semicolons."""
+        with patch("aden_tools.tools.file_system_toolkits.security.WORKSPACES_DIR", str(tmp_path)):
+            result = csv_tools["csv_sql"](
+                path=products_csv.name,
+                workspace_id=TEST_WORKSPACE_ID,
+                agent_id=TEST_AGENT_ID,
+                session_id=TEST_SESSION_ID,
+                query="SELECT * FROM data; DROP TABLE data",
+            )
+        assert "error" in result
+
     def test_where_clause(self, csv_tools, products_csv, tmp_path):
         """Filter with WHERE clause."""
         with patch("aden_tools.tools.file_system_toolkits.security.WORKSPACES_DIR", str(tmp_path)):