From 8d426bf99595ea6f05c680a67892a16099566559 Mon Sep 17 00:00:00 2001 From: Andrea Selva Date: Mon, 7 Sep 2026 11:05:03 +0200 Subject: [PATCH] Aggregate HA proxy line before parsing it (#286) Avoids error that could rise if the TCP stack decides to break the HA Proxy v1 headline before the newline terminator in multiple byte arrays. Adds an aggregator to collect all the bytes up to the first newline before move on to the decoding part done by DecoderImpl class. This message decoder is added only if proxy setting is enabled. Co-authored-by: Cas Donoghue --- lib/logstash/inputs/tcp.rb | 2 +- spec/inputs/tcp_spec.rb | 44 ++++++++++--- src/main/java/org/logstash/tcp/InputLoop.java | 13 +++- .../org/logstash/tcp/ProxyLineAggregator.java | 52 +++++++++++++++ .../logstash/tcp/ProxyLineAggregatorTest.java | 65 +++++++++++++++++++ 5 files changed, 163 insertions(+), 13 deletions(-) create mode 100644 src/main/java/org/logstash/tcp/ProxyLineAggregator.java create mode 100644 src/test/java/org/logstash/tcp/ProxyLineAggregatorTest.java diff --git a/lib/logstash/inputs/tcp.rb b/lib/logstash/inputs/tcp.rb index 897f46bc..bd96b706 100644 --- a/lib/logstash/inputs/tcp.rb +++ b/lib/logstash/inputs/tcp.rb @@ -181,7 +181,7 @@ def register validate_ssl_config! if server? - @loop = InputLoop.new(@id, @host, @port, DecoderImpl.new(@codec, self), @tcp_keep_alive, java_ssl_context) + @loop = InputLoop.new(@id, @host, @port, DecoderImpl.new(@codec, self), @tcp_keep_alive, java_ssl_context, @proxy_protocol) end end diff --git a/spec/inputs/tcp_spec.rb b/spec/inputs/tcp_spec.rb index 4aa04875..c7c2da3a 100644 --- a/spec/inputs/tcp_spec.rb +++ b/spec/inputs/tcp_spec.rb @@ -35,6 +35,23 @@ def get_port let(:port) { get_port } + def queue_pop_with_timeout(seconds, queue) + start = Time.now + deadline = start + seconds + item = nil + until item || Time.now >= deadline + begin + item = queue.pop(true) # raises ThreadError if empty + rescue ThreadError + sleep 0.1 + end + end + if item == nil + raise "Elapsed #{Time.now - start} seconds before pop a value" + end + return item + end + context "codec (PR #1372)" do it "switches from plain to line" do require "logstash/codecs/plain" @@ -108,18 +125,27 @@ def get_port } CONFIG - events = input(conf) do |pipeline, queue| + events = input(conf) do |_, queue| socket = Stud::try(5.times) { TCPSocket.new("127.0.0.1", port) } - socket.puts("PROXY TCP4 1.2.3.4 5.6.7.8 1234 5678\r"); + socket.write("PROXY TCP4 1.2.3.4 5.6.7.8 1234 5678\r") socket.flush - event_count.times do |i| - # unicode smiley for testing unicode support! - socket.puts("#{i} ☹") - socket.flush + # proceed with the rest of data writing in a separate thread + # so that crates some interleaving in the buffers seen on network layer + Thread.new do + sleep(1) + begin + socket.write("\n") + event_count.times do |i| + # unicode smiley for testing unicode support! + socket.puts("#{i} ☹") + socket.flush + end + socket.close + rescue => e + puts "Error while writing to the socket #{e.backtrace}" + end end - socket.close - - event_count.times.collect {queue.pop} + event_count.times.collect { queue_pop_with_timeout(5, queue) } end expect(events.length).to eq(event_count) diff --git a/src/main/java/org/logstash/tcp/InputLoop.java b/src/main/java/org/logstash/tcp/InputLoop.java index 86342d10..e2b914cf 100644 --- a/src/main/java/org/logstash/tcp/InputLoop.java +++ b/src/main/java/org/logstash/tcp/InputLoop.java @@ -67,9 +67,11 @@ public final class InputLoop implements Runnable, Closeable { * @param port Port to listen on * @param decoder {@link Decoder} provided by Jruby * @param keepAlive set to true to instruct the socket to issue TCP keep alive + * @param sslContext SSL configuration, or null when TLS is disabled + * @param proxy set to true to aggregate the HAProxy v1 header line before decoding */ public InputLoop(final String id, final String host, final int port, final Decoder decoder, final boolean keepAlive, - final SslContext sslContext) { + final SslContext sslContext, final boolean proxy) { this.sslContext = sslContext; this.host = host; this.port = port; @@ -79,7 +81,7 @@ public InputLoop(final String id, final String host, final int port, final Decod .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(ChannelOption.SO_KEEPALIVE, keepAlive) - .childHandler(new InputLoop.InputHandler(decoder, sslContext)); + .childHandler(new InputLoop.InputHandler(decoder, sslContext, proxy)); } @Override @@ -120,14 +122,16 @@ private static final class InputHandler extends ChannelInitializer out) throws Exception { + // Wait until we can decide: enough bytes to match the prefix, and if it is a PROXY + // line, the terminating \r\n must be present. + if (buffer.readableBytes() < PROXY_LENGTH) { + return; + } + if (startsWithProxy(buffer) && !containsCrlf(buffer)) { + return; + } + // Full PROXY line, or non-PROXY data: pass everything through and drop this handler + // so subsequent reads skip the aggregator entirely. + out.add(buffer.readRetainedSlice(buffer.readableBytes())); + ctx.pipeline().remove(this); + } + + private static boolean containsCrlf(ByteBuf buffer) { + int lfIndex = buffer.forEachByte(ByteProcessor.FIND_LF); + return lfIndex > 0 && buffer.getByte(lfIndex - 1) == '\r'; + } + + private static boolean startsWithProxy(ByteBuf buffer) { + for (int i = 0; i < PROXY_LENGTH; i++) { + if (buffer.getByte(buffer.readerIndex() + i) != PROXY_PREFIX[i]) { + return false; + } + } + return true; + } +} diff --git a/src/test/java/org/logstash/tcp/ProxyLineAggregatorTest.java b/src/test/java/org/logstash/tcp/ProxyLineAggregatorTest.java new file mode 100644 index 00000000..c7a8d172 --- /dev/null +++ b/src/test/java/org/logstash/tcp/ProxyLineAggregatorTest.java @@ -0,0 +1,65 @@ +package org.logstash.tcp; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.nullValue; + +class ProxyLineAggregatorTest { + + private EmbeddedChannel channel; + + @BeforeEach + void setUp() { + channel = new EmbeddedChannel(new ProxyLineAggregator()); + } + + @AfterEach + void tearDown() { + channel.finishAndReleaseAll(); + } + + private static ByteBuf ascii(String s) { + return Unpooled.copiedBuffer(s, StandardCharsets.US_ASCII); + } + + @Test + void partialProxyHeader_producesNoOutput() { + // "PRO" is shorter than the 5-byte "PROXY" prefix — decoder must wait + channel.writeInbound(ascii("PRO")); + assertThat(channel.readInbound(), nullValue()); + } + + @Test + void proxyPrefixWithoutCRLF_producesNoOutput() { + // Full PROXY keyword is present but the line terminator \r\n is missing + channel.writeInbound(ascii("PROXY TCP4 192.168.1.1 10.0.0.1 1234 80")); + assertThat(channel.readInbound(), nullValue()); + } + + @Test + void nonProxyDataWithSufficientBytes_passesThrough() { + // Content >= PROXY_LENGTH bytes but not a PROXY header and no \r\n: + // the aggregator should recognise it is not a PROXY line and let it through + channel.writeInbound(ascii("Hello, World!")); + ByteBuf result = channel.readInbound(); + assertThat(result, notNullValue()); + result.release(); + } + + @Test + void completeProxyLineWithCRLF_forwardsBuffer() { + channel.writeInbound(ascii("PROXY TCP4 192.168.1.1 10.0.0.1 1234 80\r\n")); + ByteBuf result = channel.readInbound(); + assertThat(result, notNullValue()); + result.release(); + } +}