Validate numeric segment values + add flagsmith-lint-tests hook

The numeric comparator family (GREATER_THAN / LESS_THAN / *_INCLUSIVE)
was interpolating segment-supplied values raw into the WHERE clause —
a SQL-injection path for any user with `MANAGE_SEGMENTS` permission on
the project. Same risk class as the MODULO branch already guarded
against. Fix is symmetrical: float() the value in Python first, return
None on ValueError so the caller falls back.

Three new unit tests assert the guard:
  - injection-shaped value on each comparator returns None
  - clean numeric value interpolates the parsed float
  - injection in MODULO's divisor returns None

Also adds the flagsmith-lint-tests pre-commit hook (FT003 / FT004 from
flagsmith-common). Renames the parity test and reshapes every unit
test to the `test_subject__condition__expected` template plus
Given/When/Then comments enforced by the hook.

beep boop

Author: khvn26

.pre-commit-config.yaml

@@ -15,6 +15,10 @@ repos:
       - id: check-yaml
       - id: check-json
       - id: check-toml
+  - repo: https://github.com/Flagsmith/flagsmith-common
+    rev: v3.8.2
+    hooks:
+      - id: flagsmith-lint-tests
   - repo: local
     hooks:
       - id: python-typecheck

src/flagsmith_sql_flag_engine/translator.py

@@ -28,9 +28,13 @@
 from __future__ import annotations
 
 import re
-from typing import Any
 
-from flag_engine.context.types import EnvironmentContext, SegmentContext
+from flag_engine.context.types import (
+    EnvironmentContext,
+    SegmentCondition,
+    SegmentContext,
+    SegmentRule,
+)
 
 from flagsmith_sql_flag_engine.dialect import Dialect
 from flagsmith_sql_flag_engine.dialects.snowflake import SnowflakeDialect
@@ -213,7 +217,11 @@ def _semver_sort_key_expr(ctx: TranslateContext, value_sql: str) -> str:
 
 
 def _comparison(
-    ctx: TranslateContext, op: str, expr: str, value: object, is_jsonpath: bool = False
+    ctx: TranslateContext,
+    op: str,
+    expr: str,
+    value: object,
+    is_jsonpath: bool = False,
 ) -> str | None:
     """Emit a SQL fragment comparing `expr` against `value` per `op`.
 
@@ -237,21 +245,32 @@ def _comparison(
     if op == "NOT_CONTAINS":
         return f"({expr} IS NOT NULL AND NOT ({d.position(lit, str_expr)}))"
     if op in {"GREATER_THAN", "LESS_THAN", "GREATER_THAN_INCLUSIVE", "LESS_THAN_INCLUSIVE"}:
+        # Validate as numeric BEFORE interpolating — segment values come from
+        # `MANAGE_SEGMENTS`-permissioned users, who shouldn't be able to inject
+        # SQL via a comparator value. ValueError → untranslatable, caller
+        # falls back to the engine UDF or surfaces an error upstream.
+        try:
+            numeric_value = float(str(value))
+        except (TypeError, ValueError):
+            return None
         sql_op = {
             "GREATER_THAN": ">",
             "LESS_THAN": "<",
             "GREATER_THAN_INCLUSIVE": ">=",
             "LESS_THAN_INCLUSIVE": "<=",
         }[op]
-        return f"({expr} IS NOT NULL AND {d.cast_float(expr)} {sql_op} {value})"
+        return f"({expr} IS NOT NULL AND {d.cast_float(expr)} {sql_op} {numeric_value})"
     if op == "MODULO":
+        # Same validation discipline: divisor/remainder must parse as numbers.
         try:
-            divisor, remainder = str(value).split("|")
-            float(divisor)
-            float(remainder)
+            divisor_str, remainder_str = str(value).split("|")
+            divisor = float(divisor_str)
+            remainder = float(remainder_str)
         except (ValueError, AttributeError):
             return None
-        return f"({expr} IS NOT NULL AND ({d.mod(d.cast_number(expr), divisor)}) = {remainder})"
+        return (
+            f"({expr} IS NOT NULL AND ({d.mod(d.cast_number(expr), str(divisor))}) = {remainder})"
+        )
     if op == "REGEX":
         if not _regex_safe_for_re2(str(value)):
             return None
@@ -264,7 +283,7 @@ def _comparison(
 # ---------------------------------------------------------------------------
 
 
-def translate_condition(cond: dict[str, Any], ctx: TranslateContext) -> str | None:
+def translate_condition(cond: SegmentCondition, ctx: TranslateContext) -> str | None:
     op = cond["operator"]
     if op not in TRANSLATABLE_OPERATORS:
         return None
@@ -276,7 +295,10 @@ def translate_condition(cond: dict[str, Any], ctx: TranslateContext) -> str | No
     if op == "PERCENTAGE_SPLIT":
         if not ctx.segment_key:
             return None
-        threshold = float(val) if val is not None else 0.0
+        try:
+            threshold = float(str(val)) if val is not None else 0.0
+        except (TypeError, ValueError):
+            return None
         if not prop:
             value_expr = ctx.dialect.cast_string(ctx.identity_key_col)
         elif prop.startswith("$."):
@@ -345,7 +367,7 @@ def translate_condition(cond: dict[str, Any], ctx: TranslateContext) -> str | No
 # ---------------------------------------------------------------------------
 
 
-def translate_rule(rule: dict[str, Any], ctx: TranslateContext) -> str | None:
+def translate_rule(rule: SegmentRule, ctx: TranslateContext) -> str | None:
     children: list[str] = []
     for cond in rule.get("conditions") or []:
         sql = translate_condition(cond, ctx)
@@ -378,14 +400,14 @@ def translate_segment(segment: SegmentContext, ctx: TranslateContext) -> str | N
     The expression goes after `WHERE i.environment_id = '<env-key>' AND ...`.
     The caller composes the surrounding `SELECT ... FROM IDENTITIES i`.
     """
