# Unsafe Dynamic Import via importlib.import_module()

Language: Python
Severity: Critical
CWE: CWE-94

## Source
8-9

## Flow
8-9-10-11

## Sink
11

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

app = Flask(__name__)

@app.route('/api/cloud/execute', methods=['POST'])
def execute_cloud_provider_action():
    provider = request.json.get('cloud_provider', 'aws')
    action = request.json.get('action', 'list_instances')
    module_path = f"cloud_integrations.{provider}.{action}"
    try:
        cloud_module = importlib.import_module(module_path)
        result = cloud_module.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 dynamically constructs a module path using unsanitized user input from 'cloud_provider' and 'action' parameters, then passes it to importlib.import_module(). An attacker can manipulate these inputs to import arbitrary Python modules, leading to arbitrary code execution through module imports like os, subprocess, or any installed package.

## Remediation

The fix introduces strict allowlists for both cloud providers and their associated actions, validating user input against these predefined sets before any dynamic import occurs. This prevents attackers from injecting arbitrary module paths since only explicitly permitted provider/action combinations can reach the importlib.import_module() call. Additionally, the error response no longer exposes internal exception details to prevent information leakage.

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

app = Flask(__name__)

# Allowlist of valid cloud providers and actions
ALLOWED_PROVIDERS = {'aws', 'azure', 'gcp'}
ALLOWED_ACTIONS = {
    'aws': {'list_instances', 'start_instance', 'stop_instance', 'describe_vpcs', 'list_buckets'},
    'azure': {'list_vms', 'start_vm', 'stop_vm', 'list_resource_groups', 'list_storage_accounts'},
    'gcp': {'list_instances', 'start_instance', 'stop_instance', 'list_buckets', 'list_networks'}
}

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

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

    # Validate provider against allowlist
    if provider not in ALLOWED_PROVIDERS:
        return jsonify({'status': 'error', 'message': f'Invalid cloud provider. Allowed: {sorted(ALLOWED_PROVIDERS)}'}), 400

    # Validate action against provider-specific allowlist
    if action not in ALLOWED_ACTIONS.get(provider, set()):
        return jsonify({'status': 'error', 'message': f'Invalid action for provider {provider}. Allowed: {sorted(ALLOWED_ACTIONS[provider])}'}), 400

    module_path = f"cloud_integrations.{provider}.{action}"
    try:
        cloud_module = importlib.import_module(module_path)
        result = cloud_module.execute(request.json.get('params', {}))
        return jsonify({'status': 'success', 'data': result})
    except Exception as e:
        return jsonify({'status': 'error', 'message': 'An internal error occurred while executing the action'}), 500
```