# Import Hijacking via Malicious sys.path Manipulation

Language: Python
Severity: Critical
CWE: CWE-706

## Source
4

## Flow
4-5-7

## Sink
5

## Vulnerable Code
```python
import sys
import importlib
def load_sensor_plugin(plugin_dir, plugin_name):
    usr_plugin_path = plugin_dir.strip()
    sys.path.insert(0, usr_plugin_path)
    try:
        sensor_mod = importlib.import_module(plugin_name)
        handler = getattr(sensor_mod, 'run_sensor')
        return handler()
    except (ImportError, AttributeError) as ex:
        return {'error': str(ex)}
    finally:
        pass
```

## Explanation

The function accepts `plugin_dir` from user input and, after a trivial `.strip()` call, inserts it at position 0 of `sys.path` with the highest priority. This allows an attacker to supply a malicious directory (e.g., `/tmp/evil`) that shadows legitimate standard-library or third-party modules. When `importlib.import_module(plugin_name)` is called on line 7, Python resolves the module from the attacker-controlled path first, loading arbitrary attacker-supplied code and achieving remote code execution. The empty `finally` block ensures the poisoned path entry persists for all subsequent imports in the process.

## Remediation

The fix addresses the vulnerability through multiple layers of defense: (1) plugin_dir is canonicalized via os.path.realpath and checked against an explicit allowlist of permitted directories, preventing arbitrary path injection; (2) plugin_name is validated against a strict regex pattern and a blocklist of critical stdlib module names to prevent module shadowing; (3) the directory is appended to sys.path (instead of inserted at position 0) so standard library modules always take precedence; (4) after import, the module's __file__ is verified to confirm it actually resides within the allowed directory; (5) the finally block now properly removes the added path entry to prevent persistent path poisoning across subsequent imports.

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

# Allowlist of permitted plugin directories (configure per deployment)
ALLOWED_PLUGIN_DIRS = [
    '/opt/iot-platform/plugins',
    '/usr/local/share/sensor-plugins',
]

# Allowlist pattern for valid plugin names (alphanumeric + underscores only)
_VALID_PLUGIN_NAME = re.compile(r'^[A-Za-z_][A-Za-z0-9_]*$')

# Set of standard-library / critical module names that must never be shadowed
_BLOCKED_MODULE_NAMES = frozenset({
    'os', 'sys', 'json', 'subprocess', 'importlib', 'builtins', 'io',
    'socket', 'http', 'shutil', 'pathlib', 'ctypes', 'code', 'pickle',
    'marshal', 'compileall', 'runpy', 'xmlrpc', 'multiprocessing',
})


def load_sensor_plugin(plugin_dir, plugin_name):
    # --- Validate plugin_name ---
    if not isinstance(plugin_name, str) or not _VALID_PLUGIN_NAME.match(plugin_name):
        return {'error': 'Invalid plugin name. Only alphanumeric characters and underscores are allowed.'}

    if plugin_name in _BLOCKED_MODULE_NAMES:
        return {'error': f'Plugin name "{plugin_name}" conflicts with a protected module and cannot be loaded.'}

    # --- Validate and canonicalize plugin_dir ---
    if not isinstance(plugin_dir, str):
        return {'error': 'plugin_dir must be a string.'}

    resolved_dir = os.path.realpath(os.path.normpath(plugin_dir.strip()))

    # Check against the allowlist of permitted directories
    if not any(resolved_dir == os.path.realpath(allowed) or
               resolved_dir.startswith(os.path.realpath(allowed) + os.sep)
               for allowed in ALLOWED_PLUGIN_DIRS):
        return {'error': f'Plugin directory "{resolved_dir}" is not in the list of allowed plugin directories.'}

    if not os.path.isdir(resolved_dir):
        return {'error': f'Plugin directory "{resolved_dir}" does not exist or is not a directory.'}

    # --- Safely modify sys.path and import ---
    # Append (not insert at 0) so that stdlib/site-packages are still resolved first,
    # preventing module shadowing even if validation is somehow bypassed.
    path_added = False
    if resolved_dir not in sys.path:
        sys.path.append(resolved_dir)
        path_added = True

    try:
        sensor_mod = importlib.import_module(plugin_name)

        # Verify the loaded module actually resides under the allowed directory
        mod_file = getattr(sensor_mod, '__file__', None)
        if mod_file is None:
            return {'error': 'Loaded module has no __file__ attribute; cannot verify origin.'}

        mod_real_path = os.path.realpath(mod_file)
        if not mod_real_path.startswith(resolved_dir + os.sep):
            return {'error': 'Loaded module does not originate from the specified plugin directory.'}

        if not hasattr(sensor_mod, 'run_sensor') or not callable(sensor_mod.run_sensor):
            return {'error': f'Plugin "{plugin_name}" does not expose a callable "run_sensor" function.'}

        handler = sensor_mod.run_sensor
        return handler()
    except (ImportError, AttributeError) as ex:
        return {'error': str(ex)}
    finally:
        # Clean up: remove the path entry so it does not persist and affect future imports
        if path_added and resolved_dir in sys.path:
            sys.path.remove(resolved_dir)
```