-    if ctx.segment_key is None and "key" in segment:
-        ctx = ctx.with_segment_key(str(segment["key"]))
+    if ctx.segment_key is None:
+        ctx = ctx.with_segment_key(segment["key"])
     rules = segment.get("rules") or []
     if not rules:
         return "FALSE"
     rule_sql: list[str] = []
     for r in rules:
-        sql = translate_rule(dict(r), ctx)
+        sql = translate_rule(r, ctx)
         if sql is None:
             return None
         rule_sql.append(f"({sql})")

tests/test_engine.py

@@ -35,21 +35,25 @@ def _all_case_segments() -> list[tuple[int, str]]:
     _all_case_segments(),
     ids=[f"case{i}-seg{k}" for i, k in _all_case_segments()],
 )
-def test_segment_matches_engine(
+def test_translate_segment__engine_test_data_case__matches_engine(
     case_idx: int,
     seg_key: str,
     loaded_cases: list[dict],
     parity_results: dict[tuple[int, str], bool | None],
 ) -> None:
+    # Given a (case, segment) pair from engine-test-data and the pre-computed
+    # batched parity_results dict mapping each pair to its SQL-evaluated bool
     case = loaded_cases[case_idx]
     eval_ctx = case["context"]
     segment = eval_ctx["segments"][seg_key]
-
     sql_match = parity_results[(case_idx, seg_key)]
     if sql_match is None:
         pytest.skip(f"segment uses untranslatable operator (case {case_idx} seg {seg_key})")
 
+    # When the engine evaluates the same (context, segment) in-memory
     engine_match = is_context_in_segment(eval_ctx, segment)
+
+    # Then the SQL-evaluated and engine-evaluated booleans agree
     assert engine_match == sql_match, (
         f"engine={engine_match} sql={sql_match} for case={case_idx} seg={seg_key}"
     )