# Unsafe `marshal.loads()` Deserialization Leading to Code Execution

Language: Python
Severity: Critical
CWE: CWE-502

## Source
9

## Flow
9-13-14-15-16-17

## Sink
17

## Vulnerable Code
```python
import marshal
import base64
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/ml/model/predict', methods=['POST'])
def execute_inference():
    model_payload = request.json.get('serialized_model')
    input_features = request.json.get('features', [])
    if not model_payload:
        return jsonify({'error': 'Model payload required'}), 400
    try:
        decoded_bytes = base64.b64decode(model_payload)
        model_code = marshal.loads(decoded_bytes)
        prediction_func = eval(model_code)
        result = prediction_func(input_features)
        return jsonify({'prediction': result, 'status': 'success'})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)
```

## Explanation

The application accepts untrusted serialized model data from user input, decodes it, and deserializes it using marshal.loads() followed by eval(). This allows an attacker to inject arbitrary Python bytecode that will be executed on the server, leading to remote code execution.

## Remediation

The fix completely eliminates the deserialization vulnerability by removing marshal.loads(), eval(), and any acceptance of serialized code from user input. Instead, models are pre-registered server-side and loaded using safe formats (numpy arrays with allow_pickle=False), and clients reference models by a validated model_id. Input validation ensures features are numeric and model IDs follow a strict format.

## Secure Code
```python
import hashlib
import hmac
import os
import json
from flask import Flask, request, jsonify
import numpy as np

app = Flask(__name__)

# Pre-registered models stored server-side (not accepted from user input)
# Models are identified by a secure model_id and loaded from a trusted location
MODEL_REGISTRY_PATH = os.environ.get('MODEL_REGISTRY_PATH', '/opt/models/registry')
ALLOWED_MODEL_IDS = set()

def load_model_registry():
    """Load the list of pre-approved model IDs from a secure registry."""
    registry_file = os.path.join(MODEL_REGISTRY_PATH, 'registry.json')
    if os.path.exists(registry_file):
        with open(registry_file, 'r') as f:
            registry = json.load(f)
            return set(registry.get('approved_models', []))
    return set()

def get_model_predictor(model_id):
    """Load a pre-trained model from the secure server-side registry.

    Models are stored as safe serialization formats (e.g., model weights/parameters)
    and loaded using safe libraries like numpy, not arbitrary code execution.
    """
    if model_id not in ALLOWED_MODEL_IDS:
        return None

    model_path = os.path.join(MODEL_REGISTRY_PATH, f"{model_id}.npz")
    if not os.path.exists(model_path):
        return None

    # Load model weights using safe numpy format (no code execution)
    model_data = np.load(model_path, allow_pickle=False)
    weights = model_data['weights']
    bias = model_data['bias']

    # Return a simple linear prediction function using loaded parameters
    def predict(features):
        features_array = np.array(features, dtype=np.float64)
        return (features_array @ weights + bias).tolist()

    return predict

@app.route('/api/ml/model/predict', methods=['POST'])
def execute_inference():
    model_id = request.json.get('model_id')
    input_features = request.json.get('features', [])

    if not model_id:
        return jsonify({'error': 'model_id required'}), 400

    # Validate model_id format (alphanumeric and hyphens only)
    if not all(c.isalnum() or c == '-' for c in model_id):
        return jsonify({'error': 'Invalid model_id format'}), 400

    # Validate input features
    if not isinstance(input_features, list):
        return jsonify({'error': 'Features must be a list'}), 400

    if not all(isinstance(f, (int, float)) for f in input_features):
        return jsonify({'error': 'Features must be numeric values'}), 400

    try:
        prediction_func = get_model_predictor(model_id)
        if prediction_func is None:
            return jsonify({'error': 'Model not found or not approved'}), 404

        result = prediction_func(input_features)
        return jsonify({'prediction': result, 'status': 'success'})
    except (ValueError, TypeError) as e:
        return jsonify({'error': f'Prediction failed: {str(e)}'}), 400
    except Exception as e:
        # Log the error server-side but don't expose details to client
        app.logger.error(f'Inference error for model {model_id}: {str(e)}')
        return jsonify({'error': 'Internal prediction error'}), 500

@app.route('/api/ml/models', methods=['GET'])
def list_models():
    """List available pre-approved models."""
    return jsonify({'available_models': list(ALLOWED_MODEL_IDS)})

if __name__ == '__main__':
    ALLOWED_MODEL_IDS = load_model_registry()
    app.run(debug=False, host='127.0.0.1', port=5000)
```