# Command Injection via subprocess.shell

Language: Python
Severity: Critical
CWE: CWE-78

## Source
3

## Flow
3-4, 3-6

## Sink
4, 6

## Vulnerable Code
```python
import subprocess

def rotate_iot_device_logs(device_id, log_format):
    archive_cmd = f"tar -czf /var/iot/logs/{device_id}.tar.gz --format={log_format} /var/iot/logs/{device_id}/"
    result = subprocess.run(archive_cmd, shell=True, capture_output=True, text=True)
    if result.returncode == 0:
        cleanup_cmd = f"find /var/iot/logs/{device_id}/ -type f -mtime +7 -exec rm {{}} \;"
        subprocess.run(cleanup_cmd, shell=True)
        return {"status": "success", "archive": f"{device_id}.tar.gz"}
    return {"status": "failed", "error": result.stderr}
```

## Explanation

The function uses unsanitized user input (device_id and log_format parameters) directly in shell commands via f-strings, then executes them with subprocess.run(shell=True). An attacker can inject arbitrary shell commands through either parameter to achieve remote code execution on the IoT platform.

## Remediation

The fix eliminates command injection by: (1) validating device_id against a strict whitelist regex pattern allowing only safe characters, (2) validating log_format against an explicit allowlist of known tar formats, and (3) replacing shell=True with argument lists passed directly to subprocess.run, which prevents shell interpretation of special characters.

## Secure Code
```python
import subprocess
import re
import os

def rotate_iot_device_logs(device_id, log_format):
    # Validate device_id: only allow alphanumeric characters, hyphens, and underscores
    if not re.match(r'^[a-zA-Z0-9_\-]+$', device_id):
        return {"status": "failed", "error": "Invalid device_id: only alphanumeric characters, hyphens, and underscores are allowed"}
    
    # Validate log_format: only allow known safe format values
    allowed_formats = ['gnu', 'oldgnu', 'pax', 'posix', 'ustar', 'v7']
    if log_format not in allowed_formats:
        return {"status": "failed", "error": f"Invalid log_format: must be one of {allowed_formats}"}
    
    # Validate paths exist
    log_dir = os.path.join('/var/iot/logs', device_id)
    archive_path = os.path.join('/var/iot/logs', f'{device_id}.tar.gz')
    
    if not os.path.isdir(log_dir):
        return {"status": "failed", "error": "Log directory does not exist"}
    
    # Use subprocess with argument list (no shell=True) to prevent command injection
    archive_cmd = [
        'tar', '-czf', archive_path,
        f'--format={log_format}',
        log_dir
    ]
    result = subprocess.run(archive_cmd, capture_output=True, text=True)
    
    if result.returncode == 0:
        cleanup_cmd = [
            'find', log_dir,
            '-type', 'f',
            '-mtime', '+7',
            '-exec', 'rm', '{}', ';'
        ]
        subprocess.run(cleanup_cmd)
        return {"status": "success", "archive": f"{device_id}.tar.gz"}
    return {"status": "failed", "error": result.stderr}
```