Patched code sample

import sys

class FixedMultipartParser:
    """
    A conceptual parser demonstrating the fix for a DoS vulnerability
    caused by large preambles and epilogues in multipart data.
    """
    _STATE_PREAMBLE = 0
    _STATE_PART = 1
    _STATE_DONE = 2

    def __init__(self, boundary: bytes):
        self.boundary = b'--' + boundary
        self.final_boundary = self.boundary + b'--'
        self.state = self._STATE_PREAMBLE
        self._buffer = b''

    def feed_data(self, data: bytes):
        """
        Feeds data to the parser.

        The fix is demonstrated by:
        1.  Immediately stopping processing and discarding data if the final
            boundary has been seen (state is _STATE_DONE).
        2.  Efficiently searching for the first boundary instead of buffering
            or slowly processing the entire preamble.
        """
        # FIX 1: If parsing is finished, immediately discard any new data (epilogue).
        # A vulnerable parser would continue to buffer this, leading to DoS.
        if self.state == self._STATE_DONE:
            return

        self._buffer += data

        if self.state == self._STATE_PREAMBLE:
            # FIX 2: Efficiently find the first boundary and discard the preamble.
            # A vulnerable parser might process the preamble byte-by-byte, consuming
            # CPU and memory for large preambles.
            try:
                boundary_pos = self._buffer.index(self.boundary)
                # Discard the preamble before the boundary
                self._buffer = self._buffer[boundary_pos:]
                self.state = self._STATE_PART
            except ValueError:
                # To prevent unbounded buffer growth from a huge preamble without a
                # boundary, only keep a small tail of the buffer.
                tail_size = len(self.boundary) - 1
                if len(self._buffer) > tail_size:
                    self._buffer = self._buffer[-tail_size:]
                return

        if self.state == self._STATE_PART:
            # In a real parser, this would process headers and data between boundaries.
            # For this demonstration, we just look for the final boundary.
            try:
                final_boundary_pos = self._buffer.index(self.final_boundary)
                # Final boundary found. Stop processing.
                self.state = self._STATE_DONE
                # Discard all buffered data, including the final boundary itself.
                self._buffer = b''
            except ValueError:
                # Final boundary not yet in buffer. A real implementation would also
                # look for intermediate boundaries to yield completed parts.
                pass