Harden beats packet parsing (#26139) (#26165)

* add hard upper bounds to declared frame sizes when parsing beats packets

* add changelog

(cherry picked from commit 562b88a)

Co-authored-by: Kay Roepke <kroepke@googlemail.com>

Author: graylog-internal-actions-access[bot]

changelog/unreleased/pr-26139.toml

@@ -0,0 +1,4 @@
+type = "f"
+message = "Add safeguards for beats protocol frame parsing."
+
+pulls = ["26139"]

graylog2-server/src/main/java/org/graylog/plugins/beats/BeatsFrameDecoder.java

@@ -64,6 +64,10 @@ enum DecodingState {
         FRAME_WINDOW_SIZE
     }
 
+    private static final int MAX_JSON_FRAME_BYTES = 16 * 1024 * 1024;
+    private static final int MAX_DATA_PAIRS = 1024;
+    private static final int MAX_DATA_ITEM_BYTES = 1024 * 1024;
+
     private long windowSize;
     private long sequenceNum;
 
@@ -165,6 +169,9 @@ private Collection<ByteBuf> parseJsonFrame(Channel channel, ByteBuf channelBuffe
         LOG.trace("Received sequence number {}", sequenceNum);
 
         final int jsonLength = Ints.saturatedCast(channelBuffer.readUnsignedInt());
+        if (jsonLength > MAX_JSON_FRAME_BYTES) {
+            throw new IllegalStateException("JSON frame size " + jsonLength + " exceeds maximum " + MAX_JSON_FRAME_BYTES);
+        }
 
         final ByteBuf buffer = channelBuffer.readBytes(jsonLength);
         sendACK(channel);
@@ -211,6 +218,9 @@ private Collection<ByteBuf> parseDataFrame(Channel channel, ByteBuf channelBuffe
         LOG.trace("Received sequence number {}", sequenceNum);
 
         final int pairs = Ints.saturatedCast(channelBuffer.readUnsignedInt());
+        if (pairs > MAX_DATA_PAIRS) {
+            throw new IllegalStateException("Data frame pair count " + pairs + " exceeds maximum " + MAX_DATA_PAIRS);
+        }
         final JsonFactory jsonFactory = new JsonFactory();
         final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
         try (final JsonGenerator jg = jsonFactory.createGenerator(outputStream)) {
@@ -231,6 +241,9 @@ private Collection<ByteBuf> parseDataFrame(Channel channel, ByteBuf channelBuffe
 
     private String parseDataItem(ByteBuf buf) {
         int length = Ints.saturatedCast(buf.readUnsignedInt());
+        if (length > MAX_DATA_ITEM_BYTES) {
+            throw new IllegalStateException("Data item size " + length + " exceeds maximum " + MAX_DATA_ITEM_BYTES);
+        }
         final ByteBuf item = buf.readSlice(length);
         return item.toString(StandardCharsets.UTF_8);
     }

graylog2-server/src/test/java/org/graylog/plugins/beats/BeatsFrameDecoderTest.java

@@ -21,6 +21,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.DecoderException;
 import io.netty.handler.logging.LoggingHandler;
 import org.graylog2.jackson.TypeReferences;
 import org.graylog2.shared.bindings.providers.ObjectMapperProvider;
@@ -32,6 +33,7 @@
 import java.util.zip.Deflater;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class BeatsFrameDecoderTest {
     private final ObjectMapper objectMapper = new ObjectMapperProvider().get();
@@ -238,4 +240,46 @@ private long extractSequenceNumber(ByteBuf buffer) {
         assertThat(buffer.readableBytes()).isEqualTo(0);
         return seqNum;
     }
-}
+
+    @Test
+    public void rejectsJsonFrameExceedingLengthLimit() {
+        final byte[] oversizedJson = ("{\"message\":\"" + "X".repeat(16 * 1024 * 1024) + "\"}")
+                .getBytes(StandardCharsets.UTF_8);
+
+        final ByteBuf frame = buildJsonFrame(oversizedJson, 0);
+
+        assertThatThrownBy(() -> {
+            channel.writeInbound(frame);
+            channel.checkException();
+        }).isInstanceOf(DecoderException.class);
+    }
+
+    @Test
+    public void rejectsDataFrameExceedingPairLimit() {
+        final ByteBuf frame = Unpooled.buffer();
+        frame.writeByte('2');
+        frame.writeByte('D');
+        frame.writeInt(0);
+        frame.writeInt(1025);
+
+        assertThatThrownBy(() -> {
+            channel.writeInbound(frame);
+            channel.checkException();
+        }).isInstanceOf(DecoderException.class);
+    }
+
+    @Test
+    public void rejectsDataItemExceedingLengthLimit() {
+        final ByteBuf frame = Unpooled.buffer();
+        frame.writeByte('2');
+        frame.writeByte('D');
+        frame.writeInt(0);
+        frame.writeInt(1);
+        frame.writeInt(1024 * 1024 + 1);
+
+        assertThatThrownBy(() -> {
+            channel.writeInbound(frame);
+            channel.checkException();
+        }).isInstanceOf(DecoderException.class);
+    }
+}