# Command Injection via Shell Metacharacter Expansion

Language: Python
Severity: Critical
CWE: CWE-78

## Source
6-7

## Flow
6-7-10-11

## Sink
11

## Vulnerable Code
```python
import subprocess
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/api/v1/network/latency', methods=['GET'])
def measure_latency():
    target_node = request.args.get('node', '')
    pkt_count = request.args.get('packets', '4')
    if not target_node:
        return jsonify({'error': 'node parameter required'}), 400
    diag_cmd = f"ping -c {pkt_count} {target_node}"
    result = subprocess.run(diag_cmd, shell=True, capture_output=True, text=True, timeout=15)
    return jsonify({'stdout': result.stdout, 'stderr': result.stderr, 'exit_code': result.returncode})
```

## Explanation

Both 'target_node' and 'pkt_count' are taken directly from user-controlled HTTP query parameters without any sanitization or validation. They are interpolated into a shell command string via f-string on line 10, then executed with shell=True on line 11, which instructs the OS shell to interpret metacharacters — enabling arbitrary command injection.

## Remediation

The fix eliminates the command injection vulnerability through three changes: (1) strict input validation using allowlist regex patterns ensures target_node can only be a valid hostname or IP address and pkt_count must be a bounded integer, rejecting any shell metacharacters; (2) the command is passed as a list of arguments instead of an f-string, and (3) shell=True is removed (defaulting to shell=False), so the OS never interprets the arguments through a shell, preventing metacharacter expansion even if validation were bypassed.

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

app = Flask(__name__)


def is_valid_hostname_or_ip(value):
    """Validate that the value is a legitimate hostname or IP address."""
    # Match IPv4 addresses
    ipv4_pattern = r'^(\d{1,3}\.){3}\d{1,3}$'
    # Match IPv6 addresses (simplified)
    ipv6_pattern = r'^[0-9a-fA-F:]+$'
    # Match valid hostnames (RFC 1123): alphanumeric, hyphens, dots
    hostname_pattern = r'^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*$'

    if re.match(ipv4_pattern, value):
        # Additionally validate each octet is 0-255
        parts = value.split('.')
        return all(0 <= int(p) <= 255 for p in parts)
    if re.match(ipv6_pattern, value) and len(value) <= 45:
        return True
    if re.match(hostname_pattern, value) and len(value) <= 253:
        return True
    return False


def is_valid_packet_count(value):
    """Validate that the packet count is a reasonable positive integer."""
    try:
        count = int(value)
        return 1 <= count <= 100
    except (ValueError, TypeError):
        return False


@app.route('/api/v1/network/latency', methods=['GET'])
def measure_latency():
    target_node = request.args.get('node', '')
    pkt_count = request.args.get('packets', '4')

    if not target_node:
        return jsonify({'error': 'node parameter required'}), 400

    if not is_valid_hostname_or_ip(target_node):
        return jsonify({'error': 'Invalid node: must be a valid hostname or IP address'}), 400

    if not is_valid_packet_count(pkt_count):
        return jsonify({'error': 'Invalid packets: must be an integer between 1 and 100'}), 400

    # Use argument list (shell=False) to prevent shell metacharacter interpretation
    result = subprocess.run(
        ['ping', '-c', str(int(pkt_count)), target_node],
        capture_output=True, text=True, timeout=15
    )
    return jsonify({'stdout': result.stdout, 'stderr': result.stderr, 'exit_code': result.returncode})
```