[
https://issues.apache.org/jira/browse/KAFKA-20966?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18106730#comment-18106730
]
Bill Bejeck commented on KAFKA-20966:
-------------------------------------
Hi [~sepurisaikrishna] is something you've obverved in a production setting?
> 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
> Priority: Major
>
> 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)
