sepuri sai krishna created KAFKA-20966:
------------------------------------------
Summary: RemoteLogInputStream can attempt an unbounded memory
allocation when reading a corrupted remote log segment
Key: KAFKA-20966
URL: https://issues.apache.org/jira/browse/KAFKA-20966
Project: Kafka
Issue Type: Bug
Components: Tiered-Storage
Reporter: sepuri sai krishna
Assignee: sepuri sai krishna
RemoteLogInputStream.nextBatch()
(clients/src/main/java/org/apache/kafka/common/record/internal/RemoteLogInputStream.java,
lines 49-59) reads a 4-byte batch-size field directly off the InputStream
returned by the pluggable RemoteStorageManager and uses it to size an
allocation, with no check other than a lower bound:
int size = logHeaderBuffer.getInt(SIZE_OFFSET);
// V0 has the smallest overhead, stricter checking is done later
if (size < LegacyRecord.RECORD_OVERHEAD_V0)
throw new CorruptRecordException(...);
int bufferSize = LOG_OVERHEAD + size;
ByteBuffer buffer = ByteBuffer.allocate(bufferSize); // no upper bound on
size
There is no check that "size" doesn't exceed a sane maximum before allocating.
"size" is a 4-byte signed int taken directly from the remote segment's bytes,
so it can be as large as ~2GB.
Its sibling class, ByteBufferLogInputStream (same package), reads the identical
length-prefixed header format but validates the declared size against
maxMessageSize before trusting it:
if (recordSize > maxMessageSize)
throw new CorruptRecordException(String.format(
"Record size %d exceeds the largest allowable message size (%d).",
recordSize, maxMessageSize));
RemoteLogInputStream has no equivalent check, and is actually the more exposed
of the two: ByteBufferLogInputStream only slices an already-in-memory,
already-bounded ByteBuffer, whereas RemoteLogInputStream calls
ByteBuffer.allocate() directly from the untrusted value, before it has even
validated that the input stream contains that many bytes.
Impact: a corrupted or bit-rotted remote log segment, or a misbehaving/buggy
pluggable RemoteStorageManager implementation (S3/GCS/HDFS-backed, etc.), can
cause the broker to attempt allocating up to ~2GB per batch read. This is
reachable from RemoteLogManager.read() (consumer fetch falling through to
tiered storage) and RemoteLogManager.lookupTimestamp() (offset-by-timestamp
lookups against tiered segments) -- both real, hot server-side code paths, not
test-only code.
Proposed fix: thread a maxMessageSize bound into RemoteLogInputStream's
constructor (mirroring ByteBufferLogInputStream's existing pattern) and throw
CorruptRecordException if the declared size exceeds it, using
UnifiedLog.config().maxMessageSize(), which is already available at both call
sites in RemoteLogManager.
I'm happy to submit a PR for this fix.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
