[
https://issues.apache.org/jira/browse/KAFKA-20966?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18106745#comment-18106745
]
sepuri sai krishna commented on KAFKA-20966:
--------------------------------------------
Hi [~bbejeck], thanks for looking at this.
No, not from production. I found it by code inspection while comparing the
tiered read path against the other {{LogInputStream}} implementations, so
please weigh the priority accordingly. That said, I've since reproduced it end
to end through {{RemoteLogManager.read()}}, so it isn't only theoretical.
Details below.
h3. What the code does
{{RemoteLogInputStream.nextBatch()}} reads the size field off the segment and
sizes an allocation from it, checking only the lower bound:
{code:java}
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);
{code}
Its closest structural twin is
{{AbstractLegacyRecordBatch.DataLogInputStream}}, same interface, also reading
from a raw {{InputStream}}, also reading {{SIZE_OFFSET}}, also allocating from
it, and that one bounds the value first:
{code:java}
if (size > maxMessageSize)
throw new CorruptRecordException(String.format("Record size exceeds the
largest allowable message size (%d).", maxMessageSize));
ByteBuffer batchBuffer = ByteBuffer.allocate(size);
{code}
{{ByteBufferLogInputStream.nextBatchSize()}} bounds it the same way.
{{FileLogInputStream}} has no {{maxMessageSize}}, but doesn't need one, it
bounds against the actual file length ({{if (position > end - LOG_OVERHEAD -
size)}}) and is file-channel backed, so it never allocates from the field at
all. {{RemoteLogInputStream}} is the only one of the four that allocates from
that field unbounded.
h3. Reproduction
I wrote a test that drives the real production path: a real segment file with
two real records, with only the 4-byte size field at {{SIZE_OFFSET}}
overwritten, returned by a stubbed {{fetchLogSegment()}}. It deliberately does
*not* stub {{getRemoteLogInputStream()}}, so the real {{RemoteLogInputStream}}
reads the real bytes. {{max.message.bytes}} is the default 1048588 throughout.
Three cases:
*Control*, untouched segment, reads back fine:
{noformat}
CONTROL: read baseOffset=0 lastOffset=1 sizeInBytes=83
{noformat}
*Size field set to Integer.MAX_VALUE*, {{LOG_OVERHEAD + size}} overflows int,
so {{allocate()}} gets a negative capacity. This is deterministic and
heap-independent:
{noformat}
OVERFLOW: RemoteLogManager.read() threw java.lang.IllegalArgumentException:
capacity < 0: (-2147483637 < 0)
at
org.apache.kafka.common.record.internal.RemoteLogInputStream.nextBatch(RemoteLogInputStream.java:59)
at
org.apache.kafka.server.log.remote.storage.RemoteLogManager.findFirstBatch(RemoteLogManager.java:2109)
at
org.apache.kafka.server.log.remote.storage.RemoteLogManager.read(RemoteLogManager.java:1899)
{noformat}
*Size field set to 512 MiB*, the allocation succeeds inside the test heap, and
then the short read makes {{nextBatch()}} return null, which {{findFirstBatch}}
treats as end-of-stream:
{noformat}
SILENT: read() returned normally, records empty=true, after allocating
536870924 bytes; no CorruptRecordException was raised
{noformat}
I found that third case the more interesting one, and I hadn't appreciated it
when I filed this. A corrupted size field is indistinguishable from a clean
end-of-stream: the broker allocates half a gigabyte on the fetch path, throws
it away, and hands the consumer an empty fetch with no error and no corruption
signal at all. Separately, in a standalone harness with the allocation sized at
~2 GiB and {{-Xmx256m}}, the same call site produces {{OutOfMemoryError: Java
heap space}}, though obviously whether that happens depends on the heap.
In all three corrupted cases a {{size > maxMessageSize}} check would have
turned it into a {{CorruptRecordException}} naming the bad value.
h3. Scope
To be clear about the trigger, since I think it's what decides whether this is
worth fixing: the bytes come back from the configured {{RemoteStorageManager}},
so reaching this needs a corrupted or truncated segment, or a plugin returning
a stream at the wrong position. It isn't attacker-controlled input in any
normal sense, and I'm not claiming a security issue. The argument is just that
the tiered path trusts a pluggable third-party store more than the local path
trusts its own disk, which seems backwards, and the fix is small, thread the
existing {{maxMessageSize}} through from the log config, which both call sites
in {{RemoteLogManager}} already have to hand.
PR is up at https://github.com/apache/kafka/pull/23206. Happy to add the
{{RemoteLogManager}}-level test above to it if that's useful, and equally happy
to close this as Not A Problem if you think the extra parameter isn't worth
carrying.
> 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)
