Backport(v1.19): in_syslog: enforce message_length_limit on TCP/TLS transport

lib/fluent/plugin/in_syslog.rb

@@ -98,8 +98,8 @@ class SyslogInput < Input
     desc 'The field name of the facility.'
     config_param :facility_key, :string, default: nil
 
-    desc "The max bytes of message"
-    config_param :message_length_limit, :size, default: 2048
+    desc "The max bytes of message. The messages exceeding this limit are dropped (UDP, TCP/TLS with traditional frame_type) or the connection is closed (TCP/TLS with octet_count frame_type)."
+    config_param :message_length_limit, :size, default: 8192
 
     config_param :blocking_timeout, :time, default: 0.5
 
@@ -190,29 +190,95 @@ def start_tcp_server(tls: false)
         resolve_name: @resolve_hostname,
         send_keepalive_packet: @send_keepalive_packet
       ) do |conn|
+        discard_till_next_delimiter = false
         conn.data do |data|
           buffer = conn.buffer
           buffer << data
           pos = 0
           if octet_count_frame
+            # Once an octet-counted stream contains an untrustworthy frame header, the stream
+            # can't be resynchronized. So close the connection instead of skipping bytes.
+            close_connection = false
             while idx = buffer.index(delimiter, pos)
-              num = Integer(buffer[pos..idx])
+              begin
+                num = Integer(buffer[pos..idx])
+              rescue ArgumentError
+                log.info "The message length is not a number, closing connection"
+                log.on_debug { log.debug "The message length is not a number:", head: buffer[pos, 32].dump }
+                close_connection = true
+                break
+              end
+
+              if num < 0
+                log.info "The message length is negative, closing connection:", size: num
+                close_connection = true
+                break
+              end
+
+              if num > @message_length_limit
+                log.info "The message length exceeds 'message_length_limit', closing connection:", limit: @message_length_limit, size: num
+                close_connection = true
+                break
+              end
+
               msg = buffer[idx + delimiter_size, num]
               if msg.size != num
+                # The frame is not fully received yet.
                 break
               end
 
               pos = idx + delimiter_size + num
               message_handler(msg, conn)
             end
+            buffer.slice!(0, pos) if pos > 0
+
+            # If the remaining buffer has no delimiter, no message length can be parsed from it.
+            # Since a valid frame header is much shorter than the limit, such a buffer never
+            # becomes a valid frame. (Note that a partially received valid frame always has a
+            # delimiter after its length, so this doesn't drop it even if it slightly exceeds
+            # the limit with its length header.)
+            if !close_connection && !buffer.index(delimiter) && buffer.bytesize > @message_length_limit
+              log.info "The buffer size exceeds 'message_length_limit', closing connection:", limit: @message_length_limit, size: buffer.bytesize
+              log.on_debug { log.debug "The buffer size exceeds 'message_length_limit':", head: buffer[...32] }
+              close_connection = true
+            end
+
+            if close_connection
+              buffer.clear
+              conn.close
+              next
+            end
           else
             while idx = buffer.index(delimiter, pos)
               msg = buffer[pos...idx]
               pos = idx + delimiter_size
+
+              if discard_till_next_delimiter
+                discard_till_next_delimiter = false
+                next
+              end
+
+              if msg.bytesize > @message_length_limit
+                log.info "The received data is larger than 'message_length_limit', dropped:", limit: @message_length_limit, size: msg.bytesize
+                log.on_debug { log.debug "The received data is larger than 'message_length_limit':", head: msg[...32] }
+                next
+              end
+
               message_handler(msg, conn)
             end
+            buffer.slice!(0, pos) if pos > 0
+
+            # If the buffer size exceeds the limit here, it means that the next message will
+            # definitely exceed the limit. So we should clear the buffer here. Otherwise, it will
+            # keep storing useless data until the next delimiter comes.
+            if buffer.bytesize > @message_length_limit
+              log.info "The buffer size exceeds 'message_length_limit', cleared:", limit: @message_length_limit, size: buffer.bytesize
+              log.on_debug { log.debug "The buffer size exceeds 'message_length_limit':", head: buffer[...32] }
+              buffer.clear
+              # We should discard the subsequent data until the next delimiter comes.
+              discard_till_next_delimiter = true
+            end
           end
-          buffer.slice!(0, pos) if pos > 0
         end
       end
     end

test/plugin/test_in_syslog.rb

@@ -509,4 +509,189 @@ def test_send_keepalive_packet_can_not_be_enabled_for_udp
       ])
     end
   end
