# Code Injection via pandas.eval()

Language: Python
Severity: Critical
CWE: CWE-94

## Source
9

## Flow
9-10

## Sink
10

## Vulnerable Code
```python
import pandas as pd
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/iot/sensor-analytics', methods=['POST'])
def compute_sensor_metrics():
    device_data = pd.DataFrame(request.json.get('sensor_readings'))
    metric_formula = request.json.get('calculation_rule')
    aggregated_result = device_data.eval(metric_formula)
    return jsonify({'computed_metrics': aggregated_result.tolist(), 'formula_applied': metric_formula})
```

## Explanation

The application accepts a user-controlled 'calculation_rule' formula from the JSON request and directly passes it to pandas DataFrame.eval() without any validation or sanitization. The eval() method executes arbitrary Python expressions, allowing attackers to inject malicious code that will be executed on the server.

## Remediation

The fix adds a comprehensive formula validation function that whitelists only safe arithmetic operations, known DataFrame column names, numeric literals, and a set of allowed mathematical functions. It blocks dangerous patterns like '__import__', 'os', 'exec', and other code injection vectors, and limits formula length to prevent abuse. Input validation is also added for the request payload structure.

## Secure Code
```python
import pandas as pd
import re
from flask import Flask, request, jsonify

app = Flask(__name__)

# Whitelist of allowed operations and patterns for sensor metric calculations
ALLOWED_COLUMN_PATTERN = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
ALLOWED_OPERATORS = {'+', '-', '*', '/', '(', ')', '.', ' '}
ALLOWED_FUNCTIONS = {'abs', 'sqrt', 'log', 'exp', 'mean', 'std', 'min', 'max', 'sum'}

def validate_formula(formula, valid_columns):
    """Validate that the formula only contains safe operations on known columns."""
    if not isinstance(formula, str) or len(formula) > 200:
        return False
    
    # Block dangerous patterns
    dangerous_patterns = [
        '__', 'import', 'eval', 'exec', 'compile', 'globals', 'locals',
        'getattr', 'setattr', 'delattr', 'open', 'file', 'input',
        'breakpoint', 'system', 'popen', 'subprocess', 'os.', 'sys.',
        'builtins', 'class', 'mro', 'subclasses'
    ]
    formula_lower = formula.lower()
    for pattern in dangerous_patterns:
        if pattern in formula_lower:
            return False
    
    # Tokenize and validate each token
    tokens = re.findall(r'[a-zA-Z_][a-zA-Z0-9_]*|[^a-zA-Z0-9_\s]+|\d+\.?\d*', formula)
    for token in tokens:
        if re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', token):
            # Token is an identifier - must be a known column or allowed function
            if token not in valid_columns and token not in ALLOWED_FUNCTIONS:
                return False
        elif re.match(r'^\d+\.?\d*$', token):
            # Token is a number - allowed
            continue
        else:
            # Token is an operator - check each character
            for char in token:
                if char not in ALLOWED_OPERATORS:
                    return False
    
    return True

@app.route('/api/iot/sensor-analytics', methods=['POST'])
def compute_sensor_metrics():
    sensor_readings = request.json.get('sensor_readings')
    if not sensor_readings or not isinstance(sensor_readings, list):
        return jsonify({'error': 'Invalid sensor readings provided'}), 400
    
    device_data = pd.DataFrame(sensor_readings)
    metric_formula = request.json.get('calculation_rule')
    
    if not metric_formula or not isinstance(metric_formula, str):
        return jsonify({'error': 'Invalid calculation rule'}), 400
    
    valid_columns = set(device_data.columns.tolist())
    
    if not validate_formula(metric_formula, valid_columns):
        return jsonify({'error': 'Invalid or disallowed formula. Only arithmetic operations on known sensor columns are permitted.'}), 400
    
    try:
        aggregated_result = device_data.eval(metric_formula)
        return jsonify({'computed_metrics': aggregated_result.tolist(), 'formula_applied': metric_formula})
    except Exception as e:
        return jsonify({'error': f'Formula evaluation failed: {str(e)}'}), 400
```