# Code Injection via eval() on Untrusted Input

Language: Python
Severity: Critical
CWE: CWE-95

## Source
4

## Flow
4-6

## Sink
6

## Vulnerable Code
```python
@app.route('/iot/device/configure', methods=['POST'])
def apply_device_config():
    device_id = request.json.get('device_id')
    threshold_expr = request.json.get('threshold_expression')
    sensor_data = iot_db.get_latest_reading(device_id)
    try:
        computed_threshold = eval(threshold_expr, {'sensor': sensor_data, 'math': __import__('math')})
        if sensor_data['value'] > computed_threshold:
            trigger_alert(device_id)
            return jsonify({'status': 'alert_triggered', 'threshold': computed_threshold})
        return jsonify({'status': 'normal', 'threshold': computed_threshold})
    except Exception as e:
        return jsonify({'error': 'Invalid threshold expression'}), 400
```

## Explanation

The application accepts user-controlled input 'threshold_expression' and directly passes it to eval() function. Although the eval() uses a restricted namespace with only 'sensor' and 'math', attackers can escape this restriction using Python introspection techniques like __import__, __builtins__, or accessing object.__subclasses__() to execute arbitrary code.

## Remediation

The fix replaces the dangerous eval() call with a custom AST-based safe expression evaluator that parses the expression into an abstract syntax tree and only allows specific, whitelisted operations: basic arithmetic operators, a curated set of math functions, math constants, and access to sensor data fields. This approach makes it impossible for attackers to execute arbitrary code, import modules, or access Python internals since only explicitly permitted node types are evaluated.

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

# Safe mathematical expression evaluator
SAFE_OPERATORS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.FloorDiv: operator.floordiv,
    ast.Mod: operator.mod,
    ast.Pow: operator.pow,
    ast.USub: operator.neg,
    ast.UAdd: operator.pos,
}

SAFE_MATH_FUNCTIONS = {
    'sqrt': math.sqrt,
    'log': math.log,
    'log10': math.log10,
    'exp': math.exp,
    'ceil': math.ceil,
    'floor': math.floor,
    'abs': abs,
    'min': min,
    'max': max,
    'pow': pow,
    'round': round,
}

SAFE_MATH_CONSTANTS = {
    'pi': math.pi,
    'e': math.e,
}


def safe_eval_expression(expr, sensor_data):
    """Safely evaluate a mathematical expression with sensor data context."""
    try:
        tree = ast.parse(expr, mode='eval')
    except SyntaxError:
        raise ValueError("Invalid expression syntax")

    return _eval_node(tree.body, sensor_data)


def _eval_node(node, sensor_data):
    if isinstance(node, ast.Constant):
        if isinstance(node.value, (int, float)):
            return node.value
        raise ValueError(f"Unsupported constant type: {type(node.value)}")

    elif isinstance(node, ast.BinOp):
        left = _eval_node(node.left, sensor_data)
        right = _eval_node(node.right, sensor_data)
        op_type = type(node.op)
        if op_type not in SAFE_OPERATORS:
            raise ValueError(f"Unsupported operator: {op_type.__name__}")
        if op_type == ast.Pow and right > 100:
            raise ValueError("Exponent too large")
        return SAFE_OPERATORS[op_type](left, right)

    elif isinstance(node, ast.UnaryOp):
        operand = _eval_node(node.operand, sensor_data)
        op_type = type(node.op)
        if op_type not in SAFE_OPERATORS:
            raise ValueError(f"Unsupported unary operator: {op_type.__name__}")
        return SAFE_OPERATORS[op_type](operand)

    elif isinstance(node, ast.Call):
        if isinstance(node.func, ast.Name) and node.func.id in SAFE_MATH_FUNCTIONS:
            args = [_eval_node(arg, sensor_data) for arg in node.args]
            return SAFE_MATH_FUNCTIONS[node.func.id](*args)
        raise ValueError(f"Unsupported function call")

    elif isinstance(node, ast.Name):
        name = node.id
        if name in SAFE_MATH_CONSTANTS:
            return SAFE_MATH_CONSTANTS[name]
        raise ValueError(f"Unsupported variable: {name}")

    elif isinstance(node, ast.Attribute):
        if isinstance(node.value, ast.Name) and node.value.id == 'sensor':
            attr = node.attr
            if attr in sensor_data:
                value = sensor_data[attr]
                if isinstance(value, (int, float)):
                    return value
                raise ValueError(f"Sensor attribute '{attr}' is not numeric")
            raise ValueError(f"Unknown sensor attribute: {attr}")
        raise ValueError("Unsupported attribute access")

    elif isinstance(node, ast.Subscript):
        if isinstance(node.value, ast.Name) and node.value.id == 'sensor':
            if isinstance(node.slice, ast.Constant) and isinstance(node.slice.value, str):
                key = node.slice.value
                if key in sensor_data:
                    value = sensor_data[key]
                    if isinstance(value, (int, float)):
                        return value
                    raise ValueError(f"Sensor key '{key}' is not numeric")
                raise ValueError(f"Unknown sensor key: {key}")
        raise ValueError("Unsupported subscript access")

    else:
        raise ValueError(f"Unsupported expression element: {type(node).__name__}")


@app.route('/iot/device/configure', methods=['POST'])
def apply_device_config():
    device_id = request.json.get('device_id')
    threshold_expr = request.json.get('threshold_expression')

    if not device_id or not threshold_expr:
        return jsonify({'error': 'Missing required fields'}), 400

    if len(threshold_expr) > 200:
        return jsonify({'error': 'Expression too long'}), 400

    sensor_data = iot_db.get_latest_reading(device_id)
    try:
        computed_threshold = safe_eval_expression(threshold_expr, sensor_data)
        if not isinstance(computed_threshold, (int, float)):
            return jsonify({'error': 'Expression must evaluate to a number'}), 400
        if sensor_data['value'] > computed_threshold:
            trigger_alert(device_id)
            return jsonify({'status': 'alert_triggered', 'threshold': computed_threshold})
        return jsonify({'status': 'normal', 'threshold': computed_threshold})
    except ValueError as e:
        return jsonify({'error': f'Invalid threshold expression: {str(e)}'}), 400
    except (ZeroDivisionError, OverflowError) as e:
        return jsonify({'error': f'Mathematical error: {str(e)}'}), 400
```