# Command Injection

Language: Python
Severity: Critical
CWE: CWE-78

## Source
9

## Flow
9-11

## Sink
11

## Vulnerable Code
```python
import subprocess
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/api/iot/ping', methods=['POST'])
def device_health_check():
    dev_addr = request.json.get('device_ip')
    if not dev_addr:
        return jsonify({'err': 'missing device_ip'}), 400
    latency_cmd = f'ping -c 3 {dev_addr}'
    proc_out = subprocess.check_output(latency_cmd, shell=True, stderr=subprocess.STDOUT)
    return jsonify({'result': proc_out.decode('utf-8')}), 200
```

## Explanation

The user-supplied 'device_ip' value is retrieved from the POST JSON body on line 6 without any sanitization or validation. It is directly interpolated into a shell command string via f-string on line 9, then executed with shell=True on line 10 via subprocess.check_output(), allowing an attacker to inject arbitrary OS commands by appending shell metacharacters (e.g., semicolons, pipes, backticks) to the IP value.

## Remediation

The fix eliminates command injection by (1) strictly validating that the supplied device_ip matches an IPv4 or IPv6 address pattern via regex before use, rejecting any input containing shell metacharacters, and (2) passing the command as a list of arguments to subprocess.check_output without shell=True, so the OS never interprets the input through a shell. Together these two layers ensure no attacker-controlled string can reach a shell for interpretation.

## Secure Code
```python
import subprocess
import re
from flask import Flask, request, jsonify

app = Flask(__name__)

IPV4_REGEX = re.compile(
    r'^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)$'
)
IPV6_REGEX = re.compile(
    r'^([0-9a-fA-F]{0,4}:){2,7}[0-9a-fA-F]{0,4}$'
)


def is_valid_ip(addr: str) -> bool:
    """Return True only if addr is a well-formed IPv4 or IPv6 address."""
    return bool(IPV4_REGEX.match(addr) or IPV6_REGEX.match(addr))


@app.route('/api/iot/ping', methods=['POST'])
def device_health_check():
    dev_addr = request.json.get('device_ip')
    if not dev_addr:
        return jsonify({'err': 'missing device_ip'}), 400

    dev_addr = dev_addr.strip()
    if not is_valid_ip(dev_addr):
        return jsonify({'err': 'invalid device_ip format'}), 400

    proc_out = subprocess.check_output(
        ['ping', '-c', '3', dev_addr],
        stderr=subprocess.STDOUT
    )
    return jsonify({'result': proc_out.decode('utf-8')}), 200
```