Patched code sample

import gzip
import io

# In Pillow, this is defined in ImageFile.py
MAX_DECOMPRESSED_BYTES = 256 * 1024 * 1024  # 256 MB limit

class DecompressionBombError(OSError):
    """Exception raised for decompression bomb attacks."""
    pass

class DecompressionBombCheck:
    """
    A file-like wrapper to protect against decompression bomb attacks.
    It tracks the number of bytes read from the underlying stream and
    raises a DecompressionBombError if the limit is exceeded.
    """
    def __init__(self, fp):
        self.fp = fp
        self._bytes_read = 0

    def read(self, size=-1):
        if self._bytes_read > MAX_DECOMPRESSED_BYTES:
            raise DecompressionBombError(
                f"Decompressed data exceeds size limit ({MAX_DECOMPRESSED_BYTES} bytes)"
            )
        
        data = self.fp.read(size)
        self._bytes_read += len(data)

        if self._bytes_read > MAX_DECOMPRESSED_BYTES:
            raise DecompressionBombError(
                f"Decompressed data exceeds size limit ({MAX_DECOMPRESSED_BYTES} bytes)"
            )
        return data
    
    # Delegate other necessary file-like methods
    def seek(self, *args, **kwargs):
        return self.fp.seek(*args, **kwargs)

    def tell(self, *args, **kwargs):
        return self.fp.tell(*args, **kwargs)

    def close(self, *args, **kwargs):
        return self.fp.close(*args, **kwargs)


def _open_fits_with_fix(fp):
    """
    Demonstrates the fix applied to the FITS image plugin's open method.

    The vulnerability was that `gzip.GzipFile` was used directly, allowing
    an unlimited amount of data to be decompressed.

    The fix is to wrap the `gzip.GzipFile` stream with the
    `DecompressionBombCheck` utility, which limits the total amount of
    decompressed data that can be read.
    """
    # Check for GZIP magic number
    if fp.read(3) == b"\x1f\x8b\x08":
        fp.seek(0)
        
        # --- THE FIX IS APPLIED HERE ---
        # The GZIP stream is wrapped in a DecompressionBombCheck instance.
        # This prevents reading an excessive amount of decompressed data.
        gzipped_fp = gzip.GzipFile(fileobj=fp)
        safe_fp = DecompressionBombCheck(gzipped_fp)
        # --- END OF FIX ---
        
        # Subsequent operations would now use `safe_fp` instead of the raw stream
        # For demonstration, we will try to read from it.
        try:
            # This would be where the rest of the image parsing happens
            header = safe_fp.read(2880)  # Read the first FITS header block
            print("Successfully read FITS header from GZIP stream.")
            return header
        except DecompressionBombError as e:
            print(f"Caught a potential decompression bomb: {e}")
            return None
    else:
        fp.seek(0)
        # For non-gzipped files, no wrapping is needed
        print("File is not GZIP-compressed.")
        return fp.read(2880)

if __name__ == '__main__':
    # --- Example usage ---

    # 1. A valid, small, gzipped file (will succeed)
    print("--- 1. Testing a safe, small gzipped file ---")
    header_data = b"SIMPLE  =                    T".ljust(80, b" ") * 36
    compressed_data = gzip.compress(header_data)
    safe_file = io.BytesIO(compressed_data)
    _open_fits_with_fix(safe_file)
    
    print("\n" + "="*40 + "\n")
    
    # 2. A "decompression bomb" (will be caught by the fix)
    # This creates a small (200 bytes) file that decompresses to >256MB
    print("--- 2. Testing a decompression bomb ---")
    
    # We need to simulate a large file without storing it in memory
    class MockGzipBomb(io.BytesIO):
        def read(self, size=-1):
            # Decompress to a massive stream of zeros
            return b'\0' * (MAX_DECOMPRESSED_BYTES + 1)
            
    # Wrap the bomb in a GzipFile to make it a valid gzip stream for the check
    bomb_file = io.BytesIO()
    with gzip.GzipFile(fileobj=bomb_file, mode='wb') as zf:
        # Write a small amount of data that will be decompressed by our mock
        zf.write(b'bomb')
    
    bomb_file.seek(0)
    
    # Now, patch the gzip module to use our mock bomb reader
    original_gzip_file = gzip.GzipFile
    gzip.GzipFile = lambda fileobj: MockGzipBomb()

    # Run the fixed function on what appears to be a small file
    _open_fits_with_fix(bomb_file)

    # Restore original gzip behavior
    gzip.GzipFile = original_gzip_file