Patched code sample

# Disclaimer: This code is a *hypothetical* example and may not perfectly represent the actual CVE-2025-2999 fix.  It aims to illustrate a potential type of fix for memory corruption in sequence unpacking, given the information available about the vulnerability. Actual code could be different.

def unpack_sequence_fixed(packed_sequence, enforce_sorted=True):
    """
    Unpacks a packed sequence into a padded tensor.

    Args:
        packed_sequence: A PackedSequence object.
        enforce_sorted: If True, raise an error if the input sequence is not sorted.

    Returns:
        A tuple containing:
        - A padded tensor of shape (T, B, *), where T is the length of the longest sequence,
          B is the batch size, and * is the input_size.
        - A LongTensor of shape (B,) containing the lengths of the sequences in the batch.
    """

    data = packed_sequence.data
    batch_sizes = packed_sequence.batch_sizes
    sorted_indices = packed_sequence.sorted_indices
    unsorted_indices = packed_sequence.unsorted_indices

    batch_size = packed_sequence.batch_sizes[0].item() #fixed bug: convert numpy int to python int

    if enforce_sorted and sorted_indices is None:
        raise ValueError("Input sequence must be sorted if enforce_sorted is True.")

    max_seq_len = batch_sizes.size(0)

    # Pre-allocate the padded tensor.  Crucial to prevent out-of-bounds writes.
    padded_tensor = torch.zeros((max_seq_len, batch_size, data.size(1)), dtype=data.dtype, device=data.device)

    # Iterate and unpack.  Adding bounds checks to prevent memory corruption.
    current_index = 0
    for t in range(max_seq_len):
        current_batch_size = batch_sizes[t].item() #fixed bug: convert numpy int to python int

        # IMPORTANT: Check if the current batch size is valid. If not, continue to prevent out-of-bounds write.
        if current_batch_size > batch_size:
            print(f"Warning: batch_size[{t}] is {current_batch_size}, exceeding maximum batch size of {batch_size}. Skipping timestep.")
            continue

        padded_tensor[t, :current_batch_size] = data[current_index:current_index + current_batch_size]
        current_index += current_batch_size

        #VERY IMPORTANT: Prevent reading past the end of the data.
        if current_index > data.shape[0]:
            print(f"Warning: current index exceeded data size. Stopping at timestep {t}.")
            break

    lengths = batch_sizes_to_lengths(batch_sizes) #Fixed bug: proper batch sizes to length convertion.

    if unsorted_indices is not None:
        padded_tensor = padded_tensor[:, unsorted_indices]
        lengths = lengths[unsorted_indices]

    return padded_tensor, lengths

def batch_sizes_to_lengths(batch_sizes):
  """
  Convert batch_sizes from PackedSequence to lengths.

  Args:
    batch_sizes: A LongTensor of shape (T,) containing the batch sizes at each time step.

  Returns:
    A LongTensor of shape (B,) containing the lengths of the sequences in the batch.
  """
  lengths = torch.zeros(batch_sizes[0].item(), dtype=torch.long, device=batch_sizes.device) #fixed bug: proper batch sizes to length convertion.
  for i, batch_size in enumerate(batch_sizes):
    lengths[:batch_size] += 1
  return lengths

import torch
from torch.nn.utils.rnn import PackedSequence
if __name__ == '__main__':
    # Example usage (illustrative)
    # Create a dummy PackedSequence
    data = torch.randn(10, 5)  # Example data
    batch_sizes = torch.tensor([5, 4, 3, 2, 1])  # Example batch sizes
    packed_sequence = PackedSequence(data, batch_sizes)

    # Unpack the sequence
    padded_tensor, lengths = unpack_sequence_fixed(packed_sequence)

    print("Padded Tensor shape:", padded_tensor.shape)
    print("Lengths:", lengths)