Fix: safe-guard byte buf allocation (#420)

since `inflateCompressedFrame` might have leaked, see #408

Author: kares

CHANGELOG.md

@@ -1,4 +1,5 @@
-## Unreleased
+## 6.1.3
+ - Fix: safe-guard byte buf allocation [#420](https://github.com/logstash-plugins/logstash-input-beats/pull/420) 
  - Updated Jackson dependencies
 
 ## 6.1.2
@@ -13,7 +14,7 @@
    `host` and `@metadata.ip_address` event fields. [#404](https://github.com/logstash-plugins/logstash-input-beats/pull/404)
 
 ## 6.0.14
-- Feat: log + unwrap generic SSL context exceptions [#405](https://github.com/logstash-plugins/logstash-input-beats/pull/405)
+ - Feat: log + unwrap generic SSL context exceptions [#405](https://github.com/logstash-plugins/logstash-input-beats/pull/405)
 
 ## 6.0.13
  - [DOC] Update links to use shared attributes

VERSION

@@ -1 +1 @@
-6.1.2
+6.1.3

src/main/java/org/logstash/beats/BeatsParser.java

@@ -47,7 +47,7 @@ private enum States {
     private boolean decodingCompressedBuffer = false;
 
     @Override
-    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws InvalidFrameProtocolException, IOException {
         if(!hasEnoughBytes(in)) {
             if (decodingCompressedBuffer){
                 throw new InvalidFrameProtocolException("Insufficient bytes in compressed content to decode: " + currentState);
@@ -124,7 +124,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
                 int fieldsCount = (int) in.readUnsignedInt();
                 int count = 0;
 
-                if(fieldsCount <= 0) {
+                if (fieldsCount <= 0) {
                     throw new InvalidFrameProtocolException("Invalid number of fields, received: " + fieldsCount);
                 }
 
@@ -178,20 +178,19 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
 
             case READ_COMPRESSED_FRAME: {
                 logger.trace("Running: READ_COMPRESSED_FRAME");
-                // Use the compressed size as the safe start for the buffer.
-                ByteBuf buffer = inflateCompressedFrame(ctx, in);
-                transition(States.READ_HEADER);
+                inflateCompressedFrame(ctx, in, (buffer) -> {
+                    transition(States.READ_HEADER);
 
-                decodingCompressedBuffer = true;
-                try {
-                    while (buffer.readableBytes() > 0) {
-                        decode(ctx, buffer, out);
+                    decodingCompressedBuffer = true;
+                    try {
+                        while (buffer.readableBytes() > 0) {
+                            decode(ctx, buffer, out);
+                        }
+                    } finally {
+                        decodingCompressedBuffer = false;
+                        transition(States.READ_HEADER);
                     }
-                } finally {
-                    decodingCompressedBuffer = false;
-                    buffer.release();
-                    transition(States.READ_HEADER);
-                }
+                });
                 break;
             }
             case READ_JSON: {
@@ -211,18 +210,28 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
         }
     }
 
-    private ByteBuf inflateCompressedFrame(final ChannelHandlerContext ctx, final ByteBuf in) throws IOException {
+    private void inflateCompressedFrame(final ChannelHandlerContext ctx, final ByteBuf in, final CheckedConsumer<ByteBuf> fn)
+        throws IOException {
+        // Use the compressed size as the safe start for the buffer.
         ByteBuf buffer = ctx.alloc().buffer(requiredBytes);
+        try {
+            decompressImpl(in, buffer);
+            fn.accept(buffer);
+        } finally {
+            buffer.release();
+        }
+    }
+
+    private void decompressImpl(final ByteBuf in, final ByteBuf out) throws IOException {
         Inflater inflater = new Inflater();
         try (
-                ByteBufOutputStream buffOutput = new ByteBufOutputStream(buffer);
+                ByteBufOutputStream buffOutput = new ByteBufOutputStream(out);
                 InflaterOutputStream inflaterStream = new InflaterOutputStream(buffOutput, inflater)
         ) {
             in.readBytes(inflaterStream, requiredBytes);
-        }finally{
+        } finally {
             inflater.end();
         }
-        return buffer;
     }
 
     private boolean hasEnoughBytes(ByteBuf in) {
@@ -234,7 +243,7 @@ private void transition(States next) {
     }
 
     private void transition(States nextState, int requiredBytes) {
-        if(logger.isTraceEnabled()) {
+        if (logger.isTraceEnabled()) {
             logger.trace("Transition, from: " + currentState + ", to: " + nextState + ", requiring " + requiredBytes + " bytes");
         }
         this.currentState = nextState;
@@ -247,4 +256,9 @@ private void batchComplete() {
         batch = null;
     }
 
+    @FunctionalInterface
+    private interface CheckedConsumer<T> {
+        void accept(T t) throws IOException;
+    }
+
 }

src/main/java/org/logstash/beats/InvalidFrameProtocolException.java

@@ -1,6 +1,6 @@
 package org.logstash.beats;
 
-public class InvalidFrameProtocolException extends Exception {
+public class InvalidFrameProtocolException extends RuntimeException {
     InvalidFrameProtocolException(String message) {
         super(message);
     }