Input sanitizer for benchmark result renderer (#2594)

Since `DataFrame.query` is potentially vulnerable we limit the possible
filter input to a fixed grammar that is roughly like this:

```
expr = left op right
left = ( expr ) | literal
right = ( expr ) | literal
op = in | >= | < | <= | == | and | or
```

this will give us boolean operations and basic comparisons. Note that
`literal` can be arbitrary python literals (strings, tuples, ...).

Author: githubnemo

method_comparison/__init__.py

@@ -21,6 +21,7 @@
 import plotly.express as px
 import plotly.graph_objects as go
 from processing import load_df
+from sanitizer import parse_and_filter
 
 
 metric_preferences = {
@@ -246,7 +247,8 @@ def update_on_task(task_name, current_filter):
             filtered = filter_data(task_name, new_models[0] if new_models else "", df)
             if current_filter.strip():
                 try:
-                    df_queried = filtered.query(current_filter)
+                    mask = parse_and_filter(filtered, current_filter)
+                    df_queried = filtered[mask]
                     if not df_queried.empty:
                         filtered = df_queried
                 except Exception:
@@ -262,7 +264,8 @@ def update_on_model(task_name, model_id, current_filter):
             filtered = filter_data(task_name, model_id, df)
             if current_filter.strip():
                 try:
-                    filtered = filtered.query(current_filter)
+                    mask = parse_and_filter(filtered, current_filter)
+                    filtered = filtered[mask]
                 except Exception:
                     pass
             return filtered
@@ -275,7 +278,8 @@ def update_pareto_plot_and_summary(task_name, model_id, metric_x, metric_y, curr
             filtered = filter_data(task_name, model_id, df)
             if current_filter.strip():
                 try:
-                    filtered = filtered.query(current_filter)
+                    mask = parse_and_filter(filtered, current_filter)
+                    filtered = filtered[mask]
                 except Exception as e:
                     return generate_pareto_plot(filtered, metric_x, metric_y), f"Filter error: {e}"
 
@@ -295,7 +299,8 @@ def apply_filter(filter_query, task_name, model_id, metric_x, metric_y):
             filtered = filter_data(task_name, model_id, df)
             if filter_query.strip():
                 try:
-                    filtered = filtered.query(filter_query)
+                    mask = parse_and_filter(filtered, filter_query)
+                    filtered = filtered[mask]
                 except Exception as e:
                     # Update the table, plot, and summary even if there is a filter error.
                     return (

method_comparison/app.py

@@ -0,0 +1,100 @@
+import ast
+
+import pandas as pd
+
+
+def _evaluate_node(df, node):
+    """
+    Recursively evaluates an AST node to generate a pandas boolean mask.
+    """
+    # Base Case: A simple comparison like 'price > 100'
+    if isinstance(node, ast.Compare):
+        if not isinstance(node.left, ast.Name):
+            raise ValueError("Left side of comparison must be a column name.")
+        col = node.left.id
+        if col not in df.columns:
+            raise ValueError(f"Column '{col}' not found in DataFrame.")
+
+        if len(node.ops) > 1:
+            raise ValueError("Chained comparisons like '10 < price < 100' are not supported.")
+
+        op_node = node.ops[0]
+        val_node = node.comparators[0]
+        try:
+            value = ast.literal_eval(val_node)
+        except ValueError:
+            raise ValueError("Right side of comparison must be a literal (number, string, list).")
+
+        operator_map = {
+            ast.Gt:    lambda c, v: df[c] > v,
+            ast.GtE:   lambda c, v: df[c] >= v,
+            ast.Lt:    lambda c, v: df[c] < v,
+            ast.LtE:   lambda c, v: df[c] <= v,
+            ast.Eq:    lambda c, v: df[c] == v,
+            ast.NotEq: lambda c, v: df[c] != v,
+            ast.In:    lambda c, v: df[c].isin(v),
+            ast.NotIn: lambda c, v: ~df[c].isin(v)
+        }
+        op_type = type(op_node)
+        if op_type not in operator_map:
+            raise ValueError(f"Unsupported operator '{op_type.__name__}'.")
+        return operator_map[op_type](col, value)
+
+    # Recursive Step: "Bitwise" operation & and | (the same as boolean operations)
+    elif isinstance(node, ast.BinOp):
+        if isinstance(node.op, ast.BitOr):
+            return _evaluate_node(df, node.left) | _evaluate_node(df, node.right)
+        elif isinstance(node.op, ast.BitAnd):
+            return _evaluate_node(df, node.left) & _evaluate_node(df, node.right)
+
+    # Recursive Step: A boolean operation like '... and ...' or '... or ...'
+    elif isinstance(node, ast.BoolOp):
+        op_type = type(node.op)
+        # Evaluate the first value in the boolean expression
+        result = _evaluate_node(df, node.values[0])
+        # Combine it with the rest of the values based on the operator
+        for i in range(1, len(node.values)):
+            if op_type is ast.And or op_type is ast.BitAnd:
+                result &= _evaluate_node(df, node.values[i])
+            elif op_type is ast.Or or op_type is ast.BitOr:
+                result |= _evaluate_node(df, node.values[i])
+        return result
+
+    elif isinstance(node, ast.UnaryOp):
+        if not isinstance(node.op, ast.Not):
+            raise ValueError("Only supported unary op is negation.")
+        return ~_evaluate_node(df, node.operand)
+
+    # If the node is not a comparison or boolean op, it's an unsupported expression type
+    else:
+        raise ValueError(f"Unsupported expression type: {type(node).__name__}")
+
+
+def parse_and_filter(df, filter_str):
+    """
+    Filters a pandas DataFrame using a string expression parsed by AST.
+    This is done to avoid the security vulnerables that `DataFrame.query`
+    brings (arbitrary code execution).
+
+    Args:
+        df (pd.DataFrame): The DataFrame to filter.
+        filter_str (str): A string representing a filter expression.
+                          e.g., "price > 100 and stock < 50"
+                          Supported operators: >, >=, <, <=, ==, !=, in, not in, and, or.
+
+    Returns:
+        pd.Series: A boolean Series representing the filter mask.
+    """
+    if not filter_str:
+        return pd.Series([True] * len(df), index=df.index)
+
+    try:
+        # 'eval' mode ensures the source is a single expression.
+        tree = ast.parse(filter_str, mode='eval')
+        expression_node = tree.body
+    except (SyntaxError, ValueError) as e:
+        raise ValueError(f"Invalid filter syntax: {e}")
+
+    # The recursive evaluation starts here
+    mask = _evaluate_node(df, expression_node)
+    return mask

method_comparison/sanitizer.py

@@ -0,0 +1,38 @@
+import pandas as pd
+import pytest
+
+from .sanitizer import parse_and_filter
+
+
+@pytest.fixture
+def df_products():
+    data = {
+        'product_id': [101, 102, 103, 104, 105, 106],
+        'category': ['Electronics', 'Books', 'Electronics', 'Home Goods', 'Books', 'Electronics'],
+        'price': [799.99, 19.99, 49.50, 120.00, 24.99, 150.00],
+        'stock': [15, 300, 50, 25, 150, 0]
+    }
+    return pd.DataFrame(data)
+
+
+def test_exploit_fails(df_products):
+    with pytest.raises(ValueError) as e:
+        mask1 = parse_and_filter(df_products,
+            """price < 50 and @os.system("/bin/echo password")""")
+    assert 'Invalid filter syntax' in str(e)
+
+
+@pytest.mark.parametrize('expression,ids', [
+    ("price < 50", [102, 103, 105]),
+    ("product_id in [101, 102]", [101, 102]),
+    ("price < 50 and category == 'Electronics'", [103]),
+    ("stock < 100 or category == 'Home Goods'", [101, 103, 104, 106]),
+    ("(price > 100 and stock < 20) or category == 'Books'", [101, 102, 105, 106]),
+    ("not (price > 50 or stock > 100)", [103]),
+    ("not price > 50", [102, 103, 105]),
+    ("(price < 50) & (category == 'Electronics')", [103]),
+    ("(stock < 100) | (category == 'Home Goods')", [101, 103, 104, 106]),
+])
+def test_operations(df_products, expression, ids):
+    mask1 = parse_and_filter(df_products, expression)
+    assert sorted(df_products[mask1].product_id) == sorted(ids)