# Code Injection via eval() on User-Supplied Expressions

Language: Python
Severity: Critical
CWE: CWE-94

## Source
9

## Flow
9-10

## Sink
10

## Vulnerable Code
```python
@app.route('/iot/device/telemetry', methods=['POST'])
def process_sensor_data():
    device_id = request.json.get('device_id')
    metric_formula = request.json.get('transform')
    raw_temp = request.json.get('temperature', 0)
    raw_humidity = request.json.get('humidity', 0)
    raw_pressure = request.json.get('pressure', 0)
    context = {'temp': raw_temp, 'hum': raw_humidity, 'press': raw_pressure}
    if metric_formula:
        computed_value = eval(metric_formula, {"__builtins__": {}}, context)
        store_telemetry(device_id, computed_value)
        return jsonify({'status': 'processed', 'result': computed_value})
    return jsonify({'error': 'missing transform'}), 400
```

## Explanation

The application accepts a user-controlled 'transform' parameter containing arbitrary Python expressions and directly passes it to eval() on line 10. Despite attempting to restrict builtins, attackers can still access dangerous functions through object introspection and execute arbitrary code on the server.

## Remediation

The fix replaces the dangerous eval() call with a custom AST-based expression parser that only allows arithmetic operations (addition, subtraction, multiplication, division, modulo, power) on numeric constants and pre-defined sensor variables. This completely eliminates the code injection vector since no arbitrary Python code, function calls, attribute access, or imports can be executed through the safe evaluator.

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

# Safe mathematical operations whitelist
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,
}

def safe_eval_expression(expr, variables):
    """Safely evaluate a mathematical expression with only arithmetic operations."""
    try:
        tree = ast.parse(expr, 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.Constant):
            if isinstance(node.value, (int, float)):
                return node.value
            raise ValueError(f"Unsupported constant type: {type(node.value).__name__}")
        elif isinstance(node, ast.Name):
            if node.id in variables:
                value = variables[node.id]
                if isinstance(value, (int, float)):
                    return value
                raise ValueError(f"Variable '{node.id}' must be numeric")
            raise ValueError(f"Unknown variable: {node.id}")
        elif isinstance(node, ast.BinOp):
            op_type = type(node.op)
            if op_type not in SAFE_OPERATORS:
                raise ValueError(f"Unsupported operator: {op_type.__name__}")
            left = _eval_node(node.left)
            right = _eval_node(node.right)
            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):
            op_type = type(node.op)
            if op_type not in SAFE_OPERATORS:
                raise ValueError(f"Unsupported unary operator: {op_type.__name__}")
            operand = _eval_node(node.operand)
            return SAFE_OPERATORS[op_type](operand)
        else:
            raise ValueError(f"Unsupported expression element: {type(node).__name__}")

    return _eval_node(tree)


@app.route('/iot/device/telemetry', methods=['POST'])
def process_sensor_data():
    device_id = request.json.get('device_id')
    metric_formula = request.json.get('transform')
    raw_temp = request.json.get('temperature', 0)
    raw_humidity = request.json.get('humidity', 0)
    raw_pressure = request.json.get('pressure', 0)

    # Validate that sensor inputs are numeric
    if not all(isinstance(v, (int, float)) for v in [raw_temp, raw_humidity, raw_pressure]):
        return jsonify({'error': 'sensor values must be numeric'}), 400

    context = {'temp': raw_temp, 'hum': raw_humidity, 'press': raw_pressure}

    if metric_formula:
        if not isinstance(metric_formula, str) or len(metric_formula) > 200:
            return jsonify({'error': 'invalid transform formula'}), 400
        try:
            computed_value = safe_eval_expression(metric_formula, context)
            store_telemetry(device_id, computed_value)
            return jsonify({'status': 'processed', 'result': computed_value})
        except (ValueError, ZeroDivisionError, OverflowError) as e:
            return jsonify({'error': f'transform error: {str(e)}'}), 400
    return jsonify({'error': 'missing transform'}), 400
```