[SPARK-57426][CORE] Validate length before allocation in Encoders.Strings.decode

[dongjoon-hyun]

What changes were proposed in this pull request?

Encoders.Strings.decode() now validates the length prefix before allocating the array, requiring 0 <= length <= buf.readableBytes() and throwing IndexOutOfBoundsException otherwise. After readInt(), the reader index is past the prefix, so readableBytes() equals the remaining payload bytes.

• Note that we use the Java built-in Objects.checkFromIndexSize because we cannot use Netty's checkReadableBytes. However, both methods throws IndexOutOfBoundsException identically for the error cases.
  • https://netty.io/4.2/api/io/netty/buffer/AbstractByteBuf.html#checkReadableBytes(int)
  • https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/Objects.html#checkFromIndexSize(int,int,int)

Why are the changes needed?

decode() allocated new byte[length] from an untrusted length.

• A negative value throws an opaque NegativeArraySizeException.
• An oversized value can trigger OutOfMemoryError on a corrupt or hostile frame.

Validating first fails fast with a clear error.

Note that OutOfMemoryError can happen even in a secured environment at the first parse on an unauthenticated channel.

spark/common/network-common/src/main/java/org/apache/spark/network/sasl/SaslRpcHandler.java

Lines 72 to 80 in 67eafbd

	public boolean doAuthChallenge(
	TransportClient client,
	ByteBuffer message,
	RpcResponseCallback callback) {
	if (saslServer == null || !saslServer.isComplete()) {
	ByteBuf nettyBuf = Unpooled.wrappedBuffer(message);
	SaslMessage saslMessage;
	try {
	saslMessage = SaslMessage.decode(nettyBuf);

spark/common/network-common/src/main/java/org/apache/spark/network/sasl/SaslMessage.java

Lines 68 to 74 in 67eafbd

	public static SaslMessage decode(ByteBuf buf) {
	if (buf.readByte() != TAG_BYTE) {
	throw new IllegalStateException("Expected SaslMessage, received something else"
	+ " (maybe your client does not have SASL enabled?)");
	}
	String appId = Encoders.Strings.decode(buf);

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Pass the CIs with newly added test cases.

Without this PR, testStringsDecodeShouldFailWhenLengthExceedsReadableBytes fails with OutOfMemoryError.

[info] Test org.apache.spark.network.protocol.EncodersSuite#testStringsDecodeShouldFailWhenLengthExceedsReadableBytes() started
[0.473s][warning][oom,vendor] java.lang.OutOfMemoryError occurred: Requested array size exceeds VM limit
[error] java.lang.OutOfMemoryError: Requested array size exceeds VM limit

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 4.8)

[dongjoon-hyun]

cc @huaxingao, FYI.

[dongjoon-hyun]

core module passed already.

[huaxingao]

LGTM

[dongjoon-hyun]

Thank you, @huaxingao .

[dongjoon-hyun]

Merged to master/4.x for now because we have Apache Spark 4.2.0 RC3 vote currently.

If it fails, I will consider to cherry-pick this to branch-4.2 and older.