Patched code sample

import pickle
import os
import base64
import json

# This CVE is fictitious. The code below demonstrates a hypothetical vulnerability
# similar to the one described and the corresponding standard fix.

# --- Step 1: Create a malicious payload ---
# An attacker would create a special object that, when deserialized by pickle,
# executes arbitrary code. The __reduce__ method is a common way to do this.

class RemoteCodeExecutor:
    def __reduce__(self):
        # This tuple tells pickle to call os.system with the provided command.
        command = 'echo "[VULNERABILITY EXPLOITED] Arbitrary code was executed."'
        return (os.system, (command,))

# Serialize the malicious object using pickle and encode it for transport.
# In a real attack, this payload would be sent over the network.
malicious_payload = pickle.dumps(RemoteCodeExecutor())
encoded_malicious_payload = base64.b64encode(malicious_payload)


# --- Step 2: Define the vulnerable and fixed functions ---

def vulnerable_deserialization(untrusted_data: bytes):
    """
    VULNERABLE: This function directly uses pickle.loads() on untrusted data,
    which can lead to Remote Code Execution (RCE).
    """
    print("--- Calling VULNERABLE function ---")
    print("Attempting to deserialize potentially malicious data...")
    try:
        # The vulnerability is here: deserializing untrusted data.
        deserialized_object = pickle.loads(base64.b64decode(untrusted_data))
        print("Deserialization successful (this is bad).")
        return deserialized_object
    except Exception as e:
        print(f"An error occurred: {e}")


def fixed_deserialization(untrusted_data: bytes):
    """
    FIXED: This function demonstrates the fix. Instead of using a dangerous
    deserializer like pickle, it uses a safe, data-only format like JSON.
    This prevents code execution because JSON can only represent data, not
    executable objects or commands.
    """
    print("\n--- Calling FIXED function ---")
    print("Attempting to deserialize data using a safe method (JSON)...")
    try:
        # THE FIX: Use a safe data interchange format like JSON.
        # The malicious pickle payload is not valid JSON, so this will
        # raise an error and the code execution will be prevented.
        # In a real-world scenario, the service would be changed to send
        # data as JSON instead of a pickled object.
        data_string = base64.b64decode(untrusted_data).decode('utf-8')
        deserialized_object = json.loads(data_string)
        print(f"Safely deserialized data: {deserialized_object}")
        return deserialized_object
    except (json.JSONDecodeError, UnicodeDecodeError, TypeError):
        # This is the expected and safe outcome for the malicious payload.
        print("[FIX APPLIED] The payload was not valid JSON and was safely rejected.")
        print("Code execution was prevented.")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    print(f"Generated malicious payload (encoded): {encoded_malicious_payload}\n")

    # Demonstrate the vulnerability
    vulnerable_deserialization(encoded_malicious_payload)

    # Demonstrate the fix
    fixed_deserialization(encoded_malicious_payload)

    # --- Example with legitimate JSON data to show the fixed function works ---
    legitimate_data = {"user": "admin", "id": 12345, "action": "read"}
    # In a real system, this would come from the network, base64 encoded.
    encoded_legitimate_data = base64.b64encode(json.dumps(legitimate_data).encode('utf-8'))
    print(f"\n--- Testing FIXED function with legitimate data ---")
    print(f"Legitimate payload (encoded): {encoded_legitimate_data}")
    fixed_deserialization(encoded_legitimate_data)