# Code Injection via unsafe eval()

Language: Python
Severity: Critical
CWE: CWE-94

## Source
5

## Flow
5-6-7-8

## Sink
8

## Vulnerable Code
```python
from flask import Flask, request, jsonify
app = Flask(__name__)
device_registry = {'sensor_01': 45.2, 'sensor_02': 78.9, 'actuator_03': 1}
def evaluate_iot_threshold(sensor_id):
    raw_expr = request.args.get('threshold_rule', '')
    current_val = device_registry.get(sensor_id, 0)
    ctx = {'value': current_val, 'sensor': sensor_id}
    result = eval(raw_expr, {'__builtins__': __builtins__}, ctx)
    return jsonify({'sensor': sensor_id, 'triggered': bool(result)})
@app.route('/iot/check/<sensor_id>')
def iot_check(sensor_id):
    return evaluate_iot_threshold(sensor_id)
```

## Explanation

The `threshold_rule` query parameter is fetched directly from user-controlled HTTP input on line 5 without any sanitization or validation. It is then passed as the expression argument to `eval()` on line 8 with `__builtins__` explicitly exposed, meaning any Python code—including OS commands via `__import__`—can be executed server-side. Despite the appearance of a sandboxed context dict, passing `__builtins__: __builtins__` (the real builtins module) completely negates any sandboxing intent.

## Remediation

The dangerous eval() call has been completely replaced with a custom AST-based expression evaluator that only permits comparisons, basic arithmetic, boolean logic, numeric constants, and a whitelisted set of variable names (only 'value'). The expression is parsed into an AST tree and each node is validated against an allowlist of safe node types, making it impossible to call functions, import modules, or access arbitrary attributes. Additionally, input length is bounded, sensor_id is validated against the registry, and errors are returned as 400 responses.

## Secure Code
```python
from flask import Flask, request, jsonify
import ast
import operator

app = Flask(__name__)
device_registry = {'sensor_01': 45.2, 'sensor_02': 78.9, 'actuator_03': 1}

# Safe operators for threshold evaluation
_SAFE_OPS = {
    ast.Gt: operator.gt,
    ast.Lt: operator.lt,
    ast.GtE: operator.ge,
    ast.LtE: operator.le,
    ast.Eq: operator.eq,
    ast.NotEq: operator.ne,
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.And: None,
    ast.Or: None,
    ast.Not: None,
}


def _safe_eval_node(node, ctx):
    """Recursively evaluate a restricted AST node against a safe context."""
    if isinstance(node, ast.Expression):
        return _safe_eval_node(node.body, ctx)
    elif isinstance(node, ast.Constant):  # numbers, strings, booleans
        if isinstance(node.value, (int, float, bool)):
            return node.value
        raise ValueError(f"Unsupported constant type: {type(node.value).__name__}")
    elif isinstance(node, ast.Name):
        if node.id in ctx:
            return ctx[node.id]
        raise ValueError(f"Unknown variable: {node.id}")
    elif isinstance(node, ast.Compare):
        left = _safe_eval_node(node.left, ctx)
        for op_node, comparator in zip(node.ops, node.comparators):
            op_func = _SAFE_OPS.get(type(op_node))
            if op_func is None:
                raise ValueError(f"Unsupported comparison operator: {type(op_node).__name__}")
            right = _safe_eval_node(comparator, ctx)
            if not op_func(left, right):
                return False
            left = right
        return True
    elif isinstance(node, ast.BinOp):
        op_func = _SAFE_OPS.get(type(node.op))
        if op_func is None:
            raise ValueError(f"Unsupported binary operator: {type(node.op).__name__}")
        left = _safe_eval_node(node.left, ctx)
        right = _safe_eval_node(node.right, ctx)
        return op_func(left, right)
    elif isinstance(node, ast.BoolOp):
        if isinstance(node.op, ast.And):
            return all(_safe_eval_node(v, ctx) for v in node.values)
        elif isinstance(node.op, ast.Or):
            return any(_safe_eval_node(v, ctx) for v in node.values)
        raise ValueError(f"Unsupported boolean operator: {type(node.op).__name__}")
    elif isinstance(node, ast.UnaryOp):
        if isinstance(node.op, ast.Not):
            return not _safe_eval_node(node.operand, ctx)
        elif isinstance(node.op, ast.USub):
            return -_safe_eval_node(node.operand, ctx)
        raise ValueError(f"Unsupported unary operator: {type(node.op).__name__}")
    else:
        raise ValueError(f"Unsupported expression node: {type(node).__name__}")


def safe_eval_threshold(expression, ctx):
    """Parse and evaluate a threshold expression using a restricted AST walker.
    Only comparisons, arithmetic, boolean logic, numbers, and whitelisted names are allowed.
    """
    if not expression or len(expression) > 200:
        raise ValueError("Invalid or excessively long threshold expression")
    try:
        tree = ast.parse(expression, mode='eval')
    except SyntaxError as e:
        raise ValueError(f"Invalid expression syntax: {e}")
    return _safe_eval_node(tree, ctx)


def evaluate_iot_threshold(sensor_id):
    raw_expr = request.args.get('threshold_rule', '')
    current_val = device_registry.get(sensor_id, 0)
    ctx = {'value': current_val}
    try:
        result = safe_eval_threshold(raw_expr, ctx)
    except ValueError as e:
        return jsonify({'error': str(e)}), 400
    return jsonify({'sensor': sensor_id, 'triggered': bool(result)})


@app.route('/iot/check/<sensor_id>')
def iot_check(sensor_id):
    if sensor_id not in device_registry:
        return jsonify({'error': 'Unknown sensor'}), 404
    return evaluate_iot_threshold(sensor_id)
```