# Import Hijacking via Untrusted sys.path Manipulation

Language: Python
Severity: Critical
CWE: CWE-94

## Source
4, 12

## Flow
4-5-6-7, 12-13-4-5-6-7

## Sink
7

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

def load_cloud_storage_plugin(tenant_id, plugin_dir):
    storage_path = os.path.join('/var/cloud/tenants', tenant_id, plugin_dir)
    if os.path.exists(storage_path):
        sys.path.insert(0, storage_path)
    import storage_adapter
    return storage_adapter.get_client()

def sync_tenant_data(tenant_id, custom_module_path):
    client = load_cloud_storage_plugin(tenant_id, custom_module_path)
    data = client.fetch_all()
    return data
```

## Explanation

The application accepts user-controlled input (tenant_id and plugin_dir) and uses it to construct a file path that is inserted into sys.path. This allows an attacker to control which directory Python searches for modules, enabling them to place a malicious storage_adapter.py in a predictable location that will be imported instead of the legitimate module.

## Remediation

The fix validates tenant_id and plugin_dir using strict allowlist regex patterns to prevent path traversal characters, canonicalizes the resulting path and verifies it stays within the expected base directory, and uses importlib.util to load the module from a specific file path instead of manipulating sys.path which could affect other imports globally.

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

ALLOWED_TENANT_ID_PATTERN = re.compile(r'^[a-zA-Z0-9_\-]+$')
ALLOWED_PLUGIN_DIR_PATTERN = re.compile(r'^[a-zA-Z0-9_\-]+$')
BASE_TENANT_PATH = '/var/cloud/tenants'

def load_cloud_storage_plugin(tenant_id, plugin_dir):
    # Validate tenant_id and plugin_dir to prevent path traversal
    if not ALLOWED_TENANT_ID_PATTERN.match(tenant_id):
        raise ValueError(f"Invalid tenant_id: {tenant_id}")
    if not ALLOWED_PLUGIN_DIR_PATTERN.match(plugin_dir):
        raise ValueError(f"Invalid plugin_dir: {plugin_dir}")

    # Construct and canonicalize the path
    storage_path = os.path.realpath(os.path.join(BASE_TENANT_PATH, tenant_id, plugin_dir))

    # Verify the resolved path is still under the expected base directory
    expected_prefix = os.path.realpath(BASE_TENANT_PATH)
    if not storage_path.startswith(expected_prefix + os.sep):
        raise ValueError(f"Path escapes tenant directory: {storage_path}")

    # Load the specific module file directly instead of manipulating sys.path
    module_file = os.path.join(storage_path, 'storage_adapter.py')
    if not os.path.isfile(module_file):
        raise FileNotFoundError(f"Storage adapter not found at: {module_file}")

    # Use importlib to load from a specific file path without modifying sys.path
    spec = importlib.util.spec_from_file_location("storage_adapter", module_file)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from: {module_file}")

    storage_adapter = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(storage_adapter)

    if not hasattr(storage_adapter, 'get_client'):
        raise AttributeError("storage_adapter module missing 'get_client' function")

    return storage_adapter.get_client()

def sync_tenant_data(tenant_id, custom_module_path):
    client = load_cloud_storage_plugin(tenant_id, custom_module_path)
    data = client.fetch_all()
    return data
```