Patched code sample

import json
import importlib

# In a real-world fix, this set would be populated with the fully-qualified
# string names of classes that are considered safe and necessary for the
# application's functionality.
# For example: 'epyt_flow.scenarios.Scenario', 'epyt_flow.data.Model'
SAFE_TO_DESERIALIZE = {
    'my_app.safe_models.Configuration',
    'my_app.safe_models.UserData'
}


def my_load_from_json_fixed(json_string: str):
    """
    Represents the fixed version of the JSON deserializer.

    The vulnerability is fixed by replacing the dangerous dynamic class loading
    with a validation mechanism. It checks if the type specified in the JSON
    payload is present in a pre-defined 'allow-list' (SAFE_TO_DESERIALIZE).

    If the type is not in the allow-list, it refuses to proceed, raising
    a TypeError. This prevents attackers from specifying and instantiating
    arbitrary, malicious classes like 'subprocess.Popen'.
    """
    data = json.loads(json_string)

    if isinstance(data, dict) and 'type' in data:
        requested_type = data.get('type')

        # --- FIX STARTS HERE ---
        # Instead of blindly importing, validate the requested type against
        # a strict, pre-defined set of allowed classes.
        if requested_type not in SAFE_TO_DESERIALIZE:
            raise TypeError(
                f"Deserialization of type '{requested_type}' is not allowed for security reasons."
            )
        # --- FIX ENDS HERE ---

        # If the type is safe, proceed with controlled instantiation.
        # This part of the logic is retained to preserve functionality for
        # legitimate, safe types.
        try:
            module_path, class_name = requested_type.rsplit('.', 1)
            module = importlib.import_module(module_path)
            cls = getattr(module, class_name)
            
            # The arguments for instantiation are still taken from the payload,
            # but only for the now-verified-safe class.
            args = data.get('args', {})
            
            # This is now safe because 'cls' is guaranteed to be one of the
            # types from the SAFE_TO_DESERIALIZE set.
            return cls(**args)
        except (ImportError, AttributeError, ValueError) as e:
            # Handle potential errors during instantiation of safe types
            raise RuntimeError(f"Failed to instantiate allowed type '{requested_type}': {e}") from e

    # If no 'type' is specified, just return the parsed data as-is.
    return data


# Example of a 'safe' module and classes for demonstration purposes.
# In a real application, these would be part of the package's own code.
class MockSafeModule:
    class Configuration:
        def __init__(self, settings=None):
            self.settings = settings or {}
            print(f"SAFE: Instantiated Configuration with settings: {self.settings}")

    class UserData:
        def __init__(self, username=None):
            self.username = username
            print(f"SAFE: Instantiated UserData for user: {self.username}")

# To make the example runnable, we can inject this mock module into sys.modules.
import sys
# This creates a dummy module 'my_app.safe_models' that can be imported.
safe_models = MockSafeModule()
sys.modules['my_app'] = type('module', (), {})()
sys.modules['my_app.safe_models'] = safe_models


if __name__ == '__main__':
    # --- DEMONSTRATION OF THE FIX ---

    # 1. Malicious Payload attempting to execute a command.
    # This payload would have worked with the vulnerable version.
    malicious_payload = """
    {
      "type": "subprocess.Popen",
      "args": {
        "args": ["touch", "/tmp/pwned"]
      }
    }
    """
    print("--- Attempting to deserialize a malicious payload ---")
    try:
        my_load_from_json_fixed(malicious_payload)
    except TypeError as e:
        print(f"SUCCESS: The fix prevented deserialization. Error: {e}\n")


    # 2. Legitimate Payload using an allowed type.
    # This payload is for a class listed in SAFE_TO_DESERIALIZE.
    legitimate_payload = """
    {
      "type": "my_app.safe_models.Configuration",
      "args": {
        "settings": {"mode": "test", "retries": 3}
      }
    }
    """
    print("--- Attempting to deserialize a legitimate payload ---")
    try:
        result = my_load_from_json_fixed(legitimate_payload)
        print(f"SUCCESS: The fix allowed deserialization of a safe object.")
        print(f"Result type: {type(result)}, Settings: {result.settings}\n")
    except Exception as e:
        print(f"FAILURE: An unexpected error occurred: {e}\n")

    
    # 3. Benign payload without a 'type' field.
    # This should always be safe.
    simple_payload = """
    {
        "data": "some_value",
        "id": 123
    }
    """
    print("--- Attempting to deserialize a simple JSON payload ---")
    try:
        result = my_load_from_json_fixed(simple_payload)
        print("SUCCESS: The fix correctly handled a simple JSON object.")
        print(f"Result: {result}")
    except Exception as e:
        print(f"FAILURE: An unexpected error occurred: {e}")