This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.7 by this push:
new 787623469 [CELEBORN-2411][CIP22] Preserve Spark IO cipher
transformation
787623469 is described below
commit 787623469cc16699c600e7aab3909ee22494459f
Author: SparksFyz <[email protected]>
AuthorDate: Wed Aug 19 19:50:27 2026 +0800
[CELEBORN-2411][CIP22] Preserve Spark IO cipher transformation
### What changes were proposed in this pull request?
Preserve Spark's effective IO cipher transformation when
`SparkCryptoHandler`
builds its minimized `SparkConf`.
Add bidirectional interoperability tests between `SparkCryptoHandler` and
Spark's native `CryptoStreamUtils` using `AES/CBC/PKCS5Padding`.
### Why are the changes needed?
`CryptoStreamUtils.toCryptoConf()` only extracts
`spark.io.encryption.commons.config.*`, while Spark reads
`IO_CRYPTO_CIPHER_TRANSFORMATION` separately from `SparkConf`.
As a result, `SparkCryptoHandler` previously dropped a non-default
transformation and silently fell back to Spark's default
`AES/CTR/NoPadding`.
Existing self-round-trip tests did not detect this because both encryption
and decryption used the same minimized configuration.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
A non-default Spark IO cipher transformation is now preserved by Celeborn.
The default `AES/CTR/NoPadding` behavior remains unchanged.
### How was this patch tested?
- Added Celeborn encryption -> Spark `CryptoStreamUtils` decryption
coverage.
- Added Spark `CryptoStreamUtils` encryption -> Celeborn decryption
coverage.
- Spark 3.5 `SparkCryptoHandlerSuiteJ`: 10 tests passed.
- Spark 3.0 / Scala 2.12.10 / JDK 11 compilation passed.
- Spark 4.0 / Scala 2.13.16 / JDK 17 compilation passed.
- Spark 3.5 Spotless check passed.
Closes #3790 from SparksFyz/CELEBORN-2411-cipher-transformation.
Authored-by: SparksFyz <[email protected]>
Signed-off-by: zhengtao <[email protected]>
(cherry picked from commit 994106443354f6c5e8d4f03f36437d6d82a0f1d4)
Signed-off-by: zhengtao <[email protected]>
---
.../spark/shuffle/celeborn/SparkCryptoHandler.java | 4 ++
.../shuffle/celeborn/SparkCryptoHandlerSuiteJ.java | 69 ++++++++++++++++++++++
2 files changed, 73 insertions(+)
diff --git
a/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandler.java
b/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandler.java
index bf3922d48..20d4a0a90 100644
---
a/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandler.java
+++
b/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandler.java
@@ -26,6 +26,7 @@ import java.io.OutputStream;
import java.util.Properties;
import org.apache.spark.SparkConf;
+import org.apache.spark.internal.config.package$;
import org.apache.spark.security.CryptoStreamUtils;
import org.apache.celeborn.client.security.CryptoHandler;
@@ -52,6 +53,9 @@ public class SparkCryptoHandler implements CryptoHandler {
for (String propKey : cryptoProps.stringPropertyNames()) {
minimalConf.set(prefix + propKey, cryptoProps.getProperty(propKey));
}
+ minimalConf.set(
+ package$.MODULE$.IO_CRYPTO_CIPHER_TRANSFORMATION(),
+ sparkConf.get(package$.MODULE$.IO_CRYPTO_CIPHER_TRANSFORMATION()));
this.sparkConf = minimalConf;
this.key = key;
}
diff --git
a/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandlerSuiteJ.java
b/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandlerSuiteJ.java
index 6baa62626..5e4aaaefc 100644
---
a/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandlerSuiteJ.java
+++
b/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SparkCryptoHandlerSuiteJ.java
@@ -17,9 +17,15 @@
package org.apache.spark.shuffle.celeborn;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.*;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
+import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.security.SecureRandom;
@@ -27,6 +33,7 @@ import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.internal.config.package$;
+import org.apache.spark.security.CryptoStreamUtils;
import org.junit.Before;
import org.junit.Test;
@@ -138,6 +145,68 @@ public class SparkCryptoHandlerSuiteJ {
assertEquals(0, decrypted.length);
}
+ @Test
+ public void
testEncryptWithConfiguredCipherInteroperatesWithSparkCryptoStreamUtils()
+ throws IOException {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(package$.MODULE$.IO_ENCRYPTION_ENABLED(), true);
+ sparkConf.set(package$.MODULE$.IO_CRYPTO_CIPHER_TRANSFORMATION(),
"AES/CBC/PKCS5Padding");
+ assertEncryptInteroperatesWithSparkCryptoStreamUtils(sparkConf);
+ }
+
+ @Test
+ public void
testDecryptWithConfiguredCipherInteroperatesWithSparkCryptoStreamUtils()
+ throws IOException {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(package$.MODULE$.IO_ENCRYPTION_ENABLED(), true);
+ sparkConf.set(package$.MODULE$.IO_CRYPTO_CIPHER_TRANSFORMATION(),
"AES/CBC/PKCS5Padding");
+ assertDecryptInteroperatesWithSparkCryptoStreamUtils(sparkConf);
+ }
+
+ @Test
+ public void
testDefaultCipherTransformationInteroperatesWithSparkCryptoStreamUtils()
+ throws IOException {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(package$.MODULE$.IO_ENCRYPTION_ENABLED(), true);
+ assertEncryptInteroperatesWithSparkCryptoStreamUtils(sparkConf);
+ assertDecryptInteroperatesWithSparkCryptoStreamUtils(sparkConf);
+ }
+
+ private void assertEncryptInteroperatesWithSparkCryptoStreamUtils(SparkConf
sparkConf)
+ throws IOException {
+ byte[] plaintext = "12345678901234567".getBytes(UTF_8);
+ CryptoHandler cryptoHandler = new SparkCryptoHandler(sparkConf, key);
+
+ byte[] encrypted = cryptoHandler.encrypt(plaintext, 0, plaintext.length);
+ try (DataInputStream input =
+ new DataInputStream(
+ CryptoStreamUtils.createCryptoInputStream(
+ new ByteArrayInputStream(
+ encrypted, Integer.BYTES, encrypted.length -
Integer.BYTES),
+ sparkConf,
+ key))) {
+ byte[] decrypted = new byte[plaintext.length];
+ input.readFully(decrypted);
+ assertArrayEquals(plaintext, decrypted);
+ }
+ }
+
+ private void assertDecryptInteroperatesWithSparkCryptoStreamUtils(SparkConf
sparkConf)
+ throws IOException {
+ byte[] plaintext = "12345678901234567".getBytes(UTF_8);
+ CryptoHandler cryptoHandler = new SparkCryptoHandler(sparkConf, key);
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ DataOutputStream dataOutput = new DataOutputStream(output);
+ dataOutput.writeInt(plaintext.length);
+ try (OutputStream cryptoOutput =
+ CryptoStreamUtils.createCryptoOutputStream(dataOutput, sparkConf,
key)) {
+ cryptoOutput.write(plaintext);
+ }
+
+ byte[] encrypted = output.toByteArray();
+ assertArrayEquals(plaintext, cryptoHandler.decrypt(encrypted, 0,
encrypted.length));
+ }
+
/**
* Verifies that the decrypt bounds check uses {@code length - 20} (4-byte
length prefix + 16-byte
* IV), not the previous {@code length - 4}. A crafted payload whose
embedded length value is
