# Arbitrary Code Execution via logging.config.listen()

Language: Python
Severity: Critical
CWE: CWE-502

## Source
8

## Flow
8-9

## Sink
9

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

def bootstrap_cloud_telemetry(port=9030, verify_cert=False):
    telemetry_host = '0.0.0.0'
    print(f'[CloudMonitor] Initializing telemetry receiver on {telemetry_host}:{port}')
    listener = threading.Thread(
        target=logging.config.listen,
        args=(port,),
        daemon=True
    )
    listener.start()
    print('[CloudMonitor] Telemetry service active - accepting remote log configurations')
    return listener
```

## Explanation

The logging.config.listen() function accepts pickled configuration objects from any network client without authentication or validation. An attacker can send a malicious serialized pickle payload that executes arbitrary code when deserialized by the logging configuration listener.

## Remediation

The fix replaces the dangerous logging.config.listen() call (which deserializes arbitrary pickle data) with a custom secure listener that only accepts JSON-formatted logging configurations, validates them against a whitelist of allowed keys and handler classes, and applies them safely via logging.config.dictConfig(). Additionally, the listener binds to localhost (127.0.0.1) instead of all interfaces (0.0.0.0) to limit network exposure.

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

def _verify_config(config_bytes):
    """Verify and validate logging configuration before applying.
    Only accepts JSON-formatted logging configurations with allowed handlers."""
    try:
        config = json.loads(config_bytes)
    except (json.JSONDecodeError, UnicodeDecodeError):
        print('[CloudMonitor] Rejected invalid configuration: not valid JSON')
        return None

    allowed_top_keys = {'version', 'disable_existing_loggers', 'formatters', 'filters', 'handlers', 'loggers', 'root', 'incremental'}
    if not set(config.keys()).issubset(allowed_top_keys):
        print('[CloudMonitor] Rejected configuration: contains disallowed keys')
        return None

    allowed_handler_classes = {
        'logging.StreamHandler',
        'logging.FileHandler',
        'logging.handlers.RotatingFileHandler',
        'logging.handlers.TimedRotatingFileHandler',
        'logging.handlers.SysLogHandler',
        'logging.NullHandler',
    }

    handlers = config.get('handlers', {})
    for handler_name, handler_config in handlers.items():
        handler_class = handler_config.get('class', '')
        if handler_class not in allowed_handler_classes:
            print(f'[CloudMonitor] Rejected configuration: disallowed handler class "{handler_class}"')
            return None

    return config


def _secure_config_listener(port, bind_address='127.0.0.1'):
    """A secure logging configuration listener that only accepts JSON configs
    and validates them before applying."""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((bind_address, port))
    server.listen(1)
    print(f'[CloudMonitor] Secure config listener bound to {bind_address}:{port}')

    while True:
        try:
            conn, addr = server.accept()
            print(f'[CloudMonitor] Connection from {addr}')

            length_bytes = conn.recv(4)
            if len(length_bytes) < 4:
                conn.close()
                continue

            length = struct.unpack('>I', length_bytes)[0]

            if length > 65536:
                print('[CloudMonitor] Rejected configuration: exceeds maximum size')
                conn.close()
                continue

            config_bytes = b''
            while len(config_bytes) < length:
                chunk = conn.recv(min(4096, length - len(config_bytes)))
                if not chunk:
                    break
                config_bytes += chunk

            conn.close()

            if len(config_bytes) != length:
                print('[CloudMonitor] Rejected configuration: incomplete data')
                continue

            config = _verify_config(config_bytes)
            if config is not None:
                logging.config.dictConfig(config)
                print('[CloudMonitor] Applied validated logging configuration')
        except Exception as e:
            print(f'[CloudMonitor] Error processing configuration: {e}')


def bootstrap_cloud_telemetry(port=9030, bind_address='127.0.0.1'):
    print(f'[CloudMonitor] Initializing telemetry receiver on {bind_address}:{port}')
    listener = threading.Thread(
        target=_secure_config_listener,
        args=(port, bind_address),
        daemon=True
    )
    listener.start()
    print('[CloudMonitor] Telemetry service active - accepting validated JSON log configurations only')
    return listener
```