# Code Injection via eval()

Language: Python
Severity: Critical
CWE: CWE-94

## Source
9

## Flow
9-11

## Sink
11

## Vulnerable Code
```python
from flask import Flask, request, jsonify
app = Flask(__name__)
sensor_limits = {}
@app.route('/iot/threshold', methods=['POST'])
def configure_threshold():
    dev_id = request.json.get('device_id')
    raw_expr = request.json.get('threshold_formula')
    try:
        computed_limit = eval(raw_expr)
        sensor_limits[dev_id] = computed_limit
        return jsonify({'status': 'ok', 'limit': computed_limit})
    except Exception as e:
        return jsonify({'error': str(e)}), 400
```

## Explanation

The `threshold_formula` field is read directly from untrusted user-supplied JSON and passed without any sanitization or validation to Python's built-in `eval()`. This allows an attacker to inject and execute arbitrary Python code on the server, including OS commands, file reads, or spawning reverse shells.

## Remediation

The dangerous `eval()` call has been replaced with a custom AST-based safe expression evaluator (`safe_eval_expr`) that parses the formula into an abstract syntax tree and only allows numeric literals, pre-approved variable names, and basic arithmetic operators (+, -, *, /, //, %, **). Any attempt to inject function calls, attribute access, imports, or other arbitrary Python code will be rejected with a ValueError. Input validation also ensures the formula is a non-empty string and any supplied variables map only strings to numbers.

## Secure Code
```python
from flask import Flask, request, jsonify
import ast
import operator

app = Flask(__name__)
sensor_limits = {}

# Safe mathematical expression evaluator
_ALLOWED_OPS = {
    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_expr(expr_str, variables=None):
    """Safely evaluate a mathematical expression containing only numbers,
    basic arithmetic operators, and optionally pre-defined variable names."""
    if variables is None:
        variables = {}

    def _eval_node(node):
        if isinstance(node, ast.Expression):
            return _eval_node(node.body)
        elif isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        elif isinstance(node, ast.Name) and node.id in variables:
            return variables[node.id]
        elif isinstance(node, ast.BinOp):
            op_type = type(node.op)
            if op_type not in _ALLOWED_OPS:
                raise ValueError(f"Unsupported operator: {op_type.__name__}")
            left = _eval_node(node.left)
            right = _eval_node(node.right)
            return _ALLOWED_OPS[op_type](left, right)
        elif isinstance(node, ast.UnaryOp):
            op_type = type(node.op)
            if op_type not in _ALLOWED_OPS:
                raise ValueError(f"Unsupported unary operator: {op_type.__name__}")
            operand = _eval_node(node.operand)
            return _ALLOWED_OPS[op_type](operand)
        else:
            raise ValueError(f"Unsupported expression element: {ast.dump(node)}")

    tree = ast.parse(expr_str.strip(), mode='eval')
    return _eval_node(tree)


@app.route('/iot/threshold', methods=['POST'])
def configure_threshold():
    dev_id = request.json.get('device_id')
    raw_expr = request.json.get('threshold_formula')

    if not isinstance(raw_expr, str) or not raw_expr.strip():
        return jsonify({'error': 'threshold_formula must be a non-empty string'}), 400

    # Optional: allow operators to reference named constants in the formula
    allowed_variables = request.json.get('variables', {})
    if not isinstance(allowed_variables, dict) or not all(
        isinstance(k, str) and isinstance(v, (int, float))
        for k, v in allowed_variables.items()
    ):
        return jsonify({'error': 'variables must be a dict of string->number'}), 400

    try:
        computed_limit = safe_eval_expr(raw_expr, variables=allowed_variables)
        sensor_limits[dev_id] = computed_limit
        return jsonify({'status': 'ok', 'limit': computed_limit})
    except Exception as e:
        return jsonify({'error': str(e)}), 400
```