Merge commit from fork

* buffer, in_http: enforce size limits on decompressed payloads

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

* Classify SizeLimitError as unrecoverable

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

* in_forward: add decompression_size_limit

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

* event: Fix constant namespace

Fix following NameError:

```
/lib/fluent/event.rb:271:in 'Fluent::CompressedMessagePackEventStream#initialize': uninitialized constant Fluent::CompressedMessagePackEventStream::Compressable (NameError)

... unpacked_times: nil, unpacked_records: nil, compress: :gzip, decompression_size_limit: Compressable::DEFAULT_DECOMPRESSION_SIZE_LIMIT)
                                                                                           ^^^^^^^^^^^^
```

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

* extractor: Add INFLATE_BYTES_TO_READ

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

---------

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/event.rb

@@ -268,11 +268,12 @@ def to_msgpack_stream(time_int: false, packer: nil)
   end
 
   class CompressedMessagePackEventStream < MessagePackEventStream
-    def initialize(data, cached_unpacker = nil, size = 0, unpacked_times: nil, unpacked_records: nil, compress: :gzip)
+    def initialize(data, cached_unpacker = nil, size = 0, unpacked_times: nil, unpacked_records: nil, compress: :gzip, decompression_size_limit: Fluent::Plugin::Compressable::DEFAULT_DECOMPRESSION_SIZE_LIMIT)
       super(data, cached_unpacker, size, unpacked_times: unpacked_times, unpacked_records: unpacked_records)
       @decompressed_data = nil
       @compressed_data = data
       @type = compress
+      @decompression_size_limit = decompression_size_limit
     end
 
     def empty?

lib/fluent/plugin/buf_file.rb

@@ -205,7 +205,7 @@ def resume
       def generate_chunk(metadata)
         # FileChunk generates real path with unique_id
         perm = @file_permission || system_config.file_permission
-        chunk = Fluent::Plugin::Buffer::FileChunk.new(metadata, @path, :create, perm: perm, compress: @compress)
+        chunk = Fluent::Plugin::Buffer::FileChunk.new(metadata, @path, :create, perm: perm, compress: @compress, decompression_size_limit: @decompression_size_limit)
         log.debug "Created new chunk", chunk_id: dump_unique_id_hex(chunk.unique_id), metadata: metadata
 
         return chunk

lib/fluent/plugin/buf_file_single.rb

@@ -183,7 +183,7 @@ def resume
           end
 
           begin
-            chunk = Fluent::Plugin::Buffer::FileSingleChunk.new(m, path, mode, @key_in_path, compress: @compress)
+            chunk = Fluent::Plugin::Buffer::FileSingleChunk.new(m, path, mode, @key_in_path, compress: @compress, decompression_size_limit: @decompression_size_limit)
             chunk.restore_size(@chunk_format) if @calc_num_records
           rescue Fluent::Plugin::Buffer::FileSingleChunk::FileChunkError => e
             exist_broken_file = true
@@ -216,7 +216,7 @@ def resume
       def generate_chunk(metadata)
         # FileChunk generates real path with unique_id
         perm = @file_permission || system_config.file_permission
-        chunk = Fluent::Plugin::Buffer::FileSingleChunk.new(metadata, @path, :create, @key_in_path, perm: perm, compress: @compress)
+        chunk = Fluent::Plugin::Buffer::FileSingleChunk.new(metadata, @path, :create, @key_in_path, perm: perm, compress: @compress, decompression_size_limit: @decompression_size_limit)
 
         log.debug "Created new chunk", chunk_id: dump_unique_id_hex(chunk.unique_id), metadata: metadata
 

lib/fluent/plugin/buf_memory.rb

@@ -27,7 +27,7 @@ def resume
       end
 
       def generate_chunk(metadata)
-        Fluent::Plugin::Buffer::MemoryChunk.new(metadata, compress: @compress)
+        Fluent::Plugin::Buffer::MemoryChunk.new(metadata, compress: @compress, decompression_size_limit: @decompression_size_limit)
       end
     end
   end

lib/fluent/plugin/buffer.rb

@@ -66,6 +66,9 @@ class BufferChunkOverflowError < BufferError; end # A record size is larger than
       desc 'Compress buffered data.'
       config_param :compress, :enum, list: [:text, :gzip, :zstd], default: :text
 
+      desc 'The size limit of the decompressed element.'
+      config_param :decompression_size_limit, :size, default: Compressable::DEFAULT_DECOMPRESSION_SIZE_LIMIT
+
       desc 'If true, chunks are thrown away when unrecoverable error happens'
       config_param :disable_chunk_backup, :bool, default: false
 

lib/fluent/plugin/buffer/chunk.rb

@@ -48,7 +48,7 @@ class Chunk
 
         # TODO: CompressedPackedMessage of forward protocol?
 
-        def initialize(metadata, compress: :text)
+        def initialize(metadata, compress: :text, decompression_size_limit: Compressable::DEFAULT_DECOMPRESSION_SIZE_LIMIT)
           super()
           @unique_id = generate_unique_id
           @metadata = metadata
@@ -64,6 +64,7 @@ def initialize(metadata, compress: :text)
           elsif compress == :zstd
             extend ZstdDecompressable
           end
+          @decompression_size_limit = decompression_size_limit
         end
 
         attr_reader :unique_id, :metadata, :state