+
+  sub_test_case 'message_length_limit for tcp' do
+    LIMIT = 1024
+    NORMAL_MESSAGE = '<6>Sep 10 00:00:00 localhost logger: hello'
+
+    def create_tcp_driver(port, frame_type: :traditional)
+      create_driver([
+        ipv4_config(port),
+        "<transport tcp>\n</transport>",
+        "frame_type #{frame_type}",
+        "message_length_limit #{LIMIT}",
+      ].join("\n"))
+    end
+
+    def max_buffer_size(d)
+      conns = d.instance.instance_variable_get(:@_server_connections)
+      conns.filter_map { |conn|
+        conn.instance_variable_get(:@callback_connection)&.buffer&.bytesize
+      }.max || 0
+    end
+
+    # The plugin closes the connection by sending RST (SO_LINGER 0), so both EOF and
+    # ECONNRESET mean that the connection was closed by the plugin.
+    def wait_until_closed(sock, timeout: 10)
+      waiting(timeout) do
+        loop do
+          begin
+            return true if sock.read_nonblock(1024).nil?
+          rescue EOFError, Errno::ECONNRESET
+            return true
+          rescue IO::WaitReadable
+            IO.select([sock], nil, nil, 0.1)
+          end
+        end
+      end
+    rescue Timeout::Error
+      false
+    end
+
+    def test_default_message_length_limit
+      assert_equal 8192, create_driver(ipv4_config).instance.message_length_limit
+    end
+
+    test 'traditional: the buffer does not grow unboundedly without delimiter' do
+      port = unused_port(protocol: :tcp)
+      d = create_tcp_driver(port)
+      observed_max = 0
+
+      d.run(expect_emits: 1, timeout: 30) do
+        TCPSocket.open('127.0.0.1', port) do |s|
+          # Send 200KB without any delimiter.
+          100.times do
+            s.write('x' * 2048)
+            s.flush
+            sleep 0.01
+            size = max_buffer_size(d)
+            observed_max = size if size > observed_max
+          end
+
+          # The tail of the oversized data is discarded up to the next delimiter,
+          # and the subsequent message must be still handled.
+          s.write("\n#{NORMAL_MESSAGE}\n")
+          s.flush
+          waiting(10) { sleep 0.1 until d.events.size >= 1 }
+        end
+      end
+
+      assert do
+        observed_max <= 32 * 1024 # much smaller than the 200KB we sent
+      end
+      assert_equal 1, d.events.size
+      assert_equal 'hello', d.events[0][2]['message']
+    end
+
+    test 'traditional: a message larger than the limit is dropped and the next one is emitted' do
+      port = unused_port(protocol: :tcp)
+      d = create_tcp_driver(port)
+
+      d.run(expect_emits: 1, timeout: 20) do
+        TCPSocket.open('127.0.0.1', port) do |s|
+          s.write("<6>Sep 10 00:00:00 localhost logger: #{'x' * (LIMIT * 2)}\n")
+          s.write("#{NORMAL_MESSAGE}\n")
+          s.flush
+          waiting(10) { sleep 0.1 until d.events.size >= 1 }
+        end
+      end
+
+      assert_equal 1, d.events.size
+      assert_equal 'hello', d.events[0][2]['message']
+    end
+
+    test 'traditional: a message split into multiple chunks is reassembled' do
+      port = unused_port(protocol: :tcp)
+      d = create_tcp_driver(port)
+
+      d.run(expect_emits: 1, timeout: 20) do
+        TCPSocket.open('127.0.0.1', port) do |s|
+          NORMAL_MESSAGE.each_char do |c|
+            s.write(c)
+            s.flush
+          end
+          s.write("\n")
+          s.flush
+        end
+      end
+
+      assert_equal 1, d.events.size
+      assert_equal 'hello', d.events[0][2]['message']
+    end
+
+    test 'octet_count: a declared length larger than the limit closes the connection' do
+      port = unused_port(protocol: :tcp)
+      d = create_tcp_driver(port, frame_type: :octet_count)
+      closed = false
+
+      d.run(timeout: 20) do
+        TCPSocket.open('127.0.0.1', port) do |s|
+          s.write("#{LIMIT + 1} #{NORMAL_MESSAGE}")
+          s.flush
+          closed = wait_until_closed(s)
+        end
+      end
+      assert_true closed
+      assert_equal 0, d.events.size
+    end
+
+    test 'octet_count: data without delimiter beyond the limit closes the connection' do
+      port = unused_port(protocol: :tcp)
+      d = create_tcp_driver(port, frame_type: :octet_count)
+      closed = false
+
+      d.run(timeout: 20) do
+        TCPSocket.open('127.0.0.1', port) do |s|
+          s.write('x' * (LIMIT * 2)) # no space delimiter at all
+          s.flush
+          closed = wait_until_closed(s)
+        end
+      end
+      assert_true closed
+      assert_equal 0, d.events.size
+    end
+
+    test 'octet_count: a message split into multiple chunks is reassembled' do
+      port = unused_port(protocol: :tcp)
+      d = create_tcp_driver(port, frame_type: :octet_count)
+
+      d.run(expect_emits: 1, timeout: 20) do
+        TCPSocket.open('127.0.0.1', port) do |s|
+          s.write("#{NORMAL_MESSAGE.size} ")
+          s.flush
+          sleep 0.1
+          s.write(NORMAL_MESSAGE[0...10])
+          s.flush
+          sleep 0.1
+          s.write(NORMAL_MESSAGE[10..-1])
+          s.flush
+        end
+      end
+
+      assert_equal 1, d.events.size
+      assert_equal 'hello', d.events[0][2]['message']
+    end
+
+    test 'octet_count: a near-limit message is not treated as an attack' do
+      port = unused_port(protocol: :tcp)
+      d = create_tcp_driver(port, frame_type: :octet_count)
+      message = "<6>Sep 10 00:00:00 localhost logger: #{'x' * (LIMIT - 37)}"
+      assert_equal LIMIT, message.size
+
+      d.run(expect_emits: 1, timeout: 20) do
+        TCPSocket.open('127.0.0.1', port) do |s|
+          # The length header + this partial frame (1025 bytes) exceeds the limit,
+          # but it must not be treated as an attack because the frame is legitimate.
+          s.write("#{message.size} #{message[0...1020]}")
+          s.flush
+          sleep 0.1
+          s.write(message[1020..-1])
+          s.flush
+        end
+      end
+
+      assert_equal 1, d.events.size
+      assert_equal 'x' * (LIMIT - 37), d.events[0][2]['message']
+    end
+  end
 end