# Logging Configuration Injection via logging.config.listen()

Language: Python
Severity: Critical
CWE: CWE-502

## Source
4

## Flow
4-6-7

## Sink
7

## Vulnerable Code
```python
import logging.config
import threading

def bootstrap_cloud_telemetry(metrics_port, enable_remote=True):
    if enable_remote:
        telemetry_host = '0.0.0.0'
        config_listener = threading.Thread(target=logging.config.listen, args=(metrics_port,), daemon=False)
        config_listener.start()
        print(f"[CloudMetrics] Telemetry listener active on {telemetry_host}:{metrics_port}")
        return config_listener
    return None

bootstrap_cloud_telemetry(9030)
```

## Explanation

The code accepts an untrusted metrics_port parameter and passes it directly to logging.config.listen(), which deserializes pickle data from network connections. An attacker can send malicious pickled objects to execute arbitrary code on the server since pickle deserialization is inherently unsafe with untrusted input.

## Remediation

The fix eliminates the use of logging.config.listen() which deserializes arbitrary pickle data from the network. Instead, it implements a custom listener that only accepts JSON-formatted configurations, validates them against a strict allowlist of keys and dangerous patterns, and applies them safely via logging.config.dictConfig(). The listener is also bound to localhost only (127.0.0.1) instead of all interfaces, and includes connection source validation and port input validation.

## Secure Code
```python
import logging.config
import json
import socket
import threading

ALLOWED_LOG_LEVELS = {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'}
ALLOWED_HOSTS = {'127.0.0.1', '::1'}

def verify_logging_config(config_text):
    """Validate and sanitize logging configuration before applying."""
    try:
        config = json.loads(config_text)
    except (json.JSONDecodeError, TypeError):
        logging.warning("[CloudMetrics] Rejected invalid JSON configuration")
        return None

    # Validate structure - only allow known safe keys
    allowed_keys = {'version', 'disable_existing_loggers', 'formatters', 'handlers', 'loggers', 'root'}
    if not isinstance(config, dict):
        return None
    if not all(key in allowed_keys for key in config.keys()):
        logging.warning("[CloudMetrics] Rejected config with unknown keys")
        return None

    # Ensure no class references that could execute arbitrary code
    config_str = json.dumps(config)
    dangerous_patterns = ['()', 'eval', 'exec', 'import', '__', 'os.', 'subprocess', 'sys.']
    for pattern in dangerous_patterns:
        if pattern in config_str:
            logging.warning(f"[CloudMetrics] Rejected config containing dangerous pattern: {pattern}")
            return None

    return config


def _config_listener_thread(port, bind_address='127.0.0.1'):
    """Safe configuration listener that accepts only JSON configs on localhost."""
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.settimeout(1.0)
    server_sock.bind((bind_address, port))
    server_sock.listen(1)
    logging.info(f"[CloudMetrics] Secure config listener on {bind_address}:{port}")

    while True:
        try:
            conn, addr = server_sock.accept()
        except socket.timeout:
            continue
        except OSError:
            break

        # Only accept connections from allowed hosts
        if addr[0] not in ALLOWED_HOSTS:
            logging.warning(f"[CloudMetrics] Rejected connection from unauthorized host: {addr[0]}")
            conn.close()
            continue

        try:
            conn.settimeout(5.0)
            # Read length-prefixed data (max 64KB)
            data = b''
            while len(data) < 65536:
                chunk = conn.recv(4096)
                if not chunk:
                    break
                data += chunk

            config_text = data.decode('utf-8')
            config = verify_logging_config(config_text)

            if config is not None:
                logging.config.dictConfig(config)
                logging.info("[CloudMetrics] Logging configuration updated successfully")
            else:
                logging.warning("[CloudMetrics] Configuration rejected by validation")
        except (UnicodeDecodeError, socket.timeout, ConnectionResetError) as e:
            logging.warning(f"[CloudMetrics] Error processing config update: {e}")
        finally:
            conn.close()


def bootstrap_cloud_telemetry(metrics_port, enable_remote=True):
    # Validate port
    if not isinstance(metrics_port, int) or not (1024 <= metrics_port <= 65535):
        raise ValueError("metrics_port must be an integer between 1024 and 65535")

    if enable_remote:
        bind_address = '127.0.0.1'  # Only bind to localhost, not 0.0.0.0
        config_listener = threading.Thread(
            target=_config_listener_thread,
            args=(metrics_port, bind_address),
            daemon=True
        )
        config_listener.start()
        print(f"[CloudMetrics] Telemetry listener active on {bind_address}:{metrics_port}")
        return config_listener
    return None

bootstrap_cloud_telemetry(9030)
```