# Command Injection via subprocess.run(shell=True)

Language: Python
Severity: Critical
CWE: CWE-78

## Source
2, 3

## Flow
2-5-6, 3-8-9

## Sink
6, 9

## Vulnerable Code
```python
import subprocess

def provision_iot_device(device_mac, firmware_url):
    sanitized_mac = device_mac.replace(':', '')
    cmd = f"curl -s {firmware_url} | sha256sum && echo 'Provisioning device {sanitized_mac}'"
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    if result.returncode == 0:
        log_entry = f"iot-provision --mac={device_mac} --status=success"
        subprocess.run(log_entry, shell=True)
        return {"status": "provisioned", "device": device_mac, "hash": result.stdout.strip()}
    return {"status": "failed", "device": device_mac}
```

## Explanation

The function accepts unsanitized user input (device_mac and firmware_url) and directly interpolates them into shell commands executed via subprocess.run() with shell=True. While device_mac receives minimal sanitization (colon removal), both parameters can contain shell metacharacters that enable command injection. The firmware_url is completely unsanitized and directly embedded in a piped command sequence, and device_mac is used unsanitized in the logging subprocess call.

## Remediation

The fix eliminates all shell=True subprocess calls by: (1) replacing curl with Python's urllib for downloading firmware, (2) computing SHA256 using Python's hashlib instead of piping to sha256sum, (3) using subprocess with argument lists (no shell) for the external logging tool, and (4) adding strict input validation with regex for MAC addresses and URL scheme whitelisting.

## Secure Code
```python
import subprocess
import re
import hashlib
import logging
import urllib.request
import urllib.parse

logger = logging.getLogger(__name__)

MAC_PATTERN = re.compile(r'^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$')
ALLOWED_SCHEMES = {'http', 'https'}

def validate_mac_address(device_mac):
    if not MAC_PATTERN.match(device_mac):
        raise ValueError(f"Invalid MAC address format: {device_mac}")
    return device_mac

def validate_firmware_url(firmware_url):
    parsed = urllib.parse.urlparse(firmware_url)
    if parsed.scheme not in ALLOWED_SCHEMES:
        raise ValueError(f"Invalid URL scheme: {parsed.scheme}. Only HTTP/HTTPS allowed.")
    if not parsed.hostname:
        raise ValueError("Invalid URL: no hostname found.")
    return firmware_url

def provision_iot_device(device_mac, firmware_url):
    device_mac = validate_mac_address(device_mac)
    firmware_url = validate_firmware_url(firmware_url)

    sanitized_mac = device_mac.replace(':', '')

    try:
        req = urllib.request.Request(firmware_url)
        with urllib.request.urlopen(req, timeout=30) as response:
            firmware_data = response.read()
    except Exception as e:
        logger.error(f"Failed to download firmware: {e}")
        return {"status": "failed", "device": device_mac}

    firmware_hash = hashlib.sha256(firmware_data).hexdigest()

    logger.info(f"iot-provision mac={sanitized_mac} status=success hash={firmware_hash}")

    try:
        subprocess.run(
            ["iot-provision", "--mac", device_mac, "--status", "success"],
            capture_output=True,
            text=True,
            check=False
        )
    except FileNotFoundError:
        logger.warning("iot-provision binary not found, skipping external logging")

    return {"status": "provisioned", "device": device_mac, "hash": firmware_hash}
```