Patched code sample

import subprocess
import shlex
from typing import List

# A predefined allowlist of safe commands. This is the core of the fix,
# ensuring only expected executables can be run. In a real system,
# this list would be carefully curated to include only necessary and
# non-dangerous commands.
ALLOWED_COMMANDS = {
    "ls",
    "echo",
    "cat",
    # Assuming 'agent_tool' is a custom, safe executable for the system.
    "agent_tool",
}

def parse_mcp_command(command_string: str):
    """
    Parses and executes a command string only if the command is in a predefined
    allowlist, representing the fix for the arbitrary code execution vulnerability.

    The vulnerable version of this function would have directly passed the parsed
    arguments to subprocess.run() without validation.
    """
    if not isinstance(command_string, str) or not command_string.strip():
        raise ValueError("Command string cannot be empty.")

    # Use shlex.split for safer parsing of shell-like command strings,
    # which helps prevent some forms of injection.
    try:
        args: List[str] = shlex.split(command_string)
    except ValueError as e:
        raise ValueError(f"Error parsing command string: {e}")

    if not args:
        raise ValueError("Parsed command is empty.")

    command = args[0]

    # --- VULNERABILITY FIX ---
    # The core of the fix is to validate the parsed command against a strict allowlist
    # before passing it to any execution context like subprocess.
    if command not in ALLOWED_COMMANDS:
        # If the command is not in the allowlist, raise a PermissionError
        # and refuse to execute. This prevents arbitrary commands like
        # 'bash', 'python', 'sh', etc., from being run.
        raise PermissionError(
            f"Execution of command '{command}' is denied. It is not in the allowlist."
        )
    # --- END OF FIX ---

    # If the command is validated, it is now safer to execute.
    # The `check=True` argument ensures that if the command itself fails
    # (e.g., returns a non-zero exit code), it will raise a CalledProcessError.
    try:
        subprocess.run(args, check=True, capture_output=True, text=True)
    except FileNotFoundError:
        # This handles the case where the allowed command does not exist on the system.
        raise FileNotFoundError(f"Allowed command '{command}' not found on the system.")
    except subprocess.CalledProcessError as e:
        # This handles errors from the executed command itself.
        raise RuntimeError(f"Command '{command}' failed with error: {e.stderr}")