lib/fluent/plugin/buffer/file_chunk.rb

@@ -39,8 +39,8 @@ class FileChunkError < StandardError; end
 
         attr_reader :path, :meta_path, :permission
 
-        def initialize(metadata, path, mode, perm: nil, compress: :text)
-          super(metadata, compress: compress)
+        def initialize(metadata, path, mode, perm: nil, compress: :text, decompression_size_limit: Compressable::DEFAULT_DECOMPRESSION_SIZE_LIMIT)
+          super(metadata, compress: compress, decompression_size_limit: decompression_size_limit)
           perm ||= Fluent::DEFAULT_FILE_PERMISSION
           @permission = perm.is_a?(String) ? perm.to_i(8) : perm
           @bytesize = @size = @adding_bytes = @adding_size = 0

lib/fluent/plugin/buffer/file_single_chunk.rb

@@ -35,8 +35,8 @@ class FileChunkError < StandardError; end
 
         attr_reader :path, :permission
 
-        def initialize(metadata, path, mode, key, perm: Fluent::DEFAULT_FILE_PERMISSION, compress: :text)
-          super(metadata, compress: compress)
+        def initialize(metadata, path, mode, key, perm: Fluent::DEFAULT_FILE_PERMISSION, compress: :text, decompression_size_limit: Compressable::DEFAULT_DECOMPRESSION_SIZE_LIMIT)
+          super(metadata, compress: compress, decompression_size_limit: decompression_size_limit)
           @key = key
           perm ||= Fluent::DEFAULT_FILE_PERMISSION
           @permission = perm.is_a?(String) ? perm.to_i(8) : perm

lib/fluent/plugin/buffer/memory_chunk.rb

@@ -20,7 +20,7 @@ module Fluent
   module Plugin
     class Buffer
       class MemoryChunk < Chunk
-        def initialize(metadata, compress: :text)
+        def initialize(metadata, compress: :text, decompression_size_limit: Compressable::DEFAULT_DECOMPRESSION_SIZE_LIMIT)
           super
           @chunk = ''.force_encoding(Encoding::ASCII_8BIT)
           @chunk_bytes = 0

lib/fluent/plugin/compressable.rb

@@ -14,13 +14,16 @@
 #    limitations under the License.
 #
 
+require 'fluent/plugin/extractor'
 require 'stringio'
 require 'zlib'
 require 'zstd-ruby'
 
 module Fluent
   module Plugin
     module Compressable
+      DEFAULT_DECOMPRESSION_SIZE_LIMIT = 256 * 1024 * 1024
+
       def compress(data, type: :gzip, **kwargs)
         output_io = kwargs[:output_io]
         io = output_io || StringIO.new
@@ -60,79 +63,21 @@ def decompress(compressed_data = nil, output_io: nil, input_io: nil, type: :gzip
 
       private
 
-      def string_decompress_gzip(compressed_data)
-        io = StringIO.new(compressed_data)
-        out = ''
-        loop do
-          reader = Zlib::GzipReader.new(io)
-          out << reader.read
-          unused = reader.unused
-          reader.finish
-          unless unused.nil?
-            adjust = unused.length
-            io.pos -= adjust
-          end
-          break if io.eof?
-        end
-        out
-      end
-
-      def string_decompress_zstd(compressed_data)
-        io = StringIO.new(compressed_data)
-        reader = Zstd::StreamReader.new(io)
-        out = ''
-        loop do
-          # Zstd::StreamReader needs to specify the size of the buffer
-          out << reader.read(1024)
-          # Zstd::StreamReader doesn't provide unused data, so we have to manually adjust the position
-          break if io.eof?
-        end
-        out
-      end
-
       def string_decompress(compressed_data, type = :gzip)
         if type == :gzip
-          string_decompress_gzip(compressed_data)
+          Extractor.decompress_gzip(compressed_data, limit: @decompression_size_limit || DEFAULT_DECOMPRESSION_SIZE_LIMIT)
         elsif type == :zstd
-          string_decompress_zstd(compressed_data)
+          Extractor.decompress_zstd(compressed_data, limit: @decompression_size_limit || DEFAULT_DECOMPRESSION_SIZE_LIMIT)
         else
           raise ArgumentError, "Unknown compression type: #{type}"
         end
       end
 
-      def io_decompress_gzip(input, output)
-        loop do
-          reader = Zlib::GzipReader.new(input)
-          v = reader.read
-          output.write(v)
-          unused = reader.unused
-          reader.finish
-          unless unused.nil?
-            adjust = unused.length
-            input.pos -= adjust
-          end
-          break if input.eof?
-        end
-        output
-      end
-
-      def io_decompress_zstd(input, output)
-        reader = Zstd::StreamReader.new(input)
-        loop do
-          # Zstd::StreamReader needs to specify the size of the buffer
-          v = reader.read(1024)
-          output.write(v)
-          # Zstd::StreamReader doesn't provide unused data, so we have to manually adjust the position
-          break if input.eof?
-        end
-        output
-      end
-
       def io_decompress(input, output, type = :gzip)
         if type == :gzip
-          io_decompress_gzip(input, output)
+          Extractor.io_decompress_gzip(input, output)
         elsif type == :zstd
-          io_decompress_zstd(input, output)
+          Extractor.io_decompress_zstd(input, output)
         else
           raise ArgumentError, "Unknown compression type: #{type}"
         end