Skip to content

[SPARK-59636][CORE] Fix integer overflow in GcmTransportCipher #58909

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aajisaka wants to merge 1 commit into
base: master
Choose a base branch
from
+68 −6
Open

[SPARK-59636][CORE] Fix integer overflow in GcmTransportCipher #58909

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 13 additions & 6 deletions common/network-common/src/main/java/org/apache/spark/network/crypto/GcmTransportCipher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@

public class GcmTransportCipher implements TransportCipher {
private static final String HKDF_ALG = "HmacSha256";
private static final int LENGTH_HEADER_BYTES = 8;
@VisibleForTesting
static final int LENGTH_HEADER_BYTES = 8;
@VisibleForTesting
static final int CIPHERTEXT_BUFFER_SIZE = 32 * 1024; // 32KB
// Maximum plaintext bytes to accumulate before flushing to downstream handlers, even
Expand Down Expand Up @@ -212,7 +213,7 @@ public boolean release(int decrement) {

@Override
public long transferTo(WritableByteChannel target, long position) throws IOException {
int transferredThisCall = 0;
long transferredThisCall = 0;
// If the header has is not empty, try to write it out to the target.
if (headerByteBuffer.hasRemaining()) {
int written = target.write(headerByteBuffer);
Expand Down Expand Up @@ -369,8 +370,9 @@ private boolean initializeExpectedLength(ByteBuf ciphertextNettyBuf) {
}
expectedLengthBuffer.flip();
expectedLength = expectedLengthBuffer.getLong();
if (expectedLength < 0) {
throw new IllegalStateException("Invalid expected ciphertext length.");
if (expectedLength < LENGTH_HEADER_BYTES + (long) headerLength) {

@dongjoon-hyun dongjoon-hyun Sep 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit. (long) looks unnecessary here because both operands are small int values.

throw new IllegalStateException(
"Invalid expected ciphertext length: " + expectedLength);
}
ciphertextRead += LENGTH_HEADER_BYTES;
}
Expand Down Expand Up @@ -442,8 +444,13 @@ public void channelRead(ChannelHandlerContext ctx, Object ciphertextMessage)
int readableBytes = Math.min(
nettyBufReadableBytes,
ciphertextBuffer.remaining());
int expectedRemaining = (int) (expectedLength - ciphertextRead);
int bytesToRead = Math.min(readableBytes, expectedRemaining);
long expectedRemaining = expectedLength - ciphertextRead;
if (expectedRemaining <= 0) {
throw new IllegalStateException(
"Invalid ciphertext state: expectedLength=" + expectedLength
+ ", ciphertextRead=" + ciphertextRead);
}
Comment on lines +448 to +452

@dongjoon-hyun dongjoon-hyun Sep 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this reachable? With the new lower-bound check in initializeExpectedLength, ciphertextRead <= expectedLength always holds after the header is read, and completed becomes true when they are equal, so this loop is not entered with expectedRemaining <= 0. Shall we remove this?

int bytesToRead = (int) Math.min((long) readableBytes, expectedRemaining);

@dongjoon-hyun dongjoon-hyun Sep 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit. (long) is unnecessary because Math.min(long, long) promotes readableBytes automatically.

// The smallest ciphertext size is 16 bytes for the auth tag
ciphertextBuffer.limit(ciphertextBuffer.position() + bytesToRead);
ciphertextNettyBuf.readBytes(ciphertextBuffer);
Expand Down
55 changes: 55 additions & 0 deletions common/network-common/src/test/java/org/apache/spark/network/crypto/GcmAuthEngineSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@

package org.apache.spark.network.crypto;

import com.google.common.primitives.Longs;
import com.google.crypto.tink.subtle.AesGcmHkdfStreaming;
import com.google.crypto.tink.subtle.StreamSegmentEncrypter;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
Expand Down Expand Up @@ -570,6 +573,58 @@ public void testSplitLengthPrefix() throws Exception {
}
}

@Test
public void testCiphertextLengthLargerThanMaxInt() throws Exception {
TransportConf gcmConf = getConf(2, false);
try (AuthEngine client = new AuthEngine("appId", "secret", gcmConf);
AuthEngine server = new AuthEngine("appId", "secret", gcmConf)) {
AuthMessage clientChallenge = client.challenge();
AuthMessage serverResponse = server.response(clientChallenge);
client.deriveSessionCipher(clientChallenge, serverResponse);
GcmTransportCipher cipher = (GcmTransportCipher) server.sessionCipher();
GcmTransportCipher.DecryptionHandler decryptionHandler = cipher.getDecryptionHandler();
AesGcmHkdfStreaming streaming = cipher.getAesGcmHkdfStreaming();

long expectedLength = (long) GcmTransportCipher.LENGTH_HEADER_BYTES +
streaming.getHeaderLength() + Integer.MAX_VALUE + 1L;
Comment on lines +588 to +589

@dongjoon-hyun dongjoon-hyun Sep 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit. Could you add a comment explaining that the remaining ciphertext length is Integer.MAX_VALUE + 1, which becomes Integer.MIN_VALUE when cast to int?

StreamSegmentEncrypter encrypter = streaming.newStreamSegmentEncrypter(
Longs.toByteArray(expectedLength));
ByteBuffer header = encrypter.getHeader();
ByteBuf ciphertext = Unpooled.buffer(GcmTransportCipher.LENGTH_HEADER_BYTES +
header.remaining() + 1)
.writeLong(expectedLength)
.writeBytes(header)
.writeByte(0);
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);

decryptionHandler.channelRead(ctx, ciphertext);

verify(ctx, never()).fireChannelRead(any());
}
}

@Test
public void testInvalidExpectedCiphertextLength() throws Exception {
TransportConf gcmConf = getConf(2, false);
try (AuthEngine client = new AuthEngine("appId", "secret", gcmConf);
AuthEngine server = new AuthEngine("appId", "secret", gcmConf)) {
AuthMessage clientChallenge = client.challenge();
AuthMessage serverResponse = server.response(clientChallenge);
client.deriveSessionCipher(clientChallenge, serverResponse);
GcmTransportCipher cipher = (GcmTransportCipher) server.sessionCipher();
GcmTransportCipher.DecryptionHandler decryptionHandler = cipher.getDecryptionHandler();
long invalidLength = (long) GcmTransportCipher.LENGTH_HEADER_BYTES +
cipher.getAesGcmHkdfStreaming().getHeaderLength() - 1;
ByteBuf ciphertext = Unpooled.buffer(8).writeLong(invalidLength);

IllegalStateException error = assertThrows(
IllegalStateException.class,
() -> decryptionHandler.channelRead(mock(ChannelHandlerContext.class), ciphertext));

assertEquals("Invalid expected ciphertext length: " + invalidLength, error.getMessage());
}
}

/**
* Regression test for the encryptedCount miscalculation that caused shuffle fetch stalls
* for plaintext sizes in (plaintextSegmentSize - getCiphertextOffset(), plaintextSegmentSize]
Expand Down