# Command Injection via subprocess.shell=True

Language: Python
Severity: Critical
CWE: CWE-78

## Source
2

## Flow
2-3-4, 2-6

## Sink
4, 6

## Vulnerable Code
```python
import subprocess

def provision_iot_device(device_mac, firmware_url):
    validation_cmd = f"ping -c 2 {device_mac.replace(':', '-')}.local && curl -o /tmp/fw.bin {firmware_url}"
    result = subprocess.run(validation_cmd, shell=True, capture_output=True, text=True)
    if result.returncode == 0:
        flash_cmd = f"iot-flasher --device {device_mac} --firmware /tmp/fw.bin"
        flash_result = subprocess.run(flash_cmd, shell=True, capture_output=True)
        return {"status": "provisioned", "output": flash_result.stdout}
    return {"status": "failed", "error": result.stderr}
```

## Explanation

The function accepts untrusted input parameters (device_mac and firmware_url) and directly interpolates them into shell commands executed via subprocess.run() with shell=True. This allows attackers to inject arbitrary shell commands through specially crafted MAC addresses or URLs that break out of the intended command structure.

## Remediation

The fix eliminates command injection by using subprocess.run() with argument lists instead of shell=True, which prevents shell interpretation of special characters. Additionally, strict input validation ensures that device_mac matches a valid MAC address regex pattern and firmware_url is a properly formed HTTP/HTTPS URL, providing defense-in-depth. The chained shell command is split into separate subprocess calls, each with its own error handling.

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

def validate_mac_address(mac):
    """Validate that the input is a properly formatted MAC address."""
    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 that the firmware URL is a proper HTTP/HTTPS URL."""
    parsed = urlparse(url)
    if parsed.scheme not in ('http', 'https'):
        raise ValueError(f"Invalid firmware URL scheme: {url}")
    if not parsed.hostname:
        raise ValueError(f"Invalid firmware URL hostname: {url}")
    return url

def provision_iot_device(device_mac, firmware_url):
    # Validate inputs strictly
    device_mac = validate_mac_address(device_mac)
    firmware_url = validate_firmware_url(firmware_url)

    # Convert MAC to hostname format for ping
    device_hostname = device_mac.replace(':', '-') + '.local'

    # Run ping as a separate command without shell=True
    ping_result = subprocess.run(
        ['ping', '-c', '2', device_hostname],
        capture_output=True, text=True
    )
    if ping_result.returncode != 0:
        return {"status": "failed", "error": "Device not reachable"}

    # Run curl as a separate command without shell=True
    curl_result = subprocess.run(
        ['curl', '-o', '/tmp/fw.bin', '--', firmware_url],
        capture_output=True, text=True
    )
    if curl_result.returncode != 0:
        return {"status": "failed", "error": curl_result.stderr}

    # Run flash command without shell=True
    flash_result = subprocess.run(
        ['iot-flasher', '--device', device_mac, '--firmware', '/tmp/fw.bin'],
        capture_output=True, text=True
    )
    if flash_result.returncode == 0:
        return {"status": "provisioned", "output": flash_result.stdout}
    return {"status": "failed", "error": flash_result.stderr}
```