You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/06/11 17:23:26 UTC
(kafka) branch trunk updated: MINOR: improve byte[] array size bounds check (#22514)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new fce22525f74 MINOR: improve byte[] array size bounds check (#22514)
fce22525f74 is described below
commit fce22525f74bbd7feeb4f8b34c79a215db9a74c3
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Thu Jun 11 10:23:18 2026 -0700
MINOR: improve byte[] array size bounds check (#22514)
Reviewers: Andrew Schofield <as...@confluent.io>, Chia-Ping Tsai
<ch...@gmail.com>, Lucas Brutschy <lb...@confluent.io>
---
.../common/serialization/ListDeserializer.java | 22 +++++++++++++---------
.../common/serialization/ListDeserializerTest.java | 6 +++---
.../internals/assignment/AssignmentInfo.java | 20 ++++++++++++--------
3 files changed, 28 insertions(+), 20 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java b/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java
index 58ec06deda0..388a4bd7c93 100644
--- a/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java
+++ b/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java
@@ -153,8 +153,8 @@ public class ListDeserializer<Inner> implements Deserializer<List<Inner>> {
if (nullIndexListSize < 0) {
throw new SerializationException("Corrupted byte[]. The number of null list entries cannot be negative.");
}
- if (nullIndexListSize > length) {
- throw new SerializationException("Corrupted byte[]. The number of null list entries cannot be larger than overall number of bytes.");
+ if (nullIndexListSize > length / Integer.BYTES) {
+ throw new SerializationException("Corrupted byte[]. The number of null list entries cannot be larger than overall number of elements.");
}
List<Integer> nullIndexList = new ArrayList<>(nullIndexListSize);
while (nullIndexListSize != 0) {
@@ -220,14 +220,18 @@ public class ListDeserializer<Inner> implements Deserializer<List<Inner>> {
final SerializationStrategy serStrategy,
final int length
) throws IOException {
- final int entrySize = serStrategy == SerializationStrategy.CONSTANT_SIZE ? primitiveSize : dis.readInt();
- if (entrySize < -1) { // value `-1` is valid, encoding a null entry (-> ListSerde.NULL_ENTRY_VALUE)
- throw new SerializationException("Corrupted byte[]. A list entry cannot have negative size.");
- }
- if (entrySize > length) {
- throw new SerializationException("Corrupted byte[]. A list entry cannot be larger than the overall number of bytes.");
+ if (serStrategy == SerializationStrategy.CONSTANT_SIZE) {
+ return primitiveSize;
+ } else {
+ final int entrySize = dis.readInt();
+ if (entrySize < -1) { // value `-1` is valid, encoding a null entry (-> ListSerde.NULL_ENTRY_VALUE)
+ throw new SerializationException("Corrupted byte[]. A list entry cannot have negative size.");
+ }
+ if (entrySize > length) {
+ throw new SerializationException("Corrupted byte[]. A list entry cannot be larger than the overall number of bytes.");
+ }
+ return entrySize;
}
- return entrySize;
}
@Override
diff --git a/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java b/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java
index 973614c88f4..95da416af4b 100644
--- a/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java
@@ -361,7 +361,7 @@ public class ListDeserializerTest {
(byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF // encodes number of null entries == -1
};
- final ListDeserializer<String> testDeserializer = new ListDeserializer<>(ArrayList.class, new StringDeserializer());
+ final ListDeserializer<Integer> testDeserializer = new ListDeserializer<>(ArrayList.class, new IntegerDeserializer());
final SerializationException exception = assertThrows(
SerializationException.class,
@@ -380,14 +380,14 @@ public class ListDeserializerTest {
(byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0xFF // encodes number of null entries == 255
};
- final ListDeserializer<String> testDeserializer = new ListDeserializer<>(ArrayList.class, new StringDeserializer());
+ final ListDeserializer<Integer> testDeserializer = new ListDeserializer<>(ArrayList.class, new IntegerDeserializer());
final SerializationException exception = assertThrows(
SerializationException.class,
() -> testDeserializer.deserialize(null, corruptedData)
);
assertEquals(
- "Corrupted byte[]. The number of null list entries cannot be larger than overall number of bytes.",
+ "Corrupted byte[]. The number of null list entries cannot be larger than overall number of elements.",
exception.getMessage()
);
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java
index 6dc60fb6773..d2cc5ab0fbd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java
@@ -389,7 +389,7 @@ public class AssignmentInfo {
final DataInputStream in,
final int length) throws IOException {
final int count = in.readInt();
- if (count < 0 || count > length) {
+ if (count < 0 || count > length / (2 * Integer.BYTES)) { // task-id is <subtopologyId[INTEGER]><partition[INTEGER]>
throw new TaskAssignmentException("Corrupted user data byte[].");
}
assignmentInfo.activeTasks = new ArrayList<>(count);
@@ -402,7 +402,7 @@ public class AssignmentInfo {
final DataInputStream in,
final int length) throws IOException {
final int count = in.readInt();
- if (count < 0 || count > length) {
+ if (count < 0 || count > length / (3 * Integer.BYTES)) { // task-id is <subtopologyId[INTEGER]><partition[INTEGER]> plus numPartitions[INTEGER]
throw new TaskAssignmentException("Corrupted user data byte[].");
}
assignmentInfo.standbyTasks = new HashMap<>(count);
@@ -415,11 +415,13 @@ public class AssignmentInfo {
private static void decodePartitionsByHost(final AssignmentInfo assignmentInfo,
final DataInputStream in,
final int length) throws IOException {
- assignmentInfo.partitionsByHost = new HashMap<>();
+ final int minEntryBytes = Short.BYTES + Integer.BYTES + Integer.BYTES; // host UTF length + port + numPartitions
+
final int numEntries = in.readInt();
- if (numEntries < 0 || numEntries > length) {
+ if (numEntries < 0 || numEntries > length / minEntryBytes) {
throw new TaskAssignmentException("Corrupted user data byte[].");
}
+ assignmentInfo.partitionsByHost = new HashMap<>(numEntries);
for (int i = 0; i < numEntries; i++) {
final HostInfo hostInfo = new HostInfo(in.readUTF(), in.readInt());
assignmentInfo.partitionsByHost.put(hostInfo, readTopicPartitions(in, length));
@@ -429,7 +431,7 @@ public class AssignmentInfo {
private static Set<TopicPartition> readTopicPartitions(final DataInputStream in,
final int length) throws IOException {
final int numPartitions = in.readInt();
- if (numPartitions < 0 || numPartitions > length) {
+ if (numPartitions < 0 || numPartitions > length / (Short.BYTES + Integer.BYTES)) { // topic name UTF length + <partitionNumber[INTEGER]>
throw new TaskAssignmentException("Corrupted user data byte[].");
}
final Set<TopicPartition> partitions = new HashSet<>(numPartitions);
@@ -442,7 +444,7 @@ public class AssignmentInfo {
private static Map<Integer, String> decodeTopicIndexAndGet(final DataInputStream in,
final int length) throws IOException {
final int dictSize = in.readInt();
- if (dictSize < 0 || dictSize > length) {
+ if (dictSize < 0 || dictSize > length / (Integer.BYTES + Short.BYTES)) { // <topicDicIndex[INTEGER]> and topic name UTF length
throw new TaskAssignmentException("Corrupted user data byte[].");
}
final Map<Integer, String> topicIndexDict = new HashMap<>(dictSize);
@@ -455,8 +457,10 @@ public class AssignmentInfo {
private static Map<HostInfo, Set<TopicPartition>> decodeHostPartitionMapUsingDictionary(final DataInputStream in,
final Map<Integer, String> topicIndexDict,
final int length) throws IOException {
+ final int minEntryBytes = Short.BYTES + Integer.BYTES + Integer.BYTES; // host UTF length + port + numPartitions
+
final int numEntries = in.readInt();
- if (numEntries < 0 || numEntries > length) {
+ if (numEntries < 0 || numEntries > length / minEntryBytes) {
throw new TaskAssignmentException("Corrupted user data byte[].");
}
final Map<HostInfo, Set<TopicPartition>> hostPartitionMap = new HashMap<>(numEntries);
@@ -486,7 +490,7 @@ public class AssignmentInfo {
final Map<Integer, String> topicIndexDict,
final int length) throws IOException {
final int numPartitions = in.readInt();
- if (numPartitions < 0 || numPartitions > length) {
+ if (numPartitions < 0 || numPartitions > length / (2 * Integer.BYTES)) { // <topicDicIndex[INTEGER]> and <partitionNumber[INTEGER]>
throw new TaskAssignmentException("Corrupted user data byte[].");
}
final Set<TopicPartition> partitions = new HashSet<>(numPartitions);