# Code Injection via eval on Untrusted Input

Language: Python
Severity: Critical
CWE: CWE-94

## Source
1

## Flow
1-2-3-4-5-6

## Sink
6

## Vulnerable Code
```python
def apply_iot_device_threshold(device_id, threshold_expr):
    sensor_data = fetch_sensor_readings(device_id)
    max_temp = sensor_data.get('temperature', 0)
    max_humidity = sensor_data.get('humidity', 0)
    max_pressure = sensor_data.get('pressure', 0)
    alert_condition = eval(threshold_expr)
    if alert_condition:
        trigger_device_alert(device_id, "Threshold exceeded")
        return {"status": "alert_triggered", "condition": threshold_expr}
    return {"status": "normal", "condition": threshold_expr}
```

## Explanation

The function accepts untrusted user input via the 'threshold_expr' parameter and directly passes it to the eval() function without any validation or sanitization. This allows an attacker to execute arbitrary Python code by injecting malicious expressions through the threshold configuration interface.

## 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 only evaluates nodes that are explicitly whitelisted: numeric constants, approved variable names (max_temp, max_humidity, max_pressure), basic arithmetic operators, comparison operators, and boolean logic (and/or). Any attempt to use function calls, attribute access, imports, or other dangerous constructs will raise a ValueError.

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

ALLOWED_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,
}

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

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


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

    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)}")
        elif isinstance(node, ast.Name):
            if node.id in variables:
                return variables[node.id]
            raise ValueError(f"Unsupported variable name: {node.id}")
        elif isinstance(node, ast.BinOp):
            op_type = type(node.op)
            if op_type not in ALLOWED_OPERATORS:
                raise ValueError(f"Unsupported binary operator: {op_type.__name__}")
            left = _eval_node(node.left)
            right = _eval_node(node.right)
            return ALLOWED_OPERATORS[op_type](left, right)
        elif isinstance(node, ast.UnaryOp):
            op_type = type(node.op)
            if op_type not in ALLOWED_OPERATORS:
                raise ValueError(f"Unsupported unary operator: {op_type.__name__}")
            operand = _eval_node(node.operand)
            return ALLOWED_OPERATORS[op_type](operand)
        elif isinstance(node, ast.Compare):
            left = _eval_node(node.left)
            result = True
            for op, comparator in zip(node.ops, node.comparators):
                op_type = type(op)
                if op_type not in ALLOWED_COMPARISONS:
                    raise ValueError(f"Unsupported comparison operator: {op_type.__name__}")
                right = _eval_node(comparator)
                if not ALLOWED_COMPARISONS[op_type](left, right):
                    result = False
                    break
                left = right
            return result
        elif isinstance(node, ast.BoolOp):
            op_type = type(node.op)
            if op_type not in ALLOWED_BOOL_OPS:
                raise ValueError(f"Unsupported 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.IfExp):
            raise ValueError("Conditional expressions are not allowed")
        else:
            raise ValueError(f"Unsupported expression node: {type(node).__name__}")

    return _eval_node(tree)


def apply_iot_device_threshold(device_id, threshold_expr):
    sensor_data = fetch_sensor_readings(device_id)
    max_temp = sensor_data.get('temperature', 0)
    max_humidity = sensor_data.get('humidity', 0)
    max_pressure = sensor_data.get('pressure', 0)

    allowed_variables = {
        'max_temp': max_temp,
        'max_humidity': max_humidity,
        'max_pressure': max_pressure,
    }

    try:
        alert_condition = safe_eval_threshold(threshold_expr, allowed_variables)
    except ValueError as e:
        return {"status": "error", "message": str(e)}

    if alert_condition:
        trigger_device_alert(device_id, "Threshold exceeded")
        return {"status": "alert_triggered", "condition": threshold_expr}
    return {"status": "normal", "condition": threshold_expr}
```