# Argument Injection via subprocess.run() with Untrusted Option Prefixes

Language: Python
Severity: Critical
CWE: CWE-88

## Source
13-14

## Flow
14->5->7->9

## Sink
9

## Vulnerable Code
```python
import subprocess

def execute_cloud_backup(bucket_name, sync_flags):
    aws_region = "us-east-1"
    cmd = ["aws", "s3", "sync", "/data/backups", f"s3://{bucket_name}"]
    if sync_flags:
        cmd.extend(sync_flags.split())
    cmd.extend(["--region", aws_region])
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout

user_bucket = request.form.get("bucket")
user_options = request.form.get("options")
execute_cloud_backup(user_bucket, user_options)
```

## Explanation

Untrusted user input from 'user_options' is passed to execute_cloud_backup() and split directly into command arguments via cmd.extend(sync_flags.split()). This allows attackers to inject arbitrary AWS CLI flags like '--endpoint-url' to exfiltrate data or '--profile' to use unauthorized credentials, bypassing intended command structure.

## Remediation

The fix implements strict allowlist validation for both the bucket name and sync flags. Only pre-approved, safe AWS CLI flags are permitted through the ALLOWED_FLAGS set, and flags that accept values are further constrained to known-safe options via ALLOWED_FLAG_VALUES. This prevents injection of dangerous flags like '--endpoint-url', '--profile', or '--include' that could be used to exfiltrate data or escalate privileges.

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

# Allowlist of safe AWS S3 sync flags that users can specify
ALLOWED_FLAGS = {
    "--delete",
    "--exact-timestamps",
    "--no-progress",
    "--only-show-errors",
    "--quiet",
    "--dryrun",
    "--no-follow-symlinks",
    "--sse",
    "--storage-class",
    "--acl"
}

# Allowed values for flags that accept parameters
ALLOWED_FLAG_VALUES = {
    "--storage-class": ["STANDARD", "REDUCED_REDUNDANCY", "STANDARD_IA", "ONEZONE_IA", "INTELLIGENT_TIERING", "GLACIER", "DEEP_ARCHIVE"],
    "--sse": ["AES256", "aws:kms"],
    "--acl": ["private", "public-read", "public-read-write", "authenticated-read", "bucket-owner-read", "bucket-owner-full-control"]
}

# Allowlist pattern for bucket names (alphanumeric, hyphens, dots)
BUCKET_NAME_PATTERN = re.compile(r'^[a-z0-9][a-z0-9.\-]{1,61}[a-z0-9]$')


def validate_bucket_name(bucket_name):
    if not bucket_name or not BUCKET_NAME_PATTERN.match(bucket_name):
        raise ValueError(f"Invalid bucket name: {bucket_name}")
    return bucket_name


def validate_sync_flags(sync_flags):
    """Parse and validate sync flags against an allowlist."""
    if not sync_flags:
        return []

    validated_args = []
    tokens = sync_flags.split()
    i = 0
    while i < len(tokens):
        token = tokens[i]
        if token not in ALLOWED_FLAGS:
            raise ValueError(f"Disallowed flag: {token}")
        
        if token in ALLOWED_FLAG_VALUES:
            # This flag requires a value parameter
            if i + 1 >= len(tokens):
                raise ValueError(f"Flag {token} requires a value")
            value = tokens[i + 1]
            if value not in ALLOWED_FLAG_VALUES[token]:
                raise ValueError(f"Disallowed value '{value}' for flag {token}")
            validated_args.extend([token, value])
            i += 2
        else:
            # Boolean flag, no value needed
            validated_args.append(token)
            i += 1

    return validated_args


def execute_cloud_backup(bucket_name, sync_flags):
    aws_region = "us-east-1"
    
    validated_bucket = validate_bucket_name(bucket_name)
    validated_flags = validate_sync_flags(sync_flags)
    
    cmd = ["aws", "s3", "sync", "/data/backups", f"s3://{validated_bucket}"]
    if validated_flags:
        cmd.extend(validated_flags)
    cmd.extend(["--region", aws_region])
    
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout


user_bucket = request.form.get("bucket")
user_options = request.form.get("options")
try:
    execute_cloud_backup(user_bucket, user_options)
except ValueError as e:
    # Handle invalid input gracefully
    abort(400, description=str(e))
```