Patched code sample

import io
from typing import Any, List, Optional, Type


class PdfReadError(Exception):
    """An error while reading a PDF file."""
    pass


class PdfObject:
    """A base class for all PDF object types."""
    def get_object(self) -> "PdfObject":
        return self

    def write_to_stream(self, stream: io.BytesIO, encryption_key: Any = None) -> None:
        pass


def read_object(stream: io.BytesIO, pdf: Any) -> Any:
    """Mock function to simulate reading a PDF object from a stream."""
    char = stream.read(1)
    stream.seek(-1, 1)
    if char == b"]":
        return None
    
    # Simulate reading a token
    while stream.read(1) not in (b" ", b"]"):
        pass
    stream.seek(-1, 1)
    return PdfObject()


class ArrayObject(list, PdfObject):
    """
    Represents a PDF array object, demonstrating the fix for excessive entries.

    The fix introduces a hard limit on the number of entries that can be read
    from a stream to prevent a Denial of Service attack.
    """
    
    # The core of the fix: A class-level constant defining the maximum
    # number of entries allowed in an array to prevent resource exhaustion.
    ARRAY_MAX_ENTRIES = 10000

    def __init__(self, arr: Optional[List[Any]] = None):
        if arr is not None:
            super().__init__(arr)
        else:
            super().__init__()

    @classmethod
    def read_from_stream(
        cls: Type["ArrayObject"], stream: io.BytesIO, pdf: Any
    ) -> "ArrayObject":
        """
        Reads an array from the input stream with a security limit.

        :raises PdfReadError: If the array exceeds the maximum number of entries.
        """
        arr = cls()
        stream.read(1)  # Consume the "["
        
        # Skip leading whitespace
        while True:
            tok = stream.read(1)
            if not tok or not tok.isspace():
                stream.seek(-1, 1)
                break

        nb_entries = 0
        while True:
            # Check for the end of the array
            tok = stream.read(1)
            stream.seek(-1, 1)
            if tok == b"]":
                break

            # --- START OF THE FIX ---
            # This check prevents the loop from running an excessive number of times,
            # which was the cause of the Denial of Service vulnerability. If a
            # malicious PDF defines an array with millions of entries, this
            # check will raise an exception instead of trying to process them all.
            if nb_entries > cls.ARRAY_MAX_ENTRIES:
                raise PdfReadError(
                    f"Array exceeds maximum number of entries: {cls.ARRAY_MAX_ENTRIES}"
                )
            # --- END OF THE FIX ---
            nb_entries += 1

            # Read the next object in the array
            arr.append(read_object(stream, pdf).get_object())

            # Skip trailing whitespace
            while True:
                tok = stream.read(1)
                if not tok or not tok.isspace():
                    stream.seek(-1, 1)
                    break
        
        stream.read(1)  # Consume the "]"
        return arr

# Example demonstrating the fix in action.

# 1. A valid, small array (will succeed)
try:
    # Simulates a PDF stream containing "[1 2 3]"
    valid_stream = io.BytesIO(b"[ 1 2 3 ]")
    array_obj = ArrayObject.read_from_stream(valid_stream, pdf=None)
    # print(f"Successfully read a valid array with {len(array_obj)} entries.")
except PdfReadError as e:
    # print(f"Error reading valid array: {e}")
    pass


# 2. A malicious, large array (will be blocked by the fix)
try:
    # This simulates a stream with more entries than ARRAY_MAX_ENTRIES.
    # We set the limit lower for this demonstration.
    ArrayObject.ARRAY_MAX_ENTRIES = 10
    
    # Simulates a PDF stream containing "[1 1 1 1 1 1 1 1 1 1 1 ... ]"
    malicious_content = b"[" + b" 1" * 15 + b" ]"
    malicious_stream = io.BytesIO(malicious_content)
    
    # This call will now raise a PdfReadError due to the implemented fix.
    ArrayObject.read_from_stream(malicious_stream, pdf=None)

except PdfReadError as e:
    # This exception demonstrates that the vulnerability is mitigated.
    # print(f"\nFIX DEMONSTRATED: Blocked malicious array.\nError: {e}")
    pass