From df8036007f5122f3268e3182e8511adb50906741 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 1 Sep 2026 12:03:31 +0200 Subject: [PATCH 1/7] [Test] Update test to try to reproduce issue of proxy parsing on not complete data --- spec/inputs/tcp_spec.rb | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/spec/inputs/tcp_spec.rb b/spec/inputs/tcp_spec.rb index 1ff6c9b..af15914 100644 --- a/spec/inputs/tcp_spec.rb +++ b/spec/inputs/tcp_spec.rb @@ -135,10 +135,11 @@ def with_bound_port(host:"::", port:0, &block) } 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 + socket.write("\n") event_count.times do |i| # unicode smiley for testing unicode support! socket.puts("#{i} ☹") From 3184ecb6a539908a8054948b30d367e4f7933bcb Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 1 Sep 2026 12:14:13 +0200 Subject: [PATCH 2/7] Implemented HAProxy lines aggregator at Netty level --- .../org/logstash/tcp/ProxyLineAggregator.java | 64 ++++++++++++++++++ .../logstash/tcp/ProxyLineAggregatorTest.java | 65 +++++++++++++++++++ 2 files changed, 129 insertions(+) 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/src/main/java/org/logstash/tcp/ProxyLineAggregator.java b/src/main/java/org/logstash/tcp/ProxyLineAggregator.java new file mode 100644 index 0000000..f671d00 --- /dev/null +++ b/src/main/java/org/logstash/tcp/ProxyLineAggregator.java @@ -0,0 +1,64 @@ +package org.logstash.tcp; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.ByteToMessageDecoder; +import io.netty.util.ByteProcessor; + +import java.nio.charset.StandardCharsets; +import java.util.List; + + +/** + * Accumulate bytes until the HAProxy format v1 headline is present in the buffer. + * The line has format "PROXY....\r\n", this aggregator reject the buffer until it has that format, + * once reached passthrough every buffer as it is. + * This is needed because Ruby class DecoderImpl expect to have the full line when processing HAProxy protocol, and + * doesn't work with fragments. + * */ +public class ProxyLineAggregator extends ByteToMessageDecoder { + + private static final byte[] PROXY_PREFIX = "PROXY".getBytes(StandardCharsets.US_ASCII); + public static final int PROXY_LENGTH = PROXY_PREFIX.length; + + enum DecoderState {READ_PROXY, COMPLETED} + + private DecoderState state; + + public ProxyLineAggregator() { + this.state = DecoderState.READ_PROXY; + } + + @Override + protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List out) throws Exception { + switch (state) { + case READ_PROXY: + if (buffer.readableBytes() < PROXY_LENGTH) { + return; + } + if (!startsWithProxy(buffer)) { + state = DecoderState.COMPLETED; + out.add(buffer.readRetainedSlice(buffer.readableBytes())); + return; + } + if (buffer.forEachByte(ByteProcessor.FIND_CRLF) == -1) { + return; + } + state = DecoderState.COMPLETED; + out.add(buffer.readRetainedSlice(buffer.readableBytes())); + break; + case COMPLETED: + out.add(buffer.readRetainedSlice(buffer.readableBytes())); + break; + } + } + + 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 0000000..c7a8d17 --- /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(); + } +} From c361409b60890b4bfa37c4e5768f9314a29b6505 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 1 Sep 2026 15:06:50 +0200 Subject: [PATCH 3/7] [Test] offloaded in side thread the write and flush of part of the HA Proxy header line --- spec/inputs/tcp_spec.rb | 38 ++++++++++++++++++++++++++++++-------- 1 file changed, 30 insertions(+), 8 deletions(-) diff --git a/spec/inputs/tcp_spec.rb b/spec/inputs/tcp_spec.rb index af15914..89c586e 100644 --- a/spec/inputs/tcp_spec.rb +++ b/spec/inputs/tcp_spec.rb @@ -41,6 +41,19 @@ def with_bound_port(host:"::", port:0, &block) server.close end + def queue_pop_with_timeout(seconds, queue) + deadline = Time.now + seconds + item = nil + until item || Time.now >= deadline + begin + item = queue.pop(true) # raises ThreadError if empty + rescue ThreadError + sleep 0.1 + end + end + return item + end + let(:port) { find_available_port("127.0.0.1") } context "codec (PR #1372)" do @@ -139,15 +152,23 @@ def with_bound_port(host:"::", port:0, &block) socket = Stud::try(5.times) { TCPSocket.new("127.0.0.1", port) } socket.write("PROXY TCP4 1.2.3.4 5.6.7.8 1234 5678\r") socket.flush - socket.write("\n") - 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) @@ -675,6 +696,7 @@ def with_bound_port(host:"::", port:0, &block) context "when ssl_enabled is true" do let(:input) { subject } let(:queue) { Queue.new } + # let(:queue) { Thread::Queue.new } before(:each) do allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility) if defined?(ecs_compatibility) subject.register From ea983d52a0a13f0fec29b18370337c867c6849fb Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 1 Sep 2026 16:14:17 +0200 Subject: [PATCH 4/7] Add the proxy line aggreagtor codec to the Netty pipelines used on client's connections --- lib/logstash/inputs/tcp.rb | 3 ++- src/main/java/org/logstash/tcp/InputLoop.java | 11 ++++++++--- .../java/org/logstash/tcp/ProxyLineAggregator.java | 7 ++++++- 3 files changed, 16 insertions(+), 5 deletions(-) diff --git a/lib/logstash/inputs/tcp.rb b/lib/logstash/inputs/tcp.rb index 593a81d..a8570c3 100644 --- a/lib/logstash/inputs/tcp.rb +++ b/lib/logstash/inputs/tcp.rb @@ -179,7 +179,8 @@ def register if server? begin @logger.info("Binding tcp input listener", :address => "#{@host}:#{@port}", :ssl_enabled => @ssl_enabled) - @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) rescue java.net.BindException => bind_exception fail LogStash::ConfigurationError, "could not bind to #{@host}:#{@port}; #{bind_exception.message}" end diff --git a/src/main/java/org/logstash/tcp/InputLoop.java b/src/main/java/org/logstash/tcp/InputLoop.java index 5a6ff4a..e8e4cde 100644 --- a/src/main/java/org/logstash/tcp/InputLoop.java +++ b/src/main/java/org/logstash/tcp/InputLoop.java @@ -69,7 +69,7 @@ public final class InputLoop implements Runnable, Closeable { * @param keepAlive set to true to instruct the socket to issue TCP keep alive */ 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; @@ -80,7 +80,7 @@ public InputLoop(final String id, final String host, final int port, final Decod .option(ChannelOption.SO_BACKLOG, 1024) .option(ChannelOption.AUTO_READ, false) // do not auto-read until plugin has a queue .childOption(ChannelOption.SO_KEEPALIVE, keepAlive) - .childHandler(new InputLoop.InputHandler(decoder, sslContext)); + .childHandler(new InputLoop.InputHandler(decoder, sslContext, proxy)); try { serverChannel = serverBootstrap.bind(host, port).sync().channel(); @@ -131,14 +131,16 @@ private static final class InputHandler extends ChannelInitializer ou out.add(buffer.readRetainedSlice(buffer.readableBytes())); return; } - if (buffer.forEachByte(ByteProcessor.FIND_CRLF) == -1) { + if (!containsCrlf(buffer)) { return; } state = DecoderState.COMPLETED; @@ -53,6 +53,11 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List ou } } + 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]) { From aceb5e4a3ed237f7f52635ca6ec3cf8baf8041ed Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 1 Sep 2026 17:06:12 +0200 Subject: [PATCH 5/7] [Test] Add explicit fail if Queue.pop timeout elapses --- spec/inputs/tcp_spec.rb | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/spec/inputs/tcp_spec.rb b/spec/inputs/tcp_spec.rb index 89c586e..8697258 100644 --- a/spec/inputs/tcp_spec.rb +++ b/spec/inputs/tcp_spec.rb @@ -42,7 +42,8 @@ def with_bound_port(host:"::", port:0, &block) end def queue_pop_with_timeout(seconds, queue) - deadline = Time.now + seconds + start = Time.now + deadline = start + seconds item = nil until item || Time.now >= deadline begin @@ -51,6 +52,9 @@ def queue_pop_with_timeout(seconds, queue) sleep 0.1 end end + if item == nil + raise "Elapsed #{Time.now - start} seconds before pop a value" + end return item end From 2ced7a105a66a5b7333f4321a105e764ae0ba1d2 Mon Sep 17 00:00:00 2001 From: Andrea Selva Date: Fri, 4 Sep 2026 08:45:38 +0200 Subject: [PATCH 6/7] Apply batched suggestions from code review Co-authored-by: Cas Donoghue --- spec/inputs/tcp_spec.rb | 1 - src/main/java/org/logstash/tcp/InputLoop.java | 2 ++ 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/spec/inputs/tcp_spec.rb b/spec/inputs/tcp_spec.rb index 8697258..44d5242 100644 --- a/spec/inputs/tcp_spec.rb +++ b/spec/inputs/tcp_spec.rb @@ -700,7 +700,6 @@ def queue_pop_with_timeout(seconds, queue) context "when ssl_enabled is true" do let(:input) { subject } let(:queue) { Queue.new } - # let(:queue) { Thread::Queue.new } before(:each) do allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility) if defined?(ecs_compatibility) subject.register diff --git a/src/main/java/org/logstash/tcp/InputLoop.java b/src/main/java/org/logstash/tcp/InputLoop.java index e8e4cde..240d322 100644 --- a/src/main/java/org/logstash/tcp/InputLoop.java +++ b/src/main/java/org/logstash/tcp/InputLoop.java @@ -67,6 +67,8 @@ 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 boolean proxy) { From b8266b2275744188bcad60924ae159fcc494a40e Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 4 Sep 2026 08:56:51 +0200 Subject: [PATCH 7/7] Simplified the ProxyLineAggregator, when the accumulated line match the proxy, it's passed through and the handler removes itself from the channel's pipeline --- .../org/logstash/tcp/ProxyLineAggregator.java | 47 ++++++------------- 1 file changed, 15 insertions(+), 32 deletions(-) diff --git a/src/main/java/org/logstash/tcp/ProxyLineAggregator.java b/src/main/java/org/logstash/tcp/ProxyLineAggregator.java index 918a1ce..44355ba 100644 --- a/src/main/java/org/logstash/tcp/ProxyLineAggregator.java +++ b/src/main/java/org/logstash/tcp/ProxyLineAggregator.java @@ -10,47 +10,30 @@ /** - * Accumulate bytes until the HAProxy format v1 headline is present in the buffer. - * The line has format "PROXY....\r\n", this aggregator reject the buffer until it has that format, - * once reached passthrough every buffer as it is. - * This is needed because Ruby class DecoderImpl expect to have the full line when processing HAProxy protocol, and - * doesn't work with fragments. + * The line has format "PROXY....\r\n"; this aggregator holds back the buffer until that full + * line is available, then passes it through and removes itself from the pipeline. + * This is needed because Ruby class DecoderImpl expects the full line when processing the HAProxy + * protocol and doesn't work with fragments. * */ public class ProxyLineAggregator extends ByteToMessageDecoder { private static final byte[] PROXY_PREFIX = "PROXY".getBytes(StandardCharsets.US_ASCII); public static final int PROXY_LENGTH = PROXY_PREFIX.length; - enum DecoderState {READ_PROXY, COMPLETED} - - private DecoderState state; - - public ProxyLineAggregator() { - this.state = DecoderState.READ_PROXY; - } - @Override protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List out) throws Exception { - switch (state) { - case READ_PROXY: - if (buffer.readableBytes() < PROXY_LENGTH) { - return; - } - if (!startsWithProxy(buffer)) { - state = DecoderState.COMPLETED; - out.add(buffer.readRetainedSlice(buffer.readableBytes())); - return; - } - if (!containsCrlf(buffer)) { - return; - } - state = DecoderState.COMPLETED; - out.add(buffer.readRetainedSlice(buffer.readableBytes())); - break; - case COMPLETED: - out.add(buffer.readRetainedSlice(buffer.readableBytes())); - break; + // 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) {