MINOR: after reading BYTES type it's possible to access data beyond i…

…ts size (apache#13261)

After reading data of type BYTES, COMPACT_BYTES, NULLABLE_BYTES or COMPACT_NULLABLE_BYTES returned ByteBuffer might have a capacity that is larger than its limit, thus these data types may access data that lies beyond its size by increasing limit of the returned ByteBuffer. I guess this is not very critical but I think it would be good to restrict increasing limit of the returned ByteBuffer by making its capacity strictly equal to its limit. I think someone might unintentionally mishandle these data types and accidentally mess up data in the ByteBuffer from which they were read.

Reviewers: Luke Chen <showuon@gmail.com>

clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java

@@ -689,9 +689,12 @@ public Object read(ByteBuffer buffer) {
             if (size > buffer.remaining())
                 throw new SchemaException("Error reading bytes of size " + size + ", only " + buffer.remaining() + " bytes available");
 
+            int limit = buffer.limit();
+            int newPosition = buffer.position() + size;
+            buffer.limit(newPosition);
             ByteBuffer val = buffer.slice();
-            val.limit(size);
-            buffer.position(buffer.position() + size);
+            buffer.limit(limit);
+            buffer.position(newPosition);
             return val;
         }
 
@@ -739,9 +742,12 @@ public Object read(ByteBuffer buffer) {
             if (size > buffer.remaining())
                 throw new SchemaException("Error reading bytes of size " + size + ", only " + buffer.remaining() + " bytes available");
 
+            int limit = buffer.limit();
+            int newPosition = buffer.position() + size;
+            buffer.limit(newPosition);
             ByteBuffer val = buffer.slice();
-            val.limit(size);
-            buffer.position(buffer.position() + size);
+            buffer.limit(limit);
+            buffer.position(newPosition);
             return val;
         }
 
@@ -800,9 +806,12 @@ public Object read(ByteBuffer buffer) {
             if (size > buffer.remaining())
                 throw new SchemaException("Error reading bytes of size " + size + ", only " + buffer.remaining() + " bytes available");
 
+            int limit = buffer.limit();
+            int newPosition = buffer.position() + size;
+            buffer.limit(newPosition);
             ByteBuffer val = buffer.slice();
-            val.limit(size);
-            buffer.position(buffer.position() + size);
+            buffer.limit(limit);
+            buffer.position(newPosition);
             return val;
         }
 
@@ -865,9 +874,12 @@ public Object read(ByteBuffer buffer) {
             if (size > buffer.remaining())
                 throw new SchemaException("Error reading bytes of size " + size + ", only " + buffer.remaining() + " bytes available");
 
+            int limit = buffer.limit();
+            int newPosition = buffer.position() + size;
+            buffer.limit(newPosition);
             ByteBuffer val = buffer.slice();
-            val.limit(size);
-            buffer.position(buffer.position() + size);
+            buffer.limit(limit);
+            buffer.position(newPosition);
             return val;
         }
 

clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java

@@ -456,4 +456,21 @@ public void testReadWithMissingNonOptionalExtraDataAtTheEnd() {
         SchemaException e = assertThrows(SchemaException.class, () -> newSchema.read(buffer));
         assertTrue(e.getMessage().contains("Missing value for field 'field2' which has no default value"));
     }
+
+    @Test
+    public void testReadBytesBeyondItsSize() {
+        Type[] types = new Type[]{
+            Type.BYTES,
+            Type.COMPACT_BYTES,
+            Type.NULLABLE_BYTES,
+            Type.COMPACT_NULLABLE_BYTES
+        };
+        for (Type type : types) {
+            ByteBuffer buffer = ByteBuffer.allocate(20);
+            type.write(buffer, ByteBuffer.allocate(4));
+            buffer.rewind();
+            ByteBuffer bytes = (ByteBuffer) type.read(buffer);
+            assertThrows(IllegalArgumentException.class, () -> bytes.limit(bytes.limit() + 1));
+        }
+    }
 }