# Code Injection via exec on Untrusted Input

Language: Python
Severity: Critical
CWE: CWE-94

## 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')
    config_expr = request.json.get('threshold_logic')
    device_state = {'temp': 25, 'humidity': 60, 'device_id': device_id}
    try:
        exec(f"alert_triggered = {config_expr}", {'__builtins__': {}}, device_state)
        return jsonify({'device': device_id, 'alert': device_state.get('alert_triggered', False)})
    except Exception as e:
        return jsonify({'error': 'Configuration failed'}), 400
```

## Explanation

User-controlled input from 'threshold_logic' is directly passed to exec() without proper sanitization. Although __builtins__ is restricted, attackers can still manipulate the device_state namespace to execute arbitrary Python code, access object methods, or perform denial of service attacks through the local namespace.

## Remediation

The fix replaces the dangerous exec() call with a custom AST-based expression evaluator that only allows safe operations: numeric comparisons (>, <, >=, <=, ==, !=), boolean operators (and, or), numeric constants, and a whitelist of allowed variable names (temp, humidity). This prevents any arbitrary code execution while still supporting legitimate threshold expressions like 'temp > 30 and humidity < 80'.

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

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

ALLOWED_BOOL_OPS = {
    ast.And: all,
    ast.Or: any,
}

ALLOWED_VARIABLES = {'temp', 'humidity'}


def safe_eval_threshold(expr, device_state):
    """Safely evaluate a threshold expression using AST parsing.
    Only allows comparisons, boolean operations, numbers, and whitelisted variables."""
    try:
        tree = ast.parse(expr, mode='eval')
    except SyntaxError:
        raise ValueError("Invalid expression syntax")

    return _eval_node(tree.body, device_state)


def _eval_node(node, device_state):
    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 ALLOWED_VARIABLES:
            return device_state[node.id]
        raise ValueError(f"Disallowed variable: {node.id}")

    elif isinstance(node, ast.Compare):
        left = _eval_node(node.left, device_state)
        for op, comparator in zip(node.ops, node.comparators):
            op_func = ALLOWED_OPERATORS.get(type(op))
            if op_func is None:
                raise ValueError(f"Unsupported operator: {type(op).__name__}")
            right = _eval_node(comparator, device_state)
            if not op_func(left, right):
                return False
            left = right
        return True

    elif isinstance(node, ast.BoolOp):
        bool_func = ALLOWED_BOOL_OPS.get(type(node.op))
        if bool_func is None:
            raise ValueError(f"Unsupported boolean operator: {type(node.op).__name__}")
        values = [_eval_node(v, device_state) for v in node.values]
        return bool_func(values)

    elif isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
        operand = _eval_node(node.operand, device_state)
        if isinstance(operand, (int, float)):
            return -operand
        raise ValueError("Unary minus only allowed on numbers")

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


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

    if not isinstance(config_expr, str) or len(config_expr) > 200:
        return jsonify({'error': 'Invalid or too long threshold expression'}), 400

    device_state = {'temp': 25, 'humidity': 60, 'device_id': device_id}
    try:
        alert_triggered = safe_eval_threshold(config_expr, device_state)
        return jsonify({'device': device_id, 'alert': bool(alert_triggered)})
    except (ValueError, KeyError) as e:
        return jsonify({'error': 'Configuration failed'}), 400
```