# Code Injection via exec() on User-Controlled Expression

Language: Python
Severity: Critical
CWE: CWE-94

## Source
4

## Flow
4-7

## Sink
7

## Vulnerable Code
```python
@app.route('/iot/device/calibrate', methods=['POST'])
def calibrate_sensor():
    device_id = request.json.get('device_id')
    calibration_formula = request.json.get('formula', 'value * 1.0')
    sensor_value = float(request.json.get('raw_value', 0))
    result_var = {}
    exec(f"calibrated = {calibration_formula}", {'value': sensor_value, '__builtins__': {}}, result_var)
    db.execute("UPDATE iot_sensors SET calibrated_value = ? WHERE id = ?", (result_var.get('calibrated'), device_id))
    return jsonify({'device': device_id, 'calibrated_value': result_var.get('calibrated')})
```

## Explanation

The calibration_formula parameter from user input is directly interpolated into an exec() call without proper sanitization. Although __builtins__ is restricted to an empty dict, attackers can still access dangerous functions through existing objects in the namespace or use attribute access to execute arbitrary code.

## Remediation

The fix replaces the dangerous exec() call with a safe AST-based mathematical expression evaluator that only allows arithmetic operations, specific safe functions (like sqrt, log, sin), and named variables. The evaluator parses the expression into an AST and walks each node, rejecting any construct that isn't a number, allowed variable, safe operator, or whitelisted function call—completely preventing code injection, attribute access, and import statements.

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

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

SAFE_FUNCTIONS = {
    'abs': abs,
    'round': round,
    'min': min,
    'max': max,
    'int': int,
    'float': float,
}

import math
SAFE_CONSTANTS = {
    'pi': math.pi,
    'e': math.e,
}

SAFE_MATH_FUNCTIONS = {
    'sqrt': math.sqrt,
    'log': math.log,
    'log10': math.log10,
    'sin': math.sin,
    'cos': math.cos,
    'tan': math.tan,
    'pow': math.pow,
}

SAFE_FUNCTIONS.update(SAFE_MATH_FUNCTIONS)


def safe_eval_expr(expr_str, variables):
    """Safely evaluate a mathematical expression with given variables."""
    try:
        tree = ast.parse(expr_str, mode='eval')
    except SyntaxError:
        raise ValueError(f"Invalid expression syntax: {expr_str}")
    return _eval_node(tree.body, variables)


def _eval_node(node, variables):
    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.Name):
        if node.id in variables:
            return variables[node.id]
        if node.id in SAFE_CONSTANTS:
            return SAFE_CONSTANTS[node.id]
        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, variables)
        right = _eval_node(node.right, variables)
        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, variables)
        return SAFE_OPERATORS[op_type](operand)
    elif isinstance(node, ast.Call):
        if isinstance(node.func, ast.Name) and node.func.id in SAFE_FUNCTIONS:
            args = [_eval_node(arg, variables) for arg in node.args]
            if node.keywords:
                raise ValueError("Keyword arguments not supported")
            return SAFE_FUNCTIONS[node.func.id](*args)
        raise ValueError(f"Unsupported function call")
    elif isinstance(node, ast.IfExp):
        raise ValueError("Conditional expressions not supported")
    elif isinstance(node, ast.Compare):
        raise ValueError("Comparison expressions not supported in calibration formulas")
    elif isinstance(node, ast.Attribute):
        raise ValueError("Attribute access not allowed")
    else:
        raise ValueError(f"Unsupported expression type: {type(node).__name__}")


@app.route('/iot/device/calibrate', methods=['POST'])
def calibrate_sensor():
    device_id = request.json.get('device_id')
    calibration_formula = request.json.get('formula', 'value * 1.0')
    sensor_value = float(request.json.get('raw_value', 0))

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

    try:
        calibrated = safe_eval_expr(calibration_formula, {'value': sensor_value})
    except (ValueError, TypeError, ZeroDivisionError, OverflowError) as e:
        return jsonify({'error': f'Invalid calibration formula: {str(e)}'}), 400

    db.execute("UPDATE iot_sensors SET calibrated_value = ? WHERE id = ?", (calibrated, device_id))
    return jsonify({'device': device_id, 'calibrated_value': calibrated})
```