# Command Injection via subprocess.shell=True

Language: Python
Severity: Critical
CWE: CWE-78

## Source
3

## Flow
3-6-7-8-11

## Sink
11

## Vulnerable Code
```python
import subprocess
import logging

def provision_iot_device(device_mac, firmware_url, config_params):
    logger = logging.getLogger('iot_provisioner')
    logger.info(f'Provisioning device: {device_mac}')
    flash_cmd = f"curl -s {firmware_url} | dd of=/dev/mtdblock0 bs=1M && "
    flash_cmd += f"echo '{config_params}' > /etc/iot/device.conf && "
    flash_cmd += f"ifconfig eth0 hw ether {device_mac}"
    try:
        result = subprocess.run(flash_cmd, shell=True, capture_output=True, timeout=120)
        if result.returncode == 0:
            logger.info(f'Device {device_mac} provisioned successfully')
            return {'status': 'success', 'device': device_mac, 'output': result.stdout.decode()}
        else:
            logger.error(f'Provisioning failed: {result.stderr.decode()}')
            return {'status': 'error', 'message': result.stderr.decode()}
    except subprocess.TimeoutExpired:
        logger.error('Provisioning timeout exceeded')
        return {'status': 'timeout', 'device': device_mac}
```

## Explanation

The function accepts untrusted input parameters (device_mac, firmware_url, config_params) and directly interpolates them into shell commands without any sanitization or validation. These tainted values flow through string concatenation operations to build the flash_cmd variable, which is then executed via subprocess.run() with shell=True, enabling command injection attacks.

## Remediation

The fix eliminates command injection by replacing the single shell=True subprocess call with multiple individual subprocess calls using argument lists (shell=False), strict input validation with regex patterns for MAC addresses, URLs, and config parameters, and writing configuration files using Python's native file I/O instead of shell echo redirection. Each operation is isolated and uses parameterized arguments that cannot be interpreted as shell commands.

## Secure Code
```python
import subprocess
import logging
import re
import shlex
from urllib.parse import urlparse


def validate_mac_address(mac):
    """Validate MAC address format strictly."""
    pattern = r'^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$'
    if not re.match(pattern, mac):
        raise ValueError(f'Invalid MAC address format: {mac}')
    return mac


def validate_firmware_url(url):
    """Validate firmware URL against allowed schemes and domains."""
    parsed = urlparse(url)
    if parsed.scheme not in ('http', 'https'):
        raise ValueError(f'Invalid URL scheme: {parsed.scheme}')
    if not parsed.hostname:
        raise ValueError('URL must have a valid hostname')
    if any(c in url for c in [';', '|', '&', '`', '$', '(', ')', '{', '}', '\n', '\r']):
        raise ValueError('URL contains invalid characters')
    return url


def validate_config_params(config_params):
    """Validate configuration parameters - allow only alphanumeric, common config characters."""
    pattern = r'^[a-zA-Z0-9_=.,:/\-\s]+$'
    if not re.match(pattern, config_params):
        raise ValueError('Config parameters contain invalid characters')
    return config_params


def provision_iot_device(device_mac, firmware_url, config_params):
    logger = logging.getLogger('iot_provisioner')
    logger.info(f'Provisioning device: {device_mac}')

    try:
        device_mac = validate_mac_address(device_mac)
        firmware_url = validate_firmware_url(firmware_url)
        config_params = validate_config_params(config_params)
    except ValueError as e:
        logger.error(f'Input validation failed: {e}')
        return {'status': 'error', 'message': f'Validation failed: {e}'}

    try:
        curl_result = subprocess.run(
            ['curl', '-s', '--max-time', '60', '-o', '/tmp/firmware.bin', firmware_url],
            capture_output=True, timeout=60
        )
        if curl_result.returncode != 0:
            logger.error(f'Firmware download failed: {curl_result.stderr.decode()}')
            return {'status': 'error', 'message': 'Firmware download failed'}

        dd_result = subprocess.run(
            ['dd', 'if=/tmp/firmware.bin', 'of=/dev/mtdblock0', 'bs=1M'],
            capture_output=True, timeout=60
        )
        if dd_result.returncode != 0:
            logger.error(f'Firmware flash failed: {dd_result.stderr.decode()}')
            return {'status': 'error', 'message': 'Firmware flash failed'}

        try:
            with open('/etc/iot/device.conf', 'w') as conf_file:
                conf_file.write(config_params)
        except IOError as e:
            logger.error(f'Config write failed: {e}')
            return {'status': 'error', 'message': 'Config write failed'}

        mac_result = subprocess.run(
            ['ip', 'link', 'set', 'dev', 'eth0', 'address', device_mac],
            capture_output=True, timeout=10
        )
        if mac_result.returncode != 0:
            logger.error(f'MAC address set failed: {mac_result.stderr.decode()}')
            return {'status': 'error', 'message': 'MAC address configuration failed'}

        logger.info(f'Device {device_mac} provisioned successfully')
        return {'status': 'success', 'device': device_mac}

    except subprocess.TimeoutExpired:
        logger.error('Provisioning timeout exceeded')
        return {'status': 'timeout', 'device': device_mac}
    except Exception as e:
        logger.error(f'Unexpected error during provisioning: {e}')
        return {'status': 'error', 'message': 'Unexpected provisioning error'}
```