# Command Injection via Untrusted Executable Path in subprocess.Popen()

Language: Python
Severity: Critical
CWE: CWE-78

## Source
3

## Flow
3-5-7

## Sink
7

## Vulnerable Code
```python
import subprocess
import os

def deploy_ml_model(model_name, deployment_env, custom_runtime):
    deployment_path = f"/opt/ml/deployments/{deployment_env}"
    os.makedirs(deployment_path, exist_ok=True)
    runtime_binary = custom_runtime if custom_runtime else "/usr/bin/python3"
    deploy_cmd = [runtime_binary, "-m", "mlflow.models.serve", "--model-uri", model_name, "--port", "8080"]
    process = subprocess.Popen(deploy_cmd, cwd=deployment_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = process.communicate(timeout=30)
    if process.returncode == 0:
        return {"status": "deployed", "output": stdout.decode()}
    return {"status": "failed", "error": stderr.decode()}
```

## Explanation

The custom_runtime parameter is untrusted user input that is directly used as the first element of the command array in subprocess.Popen() without validation. An attacker can provide a malicious executable path or inject command arguments by exploiting path traversal or symbolic links, leading to arbitrary command execution on the server.

## Remediation

The fix replaces direct use of the user-supplied runtime path with an allowlist-based lookup, where users specify a runtime key (e.g., 'python3.9') that maps to a pre-approved absolute path. Additionally, the resolved path is validated to ensure it exists as a regular file and is executable, and symlink resolution via os.path.realpath() prevents path traversal attacks. Input validation is also added for model_name and deployment_env parameters.

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

ALLOWED_RUNTIMES = {
    "python3": "/usr/bin/python3",
    "python3.8": "/usr/bin/python3.8",
    "python3.9": "/usr/bin/python3.9",
    "python3.10": "/usr/bin/python3.10",
    "python3.11": "/usr/bin/python3.11",
    "conda-python": "/opt/conda/bin/python",
}

ALLOWED_ENVIRONMENTS = re.compile(r'^[a-zA-Z0-9_\-]+$')
ALLOWED_MODEL_NAMES = re.compile(r'^[a-zA-Z0-9_\-/.]+$')


def deploy_ml_model(model_name, deployment_env, custom_runtime):
    # Validate deployment_env to prevent path traversal
    if not deployment_env or not ALLOWED_ENVIRONMENTS.match(deployment_env):
        raise ValueError(f"Invalid deployment environment: {deployment_env}")

    # Validate model_name to prevent injection
    if not model_name or not ALLOWED_MODEL_NAMES.match(model_name):
        raise ValueError(f"Invalid model name: {model_name}")

    deployment_path = f"/opt/ml/deployments/{deployment_env}"
    os.makedirs(deployment_path, exist_ok=True)

    # Validate custom_runtime against an allowlist
    if custom_runtime:
        if custom_runtime not in ALLOWED_RUNTIMES:
            raise ValueError(
                f"Invalid runtime '{custom_runtime}'. "
                f"Allowed runtimes: {list(ALLOWED_RUNTIMES.keys())}"
            )
        runtime_binary = ALLOWED_RUNTIMES[custom_runtime]
    else:
        runtime_binary = ALLOWED_RUNTIMES["python3"]

    # Verify the resolved runtime binary exists and is a regular file
    resolved_path = os.path.realpath(runtime_binary)
    if not os.path.isfile(resolved_path):
        raise FileNotFoundError(f"Runtime binary not found: {runtime_binary}")
    if not os.access(resolved_path, os.X_OK):
        raise PermissionError(f"Runtime binary is not executable: {runtime_binary}")

    deploy_cmd = [
        resolved_path, "-m", "mlflow.models.serve",
        "--model-uri", model_name,
        "--port", "8080"
    ]

    process = subprocess.Popen(
        deploy_cmd,
        cwd=deployment_path,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    stdout, stderr = process.communicate(timeout=30)

    if process.returncode == 0:
        return {"status": "deployed", "output": stdout.decode()}
    return {"status": "failed", "error": stderr.decode()}
```