From cbd1b4be7343e188ea77f548b06c40117ec7951d Mon Sep 17 00:00:00 2001 From: Shizuo Fujita Date: Mon, 28 Sep 2026 11:04:47 +0900 Subject: [PATCH] Merge commit from fork The message_length_limit parameter was previously only applied to UDP. This change enforces the limit on TCP/TLS connections as well. With frame_type traditional, drop messages larger than the limit and clear the buffer when the size is exceeded. This discards the rest of the oversized message up to the next delimiter, matching the behavior of in_tcp. The connection is kept open so that subsequent well-formed messages are still handled. The flag to discard data is maintained per connection. With frame_type octet_count, a length-framed stream cannot be resynchronized once its framing becomes untrustworthy. Therefore, close the connection if the declared length exceeds the limit, if the declared length is not a valid number, or if the pending buffer exceeds the limit without containing a delimiter to parse a length from. A partially received frame always has a delimiter after its length, so a legitimate near-limit frame is not closed even though it can slightly exceed the limit with its length header. Additionally, raise the default message_length_limit from 2048 to 8192 to match the MaxMessageSize default of rsyslog. This ensures that typical rsyslog traffic over TCP is not silently dropped now that the limit is enforced. This also relaxes the receive size of UDP, which remains backward compatible as it simply accepts larger payloads. Signed-off-by: Shizuo Fujita --- lib/fluent/plugin/in_syslog.rb | 74 ++++++++++++- test/plugin/test_in_syslog.rb | 185 +++++++++++++++++++++++++++++++++ 2 files changed, 255 insertions(+), 4 deletions(-) diff --git a/lib/fluent/plugin/in_syslog.rb b/lib/fluent/plugin/in_syslog.rb index 3d4b205c98..4c453307ef 100644 --- a/lib/fluent/plugin/in_syslog.rb +++ b/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 diff --git a/test/plugin/test_in_syslog.rb b/test/plugin/test_in_syslog.rb index 868e77bb04..9ab129c7cf 100755 --- a/test/plugin/test_in_syslog.rb +++ b/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), + "\n", + "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