# Arbitrary Module Execution via `importlib.import_module` on User-Controlled Name

Language: Python
Severity: Critical
CWE: CWE-94

## Source
8-9

## Flow
8-10-11

## Sink
11

## Vulnerable Code
```python
import importlib
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/cloud/invoke', methods=['POST'])
def invoke_cloud_service():
    provider = request.json.get('provider', 'aws')
    action = request.json.get('action', 'list_instances')
    service_module = f"cloud_adapters.{provider}.{action}"
    try:
        mod = importlib.import_module(service_module)
        result = mod.execute(request.json.get('params', {}))
        return jsonify({'status': 'success', 'data': result})
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500
```

## Explanation

The application constructs a module path using unsanitized user input from 'provider' and 'action' JSON fields, then passes it directly to importlib.import_module(). An attacker can inject malicious module paths (e.g., '__main__', 'os', or arbitrary installed packages) to execute unintended code or access sensitive functionality.

## Remediation

The fix implements a strict allowlist of valid provider and action combinations, rejecting any input that is not explicitly permitted before it reaches `importlib.import_module()`. Additionally, it validates that both values are valid Python identifiers to prevent path traversal or null-byte injection, and it avoids leaking internal error details to the client.

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

app = Flask(__name__)

# Allowlist of valid provider and action combinations
ALLOWED_PROVIDERS = {'aws', 'azure', 'gcp'}
ALLOWED_ACTIONS = {
    'aws': {'list_instances', 'start_instance', 'stop_instance', 'list_buckets'},
    'azure': {'list_vms', 'start_vm', 'stop_vm', 'list_storage'},
    'gcp': {'list_instances', 'start_instance', 'stop_instance', 'list_buckets'}
}


def validate_provider_action(provider, action):
    """Validate that provider and action are in the allowlist and contain only safe characters."""
    if provider not in ALLOWED_PROVIDERS:
        return False, f"Invalid provider: '{provider}'. Allowed: {sorted(ALLOWED_PROVIDERS)}"
    if provider not in ALLOWED_ACTIONS or action not in ALLOWED_ACTIONS[provider]:
        return False, f"Invalid action: '{action}' for provider '{provider}'. Allowed: {sorted(ALLOWED_ACTIONS.get(provider, set()))}"
    # Additional safety: ensure no path traversal or special characters
    if not provider.isidentifier() or not action.isidentifier():
        return False, "Provider and action must be valid Python identifiers"
    return True, None


@app.route('/api/cloud/invoke', methods=['POST'])
def invoke_cloud_service():
    if not request.json:
        return jsonify({'status': 'error', 'message': 'Request body must be JSON'}), 400

    provider = request.json.get('provider', 'aws')
    action = request.json.get('action', 'list_instances')

    # Validate against allowlist before constructing module path
    is_valid, error_message = validate_provider_action(provider, action)
    if not is_valid:
        return jsonify({'status': 'error', 'message': error_message}), 400

    service_module = f"cloud_adapters.{provider}.{action}"
    try:
        mod = importlib.import_module(service_module)
        result = mod.execute(request.json.get('params', {}))
        return jsonify({'status': 'success', 'data': result})
    except Exception as e:
        return jsonify({'status': 'error', 'message': 'Internal error executing cloud action'}), 500
```