dongjoon-hyun commented on code in PR #58909:
URL: https://github.com/apache/spark/pull/58909#discussion_r4424072702
##########
common/network-common/src/main/java/org/apache/spark/network/crypto/GcmTransportCipher.java:
##########
@@ -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 + headerLength) {
Review Comment:
**2. [Low] The lower bound still accepts 1 to 15 bytes of ciphertext**
A sender writes either no segment or at least one tagged segment, so
`LENGTH_HEADER_BYTES + headerLength + k` with `k` in 1..15 cannot come from a
valid sender. It cannot be decrypted either. With Tink 1.23.0, these sizes fail
with `AEADBadTagException` ("Input too short - need tag"), while a 16-byte
tag-only segment decrypts.
Such a length passes this check, waits for `k` bytes, and fails only in
`decryptSegment`. If those bytes never arrive, the handler waits until the
channel closes.
Since this check now validates the frame, could it also reject that range?
The decodable lengths are exactly `LENGTH_HEADER_BYTES + headerLength` and
anything at least 16 bytes above it, which matches the "smallest ciphertext
size is 16 bytes" comment in `channelRead`.
##########
common/network-common/src/main/java/org/apache/spark/network/crypto/GcmTransportCipher.java:
##########
@@ -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 + headerLength) {
+ throw new IllegalStateException(
+ "Invalid expected ciphertext length: " +
expectedLength);
}
Review Comment:
**1. [Low] A rejected length stays in `expectedLength` and skips validation
on re-entry**
The value is stored before it is checked, and `expectedLength < 0` also
serves as the "not read yet" sentinel. A value in `[0, LENGTH_HEADER_BYTES +
headerLength)` is rejected here but stays in the field, so the next
`channelRead()` skips this block with `ciphertextRead` still at 0. With the
default 24-byte header:
- With length 5, the next read takes `ciphertextRead` to 24,
`expectedRemaining` becomes -19, and `ciphertextBuffer.limit(-19)` throws the
same `IllegalArgumentException` this PR removes.
- With length 24, `initializeDecrypter` sets `completed = true`, so the
rejected frame is treated as an empty message and the stream keeps going.
Before this PR a rejected value was always negative, so the sentinel stayed
set and every later read threw the same `IllegalStateException` again.
`ctx.close()` in `TransportChannelHandler.exceptionCaught` stops further
reads today, so this is not a live bug. Checking a local first keeps the
handler failing on its own:
```suggestion
long length = expectedLengthBuffer.getLong();
if (length < LENGTH_HEADER_BYTES + headerLength) {
throw new IllegalStateException(
"Invalid expected ciphertext length: " + length);
}
expectedLength = length;
```
##########
common/network-common/src/test/java/org/apache/spark/network/crypto/GcmAuthEngineSuite.java:
##########
@@ -570,6 +573,60 @@ 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();
+
+ // remaining ciphertext length is Integer.MAX_VALUE + 1,
+ // which becomes Integer.MIN_VALUE when cast to int.
+ long expectedLength = (long) GcmTransportCipher.LENGTH_HEADER_BYTES +
Review Comment:
**3. [Medium] The test covers 2^31 but not 2^32, where the old code hangs
instead of throwing**
This is the one remaining length that makes the old cast negative. The old
code has a second failure mode that no test reaches:
- With exactly 2^32 remaining, `(int)` gives 0, so `bytesToRead` is 0,
nothing is consumed, and the inner `while` never exits on the EventLoop. There
is no exception and no log.
- With a remaining length between 2^32 and 2^32 + 2^31, the cast is
positive, so the handler reads until exactly 2^32 remains and then spins the
same way.
So messages of roughly 4 to 6 GiB hang the EventLoop rather than fail with
the `IllegalArgumentException` in the PR description, and the pattern repeats
every 4 GiB. The 3 GB manual run does not reach this range either.
Could we also test 2^32 and 2^32 + 1, and wrap the call in
`assertTimeoutPreemptively`? Without a timeout, a regression at 2^32 would hang
the suite instead of failing it. Mentioning the hang in the PR description
would also help anyone matching this fix to a field symptom.
##########
common/network-common/src/main/java/org/apache/spark/network/crypto/GcmTransportCipher.java:
##########
@@ -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;
Review Comment:
**5. [Low] No test covers the `int` to `long` change here**
Both new tests call only `DecryptionHandler`, so reverting this line to
`int` still passes. With `int`, a `transferTo` call that writes 2 GiB or more
returns a negative value, and Netty's `AbstractNioByteChannel.doWriteInternal`
handles any non-positive result as a full send buffer.
A small `AbstractFileRegion` with `count() > Integer.MAX_VALUE` and a
discarding `WritableByteChannel` should be enough to assert that the return
value exceeds `Integer.MAX_VALUE` and equals `transferred()`. This needs no 2
GiB buffer, only a few seconds of encryption.
##########
common/network-common/src/test/java/org/apache/spark/network/crypto/GcmAuthEngineSuite.java:
##########
@@ -570,6 +573,60 @@ 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();
+
+ // remaining ciphertext length is Integer.MAX_VALUE + 1,
+ // which becomes Integer.MIN_VALUE when cast to int.
+ long expectedLength = (long) GcmTransportCipher.LENGTH_HEADER_BYTES +
+ streaming.getHeaderLength() + Integer.MAX_VALUE + 1L;
+ 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());
Review Comment:
**4. [Medium] `never()` cannot fail here, so the test only checks that
nothing is thrown**
With one ciphertext byte, `Math.min(1, expectedRemaining)` is 1 for any
positive value and the segment buffer never fills, so `fireChannelRead` is
unreachable under any implementation. Segment boundaries, the key derived from
the length AAD and the mid-message flush are not exercised, and the AAD passed
to `newStreamSegmentEncrypter` is never used.
Since `encrypter` is already here, could this encrypt real segments with
`encrypter.encryptSegment(plaintext, false, out)`, feed at least
`MAX_PLAINTEXT_BATCH_BYTES` of them, and assert the flushed plaintext? That
would check that a message declared larger than 2 GiB actually decrypts.
##########
common/network-common/src/test/java/org/apache/spark/network/crypto/GcmAuthEngineSuite.java:
##########
@@ -570,6 +573,60 @@ 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();
+
+ // remaining ciphertext length is Integer.MAX_VALUE + 1,
+ // which becomes Integer.MIN_VALUE when cast to int.
+ long expectedLength = (long) GcmTransportCipher.LENGTH_HEADER_BYTES +
+ streaming.getHeaderLength() + Integer.MAX_VALUE + 1L;
+ 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);
Review Comment:
**9. [Nit] Unnecessary `(long)` and a literal 8**
`(long)` is unnecessary here because both operands are small `int` values,
like [the earlier
comment](https://github.com/apache/spark/pull/58909#discussion_r4050849650) on
the main code. The `(long)` at L590 is needed because that sum exceeds `int`.
`Unpooled.buffer(8)` could use `GcmTransportCipher.LENGTH_HEADER_BYTES`,
like the test above.
##########
common/network-common/src/main/java/org/apache/spark/network/crypto/GcmTransportCipher.java:
##########
@@ -442,8 +444,8 @@ 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;
+ int bytesToRead = (int) Math.min(readableBytes,
expectedRemaining);
Review Comment:
**8. [Nit] Existing code: the `ciphertextRead > expectedLength` branch at
L456 is unreachable**
`bytesToRead` is at most `expectedRemaining`, so `ciphertextRead` cannot
pass `expectedLength`, and a negative `bytesToRead` fails earlier in
`ciphertextBuffer.limit()`. This is the same reasoning as [the earlier
thread](https://github.com/apache/spark/pull/58909#discussion_r4050849659) that
removed the `expectedRemaining <= 0` guard. Could the check be reduced to `if
(ciphertextRead == expectedLength)`?
##########
common/network-common/src/test/java/org/apache/spark/network/crypto/GcmAuthEngineSuite.java:
##########
@@ -570,6 +573,60 @@ 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();
+
+ // remaining ciphertext length is Integer.MAX_VALUE + 1,
+ // which becomes Integer.MIN_VALUE when cast to int.
+ long expectedLength = (long) GcmTransportCipher.LENGTH_HEADER_BYTES +
+ streaming.getHeaderLength() + Integer.MAX_VALUE + 1L;
+ 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 {
Review Comment:
**6. [Low] Only one side of the new bound is tested**
This covers `LENGTH_HEADER_BYTES + headerLength - 1`. The accepted minimum,
which is an empty plaintext message, has no test, and this suite has no
empty-message round trip at all. Changing `<` to `<=` would make an empty
message close the connection while every test still passes. Negative lengths,
the only case rejected before this PR, have no test either.
Could this also cover an empty-buffer round trip, plus rejection of 0, -1
and `Long.MIN_VALUE`?
##########
common/network-common/src/main/java/org/apache/spark/network/crypto/GcmTransportCipher.java:
##########
@@ -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;
Review Comment:
**7. [Low] Existing code: the `while` loop at L235 can run for a whole
message without yielding**
The `long` here is the right minimal fix, and this is not something this PR
introduced. But for this counter to pass 2 GiB, one call has to encrypt and
write 2 GiB on the EventLoop. The loop at L235 returns early only when the
socket stops taking a full segment or the file region delivers less than
requested. With a fast reader, the other channels on that EventLoop wait for
the whole message, and the comments in `DecryptionHandler` already treat
EventLoop starvation as a real problem.
Capping the bytes written per call and returning would let Netty continue
through `writeSpinCount` and its flush task, and would make this overflow
impossible. That is larger than this fix, so I think it belongs in a separate
JIRA.
##########
common/network-common/src/main/java/org/apache/spark/network/crypto/GcmTransportCipher.java:
##########
@@ -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 + headerLength) {
+ throw new IllegalStateException(
Review Comment:
**10. [Low] The PR description does not mention this check or the
`transferTo` change**
The description only explains the `IllegalArgumentException` in the
decryption path. This PR also newly rejects lengths in `[0, LENGTH_HEADER_BYTES
+ headerLength)` and changes the sender's counter, and reviewers and backport
decisions depend on knowing that.
The compatibility answer is worth stating too. Older senders do not trip the
new bound, because both the current `encryptedCount` and the formula before
SPARK-56227 (`LENGTH_HEADER_BYTES + expectedCiphertextSize(P)`) are always at
least this minimum.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
