# Insecure Deserialization via Python Pickle in Multiprocessing Queue IPC

Language: Python
Severity: Critical
CWE: CWE-502

## Source
9

## Flow
9-11

## Sink
11

## Vulnerable Code
```python
import pickle
import multiprocessing
import base64
from flask import Flask, request
app = Flask(__name__)
task_queue = multiprocessing.Queue()
@app.route('/submit_sensor_payload', methods=['POST'])
def ingest_sensor_data():
    raw_b64 = request.form.get('sensor_blob')
    decoded_bytes = base64.b64decode(raw_b64)
    sensor_obj = pickle.loads(decoded_bytes)
    task_queue.put(sensor_obj)
    return {'status': 'queued', 'type': str(type(sensor_obj).__name__)}, 202
```

## Explanation

The endpoint reads `sensor_blob` directly from attacker-controlled HTTP form data (line 9), base64-decodes it (line 10), and passes the resulting bytes to `pickle.loads()` (line 11) without any validation, signing, or integrity check. Python's pickle deserialization can execute arbitrary code during the unpickling process via `__reduce__` or similar magic methods, giving an attacker full remote code execution on the gateway host.

## Remediation

The fix completely replaces `pickle.loads()` with `json.loads()`, which is a safe deserialization format that cannot execute arbitrary code during parsing. Additionally, a strict schema validation function (`validate_sensor_data`) is applied to the deserialized data, ensuring only expected fields with allowed types are accepted. Input size limits and proper error handling for base64 decoding and JSON parsing are also added to harden the endpoint.

## Secure Code
```python
import json
import multiprocessing
import base64
from flask import Flask, request, jsonify

app = Flask(__name__)
task_queue = multiprocessing.Queue()

# Define the strict schema of allowed sensor data fields and types
ALLOWED_SENSOR_FIELDS = {
    "device_id": str,
    "timestamp": (int, float),
    "temperature": (int, float),
    "humidity": (int, float),
    "pressure": (int, float),
    "battery_level": (int, float),
    "status": str,
}

MAX_PAYLOAD_SIZE = 1 * 1024 * 1024  # 1 MB limit


def validate_sensor_data(data):
    """Validate that sensor data conforms to the expected schema."""
    if not isinstance(data, dict):
        raise ValueError("Sensor data must be a JSON object")

    # Require at least a device_id
    if "device_id" not in data:
        raise ValueError("Missing required field: device_id")

    for key, value in data.items():
        if key not in ALLOWED_SENSOR_FIELDS:
            raise ValueError(f"Unexpected field: {key}")
        expected_type = ALLOWED_SENSOR_FIELDS[key]
        if not isinstance(value, expected_type):
            raise ValueError(
                f"Invalid type for field '{key}': expected {expected_type}, got {type(value).__name__}"
            )

    return data


@app.route('/submit_sensor_payload', methods=['POST'])
def ingest_sensor_data():
    raw_b64 = request.form.get('sensor_blob')
    if not raw_b64:
        return jsonify({'error': 'Missing sensor_blob field'}), 400

    # Enforce payload size limit before decoding
    if len(raw_b64) > MAX_PAYLOAD_SIZE:
        return jsonify({'error': 'Payload too large'}), 413

    try:
        decoded_bytes = base64.b64decode(raw_b64, validate=True)
    except Exception:
        return jsonify({'error': 'Invalid base64 encoding'}), 400

    # Use JSON for safe deserialization instead of pickle
    try:
        sensor_obj = json.loads(decoded_bytes)
    except (json.JSONDecodeError, UnicodeDecodeError):
        return jsonify({'error': 'Invalid JSON payload'}), 400

    # Validate the deserialized data against a strict schema
    try:
        validated_data = validate_sensor_data(sensor_obj)
    except ValueError as e:
        return jsonify({'error': f'Validation failed: {str(e)}'}), 422

    task_queue.put(validated_data)
    return jsonify({'status': 'queued', 'device_id': validated_data.get('device_id', 'unknown')}), 202
```