# Import Hijacking via sys.path Manipulation

Language: Python
Severity: Critical
CWE: CWE-427

## Source
20

## Flow
20-19-5-12

## Sink
5, 12

## Vulnerable Code
```python
import sys
import os

def load_cloud_storage_plugin(plugin_dir, plugin_name):
    if not os.path.isabs(plugin_dir):
        plugin_dir = os.path.join('/var/app/plugins', plugin_dir)
    sys.path.insert(0, plugin_dir)
    try:
        storage_module = __import__(plugin_name)
        return storage_module.CloudStorageAdapter()
    except ImportError as e:
        sys.path.insert(0, '/tmp/fallback_plugins')
        storage_module = __import__(plugin_name)
        return storage_module.CloudStorageAdapter()

def sync_iot_telemetry_data(device_id, custom_plugin_path):
    adapter = load_cloud_storage_plugin(custom_plugin_path, 'aws_s3_handler')
    telemetry = adapter.fetch_device_metrics(device_id)
    return adapter.upload_to_bucket(telemetry, f'iot-data/{device_id}')
```

## Explanation

The function accepts an untrusted `custom_plugin_path` parameter that is directly used to manipulate `sys.path` via `sys.path.insert()`, allowing an attacker to control Python's module search path. This enables import hijacking where malicious modules can be loaded via `__import__()`, leading to arbitrary code execution when the imported module is instantiated.

## Remediation

The fix eliminates sys.path manipulation entirely by using importlib.util.spec_from_file_location to load modules directly from validated file paths. It enforces an allowlist of permitted plugin names and constrains plugin directories to a specific base path with symlink-aware canonicalization to prevent directory traversal attacks. The fallback to /tmp/fallback_plugins has been removed entirely as it represented an additional untrusted path.

## Secure Code
```python
import os
import importlib
import importlib.util

ALLOWED_PLUGIN_BASE_DIR = '/var/app/plugins'
ALLOWED_PLUGIN_NAMES = {'aws_s3_handler', 'azure_blob_handler', 'gcs_handler'}


def _validate_plugin_dir(plugin_dir):
    """Validate that the plugin directory is within the allowed base directory."""
    if not os.path.isabs(plugin_dir):
        plugin_dir = os.path.join(ALLOWED_PLUGIN_BASE_DIR, plugin_dir)
    
    # Resolve to canonical path to prevent directory traversal
    resolved_dir = os.path.realpath(plugin_dir)
    allowed_base = os.path.realpath(ALLOWED_PLUGIN_BASE_DIR)
    
    if not resolved_dir.startswith(allowed_base + os.sep) and resolved_dir != allowed_base:
        raise ValueError(
            f"Plugin directory '{plugin_dir}' is outside the allowed base directory."
        )
    
    if not os.path.isdir(resolved_dir):
        raise FileNotFoundError(
            f"Plugin directory '{resolved_dir}' does not exist."
        )
    
    return resolved_dir


def _validate_plugin_name(plugin_name):
    """Validate that the plugin name is in the allowlist and is a safe identifier."""
    if not plugin_name.isidentifier():
        raise ValueError(f"Invalid plugin name: '{plugin_name}'")
    
    if plugin_name not in ALLOWED_PLUGIN_NAMES:
        raise ValueError(
            f"Plugin '{plugin_name}' is not in the list of allowed plugins."
        )
    
    return plugin_name


def load_cloud_storage_plugin(plugin_dir, plugin_name):
    """Securely load a cloud storage plugin from a validated directory."""
    validated_dir = _validate_plugin_dir(plugin_dir)
    validated_name = _validate_plugin_name(plugin_name)
    
    # Load the module directly from the specific file path without modifying sys.path
    module_path = os.path.join(validated_dir, f"{validated_name}.py")
    
    if not os.path.isfile(module_path):
        raise ImportError(
            f"Plugin module '{validated_name}' not found in '{validated_dir}'."
        )
    
    # Verify the file is within the validated directory (prevent symlink attacks)
    resolved_module_path = os.path.realpath(module_path)
    if not resolved_module_path.startswith(os.path.realpath(validated_dir) + os.sep):
        raise ValueError(
            f"Plugin file resolves outside the allowed directory (possible symlink attack)."
        )
    
    # Use importlib to load from a specific file without modifying sys.path
    spec = importlib.util.spec_from_file_location(validated_name, resolved_module_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot create module spec for '{validated_name}'.")
    
    storage_module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(storage_module)
    
    if not hasattr(storage_module, 'CloudStorageAdapter'):
        raise AttributeError(
            f"Plugin '{validated_name}' does not define 'CloudStorageAdapter'."
        )
    
    return storage_module.CloudStorageAdapter()


def sync_iot_telemetry_data(device_id, custom_plugin_path):
    """Sync IoT telemetry data using a validated cloud storage plugin."""
    # Validate device_id to prevent path injection
    if not device_id.replace('-', '').replace('_', '').isalnum():
        raise ValueError(f"Invalid device ID: '{device_id}'")
    
    adapter = load_cloud_storage_plugin(custom_plugin_path, 'aws_s3_handler')
    telemetry = adapter.fetch_device_metrics(device_id)
    return adapter.upload_to_bucket(telemetry, f'iot-data/{device_id}')
```