# Arbitrary Code Execution via unsafe eval() on Untrusted Expressions

Language: Python
Severity: Critical
CWE: CWE-95

## Source
4

## Flow
4-8

## Sink
8

## Vulnerable Code
```python
@app.route('/iot/device/threshold', methods=['POST'])
def configure_sensor_threshold():
    device_id = request.json.get('device_id')
    threshold_expr = request.json.get('threshold_expression')
    sensor_data = iot_db.get_latest_reading(device_id)
    context = {'temp': sensor_data['temperature'], 'humidity': sensor_data['humidity'], 'pressure': sensor_data['pressure']}
    try:
        alert_triggered = eval(threshold_expr, {'__builtins__': {}}, context)
        if alert_triggered:
            trigger_iot_alert(device_id, sensor_data)
        return jsonify({'status': 'threshold_evaluated', 'alert': bool(alert_triggered)})
    except Exception as e:
        return jsonify({'error': 'evaluation_failed'}), 400
```

## Explanation

The application accepts untrusted user input via 'threshold_expression' and directly passes it to eval() on line 8. Although __builtins__ is restricted, this protection is bypassable using techniques like accessing object.__subclasses__() through the context variables, allowing arbitrary code execution.

## Remediation

The fix replaces the dangerous eval() call with a custom safe expression evaluator built on Python's ast module. The safe_eval_threshold function parses the expression into an AST and recursively evaluates only explicitly allowed node types: comparisons, basic arithmetic, boolean logic, numeric constants, and a whitelist of variable names (temp, humidity, pressure). Any attempt to use function calls, attribute access, imports, or other dangerous constructs is rejected.

## Secure Code
```python
import ast
import operator

ALLOWED_OPERATORS = {
    ast.Lt: operator.lt,
    ast.LtE: operator.le,
    ast.Gt: operator.gt,
    ast.GtE: operator.ge,
    ast.Eq: operator.eq,
    ast.NotEq: operator.ne,
}

ALLOWED_BOOL_OPS = {
    ast.And: lambda values: all(values),
    ast.Or: lambda values: any(values),
}

ALLOWED_UNARY_OPS = {
    ast.Not: operator.not_,
    ast.USub: operator.neg,
}

ALLOWED_BIN_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
}

ALLOWED_VARIABLES = {'temp', 'humidity', 'pressure'}


def safe_eval_threshold(expr_str, context):
    """Safely evaluate a threshold expression using AST parsing.
    Only allows comparisons, basic arithmetic, boolean logic, numbers, and allowed variable names."""
    try:
        tree = ast.parse(expr_str, mode='eval')
    except SyntaxError:
        raise ValueError("Invalid expression syntax")

    def _eval_node(node):
        if isinstance(node, ast.Expression):
            return _eval_node(node.body)
        elif isinstance(node, ast.Compare):
            left = _eval_node(node.left)
            for op, comparator in zip(node.ops, node.comparators):
                op_type = type(op)
                if op_type not in ALLOWED_OPERATORS:
                    raise ValueError(f"Disallowed comparison operator: {op_type.__name__}")
                right = _eval_node(comparator)
                if not ALLOWED_OPERATORS[op_type](left, right):
                    return False
                left = right
            return True
        elif isinstance(node, ast.BoolOp):
            op_type = type(node.op)
            if op_type not in ALLOWED_BOOL_OPS:
                raise ValueError(f"Disallowed boolean operator: {op_type.__name__}")
            values = [_eval_node(v) for v in node.values]
            return ALLOWED_BOOL_OPS[op_type](values)
        elif isinstance(node, ast.UnaryOp):
            op_type = type(node.op)
            if op_type not in ALLOWED_UNARY_OPS:
                raise ValueError(f"Disallowed unary operator: {op_type.__name__}")
            operand = _eval_node(node.operand)
            return ALLOWED_UNARY_OPS[op_type](operand)
        elif isinstance(node, ast.BinOp):
            op_type = type(node.op)
            if op_type not in ALLOWED_BIN_OPS:
                raise ValueError(f"Disallowed binary operator: {op_type.__name__}")
            left = _eval_node(node.left)
            right = _eval_node(node.right)
            return ALLOWED_BIN_OPS[op_type](left, right)
        elif isinstance(node, ast.Num):
            return node.n
        elif isinstance(node, ast.Constant):
            if isinstance(node.value, (int, float)):
                return node.value
            raise ValueError(f"Disallowed constant type: {type(node.value).__name__}")
        elif isinstance(node, ast.Name):
            if node.id not in ALLOWED_VARIABLES:
                raise ValueError(f"Disallowed variable: {node.id}")
            if node.id not in context:
                raise ValueError(f"Unknown variable: {node.id}")
            return context[node.id]
        else:
            raise ValueError(f"Disallowed expression type: {type(node).__name__}")

    return _eval_node(tree)


@app.route('/iot/device/threshold', methods=['POST'])
def configure_sensor_threshold():
    device_id = request.json.get('device_id')
    threshold_expr = request.json.get('threshold_expression')
    if not threshold_expr or not isinstance(threshold_expr, str):
        return jsonify({'error': 'invalid_expression'}), 400
    if len(threshold_expr) > 200:
        return jsonify({'error': 'expression_too_long'}), 400
    sensor_data = iot_db.get_latest_reading(device_id)
    context = {'temp': sensor_data['temperature'], 'humidity': sensor_data['humidity'], 'pressure': sensor_data['pressure']}
    try:
        alert_triggered = safe_eval_threshold(threshold_expr, context)
        if alert_triggered:
            trigger_iot_alert(device_id, sensor_data)
        return jsonify({'status': 'threshold_evaluated', 'alert': bool(alert_triggered)})
    except ValueError as e:
        return jsonify({'error': 'evaluation_failed', 'detail': str(e)}), 400
    except Exception as e:
        return jsonify({'error': 'evaluation_failed'}), 400
```