# Code Injection via exec() on Untrusted Configuration Data

Language: Python
Severity: Critical
CWE: CWE-94

## Source
3

## Flow
3-4-8-9-10, 3-4-11-12-13

## Sink
10, 13

## Vulnerable Code
```python
import json
import os

def apply_iot_device_policy(device_id, policy_json):
    policy_data = json.loads(policy_json)
    device_config = {
        'device_id': device_id,
        'max_temp': 85,
        'alert_threshold': 75,
        'sampling_rate': 1000
    }
    if 'custom_rules' in policy_data:
        rules_code = policy_data['custom_rules']
        exec(rules_code, {'config': device_config, 'os': os})
    if 'override_params' in policy_data:
        for param, value_expr in policy_data['override_params'].items():
            device_config[param] = eval(value_expr)
    return device_config
```

## Explanation

The function accepts untrusted JSON input via policy_json parameter, parses it, and directly passes user-controlled data to exec() and eval() functions without any validation or sanitization. This allows an attacker to inject arbitrary Python code through the 'custom_rules' field (executed via exec()) or 'override_params' values (executed via eval()), enabling complete system compromise.

## Remediation

The fix completely removes exec() and eval() calls, replacing them with a safe declarative rule engine that only allows whitelisted parameters and numeric values. Custom rules are now structured as JSON objects with predefined actions and conditions instead of arbitrary code strings, and override parameters are validated against an allowlist with safe numeric parsing.

## Secure Code
```python
import json

ALLOWED_PARAMS = {'max_temp', 'alert_threshold', 'sampling_rate', 'min_temp', 'polling_interval'}
ALLOWED_RULE_ACTIONS = {'set_max_temp', 'set_alert_threshold', 'set_sampling_rate', 'set_min_temp', 'set_polling_interval'}

def validate_numeric_value(value):
    """Safely parse a numeric value from a string."""
    try:
        if '.' in str(value):
            return float(value)
        return int(value)
    except (ValueError, TypeError):
        raise ValueError(f"Invalid numeric value: {value}")

def apply_rule(device_config, rule):
    """Apply a single validated rule to device configuration."""
    action = rule.get('action')
    if action not in ALLOWED_RULE_ACTIONS:
        raise ValueError(f"Disallowed rule action: {action}")
    
    param_name = action.replace('set_', '', 1)
    if param_name not in ALLOWED_PARAMS:
        raise ValueError(f"Invalid parameter in rule: {param_name}")
    
    value = validate_numeric_value(rule.get('value'))
    
    condition = rule.get('condition')
    if condition:
        compare_param = condition.get('param')
        if compare_param not in ALLOWED_PARAMS and compare_param not in device_config:
            raise ValueError(f"Invalid condition parameter: {compare_param}")
        operator = condition.get('operator')
        threshold = validate_numeric_value(condition.get('value'))
        current_value = device_config.get(compare_param, 0)
        
        if operator == 'greater_than' and not (current_value > threshold):
            return
        elif operator == 'less_than' and not (current_value < threshold):
            return
        elif operator == 'equals' and not (current_value == threshold):
            return
        elif operator not in ('greater_than', 'less_than', 'equals'):
            raise ValueError(f"Invalid operator: {operator}")
    
    device_config[param_name] = value

def apply_iot_device_policy(device_id, policy_json):
    policy_data = json.loads(policy_json)
    device_config = {
        'device_id': device_id,
        'max_temp': 85,
        'alert_threshold': 75,
        'sampling_rate': 1000
    }
    
    if 'custom_rules' in policy_data:
        rules = policy_data['custom_rules']
        if not isinstance(rules, list):
            raise ValueError("custom_rules must be a list of rule objects")
        for rule in rules:
            if not isinstance(rule, dict):
                raise ValueError("Each rule must be a dictionary")
            apply_rule(device_config, rule)
    
    if 'override_params' in policy_data:
        override_params = policy_data['override_params']
        if not isinstance(override_params, dict):
            raise ValueError("override_params must be a dictionary")
        for param, value in override_params.items():
            if param not in ALLOWED_PARAMS:
                raise ValueError(f"Disallowed parameter override: {param}")
            device_config[param] = validate_numeric_value(value)
    
    return device_config
```