# Import Hijacking via Untrusted Module Resolution

Language: Python
Severity: Critical
CWE: CWE-94

## Source
16-17

## Flow
16-17-18-4-6-9

## Sink
6, 9

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

def load_cloud_credentials(provider_name, config_dir):
    credential_path = os.path.join(config_dir, 'plugins')
    sys.path.insert(0, credential_path)
    try:
        cloud_module = __import__(f'{provider_name}_auth')
        authenticator = cloud_module.CloudAuthenticator()
        session = authenticator.get_session()
        return session
    except ImportError:
        return None

def deploy_to_cloud(service_config):
    user_provider = service_config.get('cloud_provider', 'aws')
    plugin_dir = service_config.get('plugin_directory', '/tmp/cloud_plugins')
    session = load_cloud_credentials(user_provider, plugin_dir)
    return session.deploy(service_config['manifest'])
```

## Explanation

The application accepts user-controlled input for both the plugin directory path and cloud provider module name, then inserts this path at the beginning of sys.path and dynamically imports modules. An attacker can specify a malicious directory containing a crafted Python module that will be imported and executed with the application's privileges, leading to arbitrary code execution.

## Remediation

The fix removes user control over both the plugin directory and module name by enforcing an allowlist of known cloud providers and loading modules only from a hardcoded trusted directory using importlib.util.spec_from_file_location (avoiding sys.path manipulation). Path traversal is prevented by resolving the real path and verifying it remains within the trusted directory.

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

# Allowlist of trusted cloud providers
ALLOWED_PROVIDERS = {'aws', 'azure', 'gcp'}

# Trusted plugin directory (not user-configurable)
TRUSTED_PLUGIN_DIR = '/opt/cloud_deploy/plugins'


def _validate_provider_name(provider_name):
    """Validate that the provider name is in the allowlist."""
    if provider_name not in ALLOWED_PROVIDERS:
        raise ValueError(
            f"Unsupported cloud provider: '{provider_name}'. "
            f"Allowed providers: {', '.join(sorted(ALLOWED_PROVIDERS))}"
        )
    return provider_name


def load_cloud_credentials(provider_name):
    """Load cloud credentials from a trusted, fixed plugin directory."""
    # Validate provider name against allowlist
    validated_provider = _validate_provider_name(provider_name)

    module_name = f'{validated_provider}_auth'
    module_file = f'{module_name}.py'
    module_path = os.path.join(TRUSTED_PLUGIN_DIR, module_file)

    # Resolve to absolute path and verify it stays within trusted directory
    module_path = os.path.realpath(module_path)
    trusted_dir = os.path.realpath(TRUSTED_PLUGIN_DIR)

    if not module_path.startswith(trusted_dir + os.sep):
        raise ValueError("Module path escapes trusted plugin directory.")

    if not os.path.isfile(module_path):
        return None

    # Load module from specific file path without modifying sys.path
    try:
        spec = importlib.util.spec_from_file_location(module_name, module_path)
        if spec is None or spec.loader is None:
            return None
        cloud_module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(cloud_module)
        authenticator = cloud_module.CloudAuthenticator()
        session = authenticator.get_session()
        return session
    except (ImportError, AttributeError, OSError):
        return None


def deploy_to_cloud(service_config):
    user_provider = service_config.get('cloud_provider', 'aws')
    # plugin_directory from config is intentionally ignored; only trusted dir is used
    session = load_cloud_credentials(user_provider)
    if session is None:
        raise RuntimeError(f"Failed to load credentials for provider: {user_provider}")
    return session.deploy(service_config['manifest'])
```