# Import Hijacking via sys.path Shadowing

Language: Python
Severity: Critical
CWE: CWE-427

## Source
4-5

## Flow
4-5-6

## Sink
6

## Vulnerable Code
```python
import sys
import os
def load_cloud_credentials(bucket_name, plugin_dir):
    user_plugin_path = os.path.join('/var/app/plugins', plugin_dir)
    sys.path.insert(0, user_plugin_path)
    import boto3
    s3_client = boto3.client('s3', aws_access_key_id=os.getenv('AWS_KEY'), aws_secret_access_key=os.getenv('AWS_SECRET'))
    response = s3_client.list_objects_v2(Bucket=bucket_name)
    return response['Contents']
def initialize_storage(tenant_id, custom_plugin):
    bucket = f'tenant-{tenant_id}-storage'
    return load_cloud_credentials(bucket, custom_plugin)
```

## Explanation

The code inserts an untrusted user-controlled plugin directory at the beginning of sys.path before importing boto3. A malicious tenant can place a fake boto3.py module in their plugin directory to shadow the legitimate library, intercepting AWS credentials passed during client initialization.

## Remediation

The fix imports boto3 at module level before any user-controlled path manipulation can occur, completely eliminating the possibility of import shadowing. Additionally, sys.path is never modified with user-controlled directories; instead, plugins are loaded explicitly using importlib.util.spec_from_file_location for isolated module loading. A validation layer checks plugin directories for path traversal and scans for files that would shadow critical libraries.

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

# Import boto3 at module level BEFORE any path manipulation
import boto3

ALLOWED_PLUGIN_PREFIX = '/var/app/plugins'
ALLOWED_PLUGIN_MODULES = set()  # Whitelist of allowed plugin module names
BLOCKED_MODULE_NAMES = {'boto3', 'botocore', 'os', 'sys', 'subprocess', 'shutil', 'socket', 'http', 'urllib'}


def _validate_plugin_dir(plugin_dir):
    """Validate and sanitize the plugin directory path."""
    # Prevent path traversal
    sanitized = os.path.normpath(plugin_dir)
    if '..' in sanitized or sanitized.startswith('/'):
        raise ValueError(f"Invalid plugin directory: {plugin_dir}")
    
    full_path = os.path.realpath(os.path.join(ALLOWED_PLUGIN_PREFIX, sanitized))
    
    # Ensure resolved path is still under the allowed prefix
    if not full_path.startswith(os.path.realpath(ALLOWED_PLUGIN_PREFIX)):
        raise ValueError(f"Plugin directory escapes allowed prefix: {plugin_dir}")
    
    if not os.path.isdir(full_path):
        raise ValueError(f"Plugin directory does not exist: {full_path}")
    
    return full_path


def _check_no_shadowing(plugin_path):
    """Ensure the plugin directory does not contain modules that shadow critical libraries."""
    if not os.path.isdir(plugin_path):
        return
    for entry in os.listdir(plugin_path):
        module_name = entry.replace('.py', '').replace('.pyc', '')
        if os.path.isdir(os.path.join(plugin_path, entry)):
            module_name = entry
        if module_name in BLOCKED_MODULE_NAMES:
            raise SecurityError(
                f"Plugin directory contains blocked module name: {module_name}"
            )


class SecurityError(Exception):
    """Raised when a security violation is detected."""
    pass


def load_plugin_safely(plugin_path, module_name):
    """Load a specific plugin module safely without modifying sys.path."""
    if module_name in BLOCKED_MODULE_NAMES:
        raise SecurityError(f"Cannot load blocked module name: {module_name}")
    
    module_file = os.path.join(plugin_path, f"{module_name}.py")
    if not os.path.isfile(module_file):
        raise FileNotFoundError(f"Plugin module not found: {module_file}")
    
    spec = importlib.util.spec_from_file_location(module_name, module_file)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def load_cloud_credentials(bucket_name, plugin_dir):
    """Load cloud credentials and list bucket contents.
    
    boto3 is imported at module level to prevent shadowing.
    Plugin directories are validated and never prepended to sys.path.
    """
    # Validate plugin directory
    validated_plugin_path = _validate_plugin_dir(plugin_dir)
    
    # Check for module shadowing attempts
    _check_no_shadowing(validated_plugin_path)
    
    # Use the already-imported boto3 (imported at module level)
    s3_client = boto3.client(
        's3',
        aws_access_key_id=os.getenv('AWS_KEY'),
        aws_secret_access_key=os.getenv('AWS_SECRET')
    )
    response = s3_client.list_objects_v2(Bucket=bucket_name)
    return response.get('Contents', [])


def initialize_storage(tenant_id, custom_plugin):
    """Initialize storage for a tenant with validated inputs."""
    # Validate tenant_id to prevent injection
    if not tenant_id.isalnum():
        raise ValueError(f"Invalid tenant_id: {tenant_id}")
    
    bucket = f'tenant-{tenant_id}-storage'
    return load_cloud_credentials(bucket, custom_plugin)
```