fix: multi-word FULLTEXT escaping, stable score alias, slop float validation

- Escape individual terms in multi-word FULLTEXT branch while preserving
  ~ optional-term prefix (prevents @field injection and accidental negation)
- Move score alias collision check out of per-row loop into
  _resolve_score_alias() so all rows use the same column name
- Reject float slop values (e.g. 2.9) instead of silently truncating
- Add tests for all three fixes (350 total)

Author: nkanu17

sql_redis/executor.py

@@ -104,7 +104,7 @@ class QueryResult:
 
 
 class _ScoreParseMixin:
-    """Shared helper for detecting RETURN 0 (NOCONTENT) in translated args."""
+    """Shared helpers for score-related response parsing."""
 
     @staticmethod
     def _has_return_0(args: list[str]) -> bool:
@@ -115,6 +115,23 @@ def _has_return_0(args: list[str]) -> bool:
         except (ValueError, IndexError):
             return False
 
+    @staticmethod
+    def _resolve_score_alias(score_alias: str | None, args: list[str]) -> str:
+        """Determine a stable score column name that won't collide with
+        document fields.  The alias is decided once before iterating rows
+        so every row uses the same column name."""
+        alias = score_alias or "__score"
+        # Extract RETURN field names from args to detect collision
+        try:
+            idx = args.index("RETURN")
+            count = int(args[idx + 1])
+            return_fields = set(args[idx + 2 : idx + 2 + count])
+        except (ValueError, IndexError):
+            return_fields = set()
+        if alias in return_fields:
+            alias = f"__score_{alias}"
+        return alias
+
 
 class Executor(_ScoreParseMixin):
     """Executes SQL queries against Redis."""
@@ -181,17 +198,21 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
         if translated.command == "FT.SEARCH":
             # Check if WITHSCORES was requested — changes response format
             with_scores = "WITHSCORES" in translated.args
-            score_alias = translated.score_alias
             # RETURN 0 suppresses document fields (like NOCONTENT);
             # with WITHSCORES the reply is [count, id, score, id, score, ...]
             no_content = self._has_return_0(translated.args)
+            # Determine score column name once so every row is consistent
+            alias = (
+                self._resolve_score_alias(translated.score_alias, translated.args)
+                if with_scores
+                else ""
+            )
 
             if with_scores and no_content:
                 # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...]
                 # Stride of 2: key, score (no field array)
                 for i in range(1, len(raw_result) - 1, 2):
                     score = raw_result[i + 1]
-                    alias = score_alias or "__score"
                     row = {alias: score}
                     rows.append(row)
             elif with_scores:
@@ -201,10 +222,6 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
                     score = raw_result[i + 1]
                     row_data = raw_result[i + 2]
                     row = dict(zip(row_data[::2], row_data[1::2]))
-                    alias = score_alias or "__score"
-                    # Guard: if alias collides with a document field, prefix it
-                    if alias in row:
-                        alias = f"__score_{alias}"
                     row[alias] = score
                     rows.append(row)
             else:
@@ -294,14 +311,17 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
 
         if translated.command == "FT.SEARCH":
             with_scores = "WITHSCORES" in translated.args
-            score_alias = translated.score_alias
             no_content = self._has_return_0(translated.args)
+            alias = (
+                self._resolve_score_alias(translated.score_alias, translated.args)
+                if with_scores
+                else ""
+            )
 
             if with_scores and no_content:
                 # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...]
                 for i in range(1, len(raw_result) - 1, 2):
                     score = raw_result[i + 1]
-                    alias = score_alias or "__score"
                     row = {alias: score}
                     rows.append(row)
             elif with_scores:
@@ -310,9 +330,6 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
                     score = raw_result[i + 1]
                     row_data = raw_result[i + 2]
                     row = dict(zip(row_data[::2], row_data[1::2]))
-                    alias = score_alias or "__score"
-                    if alias in row:
-                        alias = f"__score_{alias}"
                     row[alias] = score
                     rows.append(row)
             else:

sql_redis/parser.py

@@ -980,6 +980,12 @@ def _add_function_condition(
             if len(args) >= 3:
                 slop_val = self._extract_literal_value(args[2])
                 if slop_val is not None:
+                    # Reject non-integer values (e.g. 2.9) instead of
+                    # silently truncating via int().
+                    if isinstance(slop_val, float) and slop_val != int(slop_val):
+                        raise ValueError(
+                            f"FULLTEXT slop argument must be an integer (got {slop_val})"
+                        )
                     slop = int(slop_val)
                     if slop < 0:
                         raise ValueError(

sql_redis/query_builder.py

@@ -150,8 +150,9 @@ def build_text_condition(
             search_value = f'"{escaped}"'
         elif " " in value and " OR " not in value:
             # FULLTEXT/MATCH with multi-word: tokenized search with stopword filtering.
-            # FULLTEXT is intentionally pass-through — users craft RediSearch queries
-            # via fulltext(), so operator chars like ~, |, - are preserved.
+            # Each term is escaped to prevent accidental operator injection, but a
+            # leading ~ (optional-term modifier) is preserved as an intentional
+            # RediSearch operator.
             words = value.split()
             removed_stopwords = [
                 w for w in words if w.lower() in REDIS_DEFAULT_STOPWORDS
@@ -170,7 +171,15 @@ def build_text_condition(
                     stacklevel=2,
                 )
 
-            terms = " ".join(filtered_words) if filtered_words else value
+            escaped_words = []
+            for w in (filtered_words if filtered_words else words):
+                if w.startswith("~"):
+                    # Preserve ~ optional-term prefix, escape the rest
+                    escaped_words.append("~" + self._escape_fulltext_term(w[1:]))
+                else:
+                    escaped_words.append(self._escape_fulltext_term(w))
+
+            terms = " ".join(escaped_words)
             search_value = f"({terms})"
         elif " OR " in value:
             # OR union within text field: split on ' OR ', escape each term, join with |

tests/test_query_builder.py

@@ -582,3 +582,21 @@ def test_or_terms_escape_at_sign(self):
         builder = QueryBuilder()
         result = builder.build_text_condition("title", "FULLTEXT", "hello OR @world")
         assert result == "@title:(hello|\\@world)"
+
+    def test_multiword_fulltext_escapes_special_chars(self):
+        """Multi-word FULLTEXT escapes dangerous chars like @ and |."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("title", "FULLTEXT", "hello @world")
+        assert result == "@title:(hello \\@world)"
+
+    def test_multiword_fulltext_preserves_optional_prefix(self):
+        """Multi-word FULLTEXT preserves ~ optional-term prefix."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("title", "FULLTEXT", "laptop ~gaming")
+        assert result == "@title:(laptop ~gaming)"
+
+    def test_multiword_fulltext_escapes_dash(self):
+        """Multi-word FULLTEXT escapes - to prevent accidental negation."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("title", "FULLTEXT", "hello anti-virus")
+        assert result == "@title:(hello anti\\-virus)"

tests/test_sql_parser.py

@@ -865,3 +865,9 @@ def test_fulltext_negative_slop_raises(self):
         parser = SQLParser()
         with pytest.raises(ValueError, match="non-negative integer"):
             parser.parse("SELECT * FROM idx WHERE fulltext(title, 'hello world', -1)")
+
+    def test_fulltext_float_slop_raises(self):
+        """Float slop in fulltext() raises ValueError instead of silently truncating."""
+        parser = SQLParser()
+        with pytest.raises(ValueError, match="must be an integer"):
+            parser.parse("SELECT * FROM idx WHERE fulltext(title, 'hello world', 2.9)")