From ef48cd3b34f8e853e9ac0dbd13c9203554ca768f Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Dec 2025 14:18:02 -0500 Subject: [PATCH 1/8] Fix unsafe deserialization --- .../implementation/KafkaCosmosUtils.java | 17 ++++- .../connect/CosmosSinkConnectorTest.java | 62 ++++++++++++++----- 2 files changed, 62 insertions(+), 17 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java index 6009821ea188..a1d9bc85fe8d 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java @@ -12,20 +12,31 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InvalidClassException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; +import java.io.ObjectStreamClass; import java.util.Base64; public class KafkaCosmosUtils { private static final Logger LOGGER = LoggerFactory.getLogger(KafkaCosmosUtils.class); + private static final String ALLOWED_CLASSES = "CosmosClientMetadataCachesSnapshot"; public static CosmosClientMetadataCachesSnapshot getCosmosClientMetadataFromString(String metadataCacheString) { if (StringUtils.isNotEmpty(metadataCacheString)) { byte[] inputByteArray = Base64.getDecoder().decode(metadataCacheString); try (ObjectInputStream objectInputStream = - new ObjectInputStream(new ByteArrayInputStream(inputByteArray))) { - - return (CosmosClientMetadataCachesSnapshot) objectInputStream.readObject(); + new ObjectInputStream(new ByteArrayInputStream(inputByteArray)) { + @Override + protected Class resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException { + // Whitelist only allowed classes to prevent rce from arbitrary classes + if (!ALLOWED_CLASSES.contains(desc.getName())) { + throw new InvalidClassException("Unauthorized deserialization attempt", desc.getName()); + } + return super.resolveClass(desc); + } + }){ + return (CosmosClientMetadataCachesSnapshot) objectInputStream.readObject(); } catch (IOException | ClassNotFoundException e) { LOGGER.warn("Failed to deserialize cosmos client metadata cache snapshot"); return null; diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java index 5782819f79a5..b818e3f9ea55 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java @@ -18,7 +18,6 @@ import com.azure.cosmos.kafka.connect.implementation.sink.CosmosSinkTaskConfig; import com.azure.cosmos.kafka.connect.implementation.sink.IdStrategyType; import com.azure.cosmos.kafka.connect.implementation.sink.ItemWriteStrategy; -import com.azure.cosmos.kafka.connect.implementation.sink.patch.KafkaCosmosPatchOperationType; import com.azure.cosmos.models.CosmosContainerProperties; import org.apache.kafka.common.config.Config; import org.apache.kafka.common.config.ConfigDef; @@ -29,14 +28,23 @@ import org.testng.annotations.Test; import reactor.core.publisher.Mono; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InvalidClassException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; import java.util.ArrayList; import java.util.Arrays; +import java.util.Base64; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; @@ -214,6 +222,23 @@ public void taskConfigsForClientMetadataCachesSnapshot() { } } + @Test(groups = "unit") + public void evilDeserializationIsBlocked() throws Exception { + AtomicReference payload = new AtomicReference<>("Test RCE payload"); + Evil evil = new Evil(payload); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(baos)) { + oos.writeObject(evil); + } + String evilBase64 = Base64.getEncoder().encodeToString(baos.toByteArray()); + + // Through KafkaCosmosUtils: should be blocked and return null + CosmosClientMetadataCachesSnapshot snapshot = + KafkaCosmosUtils.getCosmosClientMetadataFromString(evilBase64); + assertThat(snapshot).isNull(); + assertThat(payload.get()).isEqualTo("Test RCE payload"); + } + @Test(groups = "unit") public void misFormattedConfig() { CosmosSinkConnector sinkConnector = new CosmosSinkConnector(); @@ -450,19 +475,6 @@ public static class SinkConfigs { "azure.cosmos.sink.write.strategy", ItemWriteStrategy.ITEM_OVERWRITE.getName(), true), - new KafkaCosmosConfigEntry( - "azure.cosmos.sink.write.patch.operationType.default", - KafkaCosmosPatchOperationType.SET.getName(), - true), - new KafkaCosmosConfigEntry( - "azure.cosmos.sink.write.patch.property.configs", - Strings.Emtpy, - true), - new KafkaCosmosConfigEntry( - "azure.cosmos.sink.write.patch.filter", - Strings.Emtpy, - true), - new KafkaCosmosConfigEntry("azure.cosmos.sink.maxRetryCount", 10, true), new KafkaCosmosConfigEntry("azure.cosmos.sink.database.name", null, false), new KafkaCosmosConfigEntry("azure.cosmos.sink.containers.topicMap", null, false), new KafkaCosmosConfigEntry( @@ -471,4 +483,26 @@ public static class SinkConfigs { true) ); } + + public static class Evil implements Serializable { + private static final long serialVersionUID = 1L; + + private final AtomicReference payload; + + public Evil(AtomicReference payload) { + this.payload = payload; + } + + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { + in.defaultReadObject(); + System.out.println("Payload executed"); + payload.set("Payload executed"); + } + + @Override + public String toString() { + return "Evil{payload='" + payload.get() + "'}"; + } + } + } From f1c4fb9ab175a5171ee1316c3b56b82d91b1b48b Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Dec 2025 14:31:42 -0500 Subject: [PATCH 2/8] add changelog --- .../azure-cosmos-kafka-connect/CHANGELOG.md | 1 + .../kafka/connect/CosmosSinkConnectorTest.java | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md index 0f8d76dfe6a2..dd1de2564019 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md @@ -7,6 +7,7 @@ #### Breaking Changes #### Bugs Fixed +* Added filtering to deserialization of `"azure.cosmos.client.metadata.caches.snapshot"` - See [PR 47594](https://github.com/Azure/azure-sdk-for-java/pull/47594) #### Other Changes diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java index b818e3f9ea55..96ad12c57940 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java @@ -18,6 +18,7 @@ import com.azure.cosmos.kafka.connect.implementation.sink.CosmosSinkTaskConfig; import com.azure.cosmos.kafka.connect.implementation.sink.IdStrategyType; import com.azure.cosmos.kafka.connect.implementation.sink.ItemWriteStrategy; +import com.azure.cosmos.kafka.connect.implementation.sink.patch.KafkaCosmosPatchOperationType; import com.azure.cosmos.models.CosmosContainerProperties; import org.apache.kafka.common.config.Config; import org.apache.kafka.common.config.ConfigDef; @@ -28,10 +29,8 @@ import org.testng.annotations.Test; import reactor.core.publisher.Mono; -import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.io.InvalidClassException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.io.Serializable; @@ -451,6 +450,19 @@ public static class SinkConfigs { new KafkaCosmosConfigEntry("azure.cosmos.application.name", Strings.Emtpy, true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.enabled", false, true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.account.endpoint", Strings.Emtpy, true), + new KafkaCosmosConfigEntry( + "azure.cosmos.sink.write.patch.operationType.default", + KafkaCosmosPatchOperationType.SET.getName(), + true), + new KafkaCosmosConfigEntry( + "azure.cosmos.sink.write.patch.property.configs", + Strings.Emtpy, + true), + new KafkaCosmosConfigEntry( + "azure.cosmos.sink.write.patch.filter", + Strings.Emtpy, + true), + new KafkaCosmosConfigEntry("azure.cosmos.sink.maxRetryCount", 10, true), new KafkaCosmosConfigEntry("azure.cosmos.throughputControl.account.tenantId", Strings.Emtpy, true), new KafkaCosmosConfigEntry("azure.cosmos.throughputControl.auth.type", CosmosAuthType.MASTER_KEY.getName(), true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.account.key", Strings.Emtpy, true, true), From 9764838d44fb2cc1cec9780bbb4965991e98f3ca Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Dec 2025 14:34:32 -0500 Subject: [PATCH 3/8] revert changes --- .../connect/CosmosSinkConnectorTest.java | 25 ++++++++++--------- 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java index 96ad12c57940..3c52549f49f7 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java @@ -450,18 +450,6 @@ public static class SinkConfigs { new KafkaCosmosConfigEntry("azure.cosmos.application.name", Strings.Emtpy, true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.enabled", false, true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.account.endpoint", Strings.Emtpy, true), - new KafkaCosmosConfigEntry( - "azure.cosmos.sink.write.patch.operationType.default", - KafkaCosmosPatchOperationType.SET.getName(), - true), - new KafkaCosmosConfigEntry( - "azure.cosmos.sink.write.patch.property.configs", - Strings.Emtpy, - true), - new KafkaCosmosConfigEntry( - "azure.cosmos.sink.write.patch.filter", - Strings.Emtpy, - true), new KafkaCosmosConfigEntry("azure.cosmos.sink.maxRetryCount", 10, true), new KafkaCosmosConfigEntry("azure.cosmos.throughputControl.account.tenantId", Strings.Emtpy, true), new KafkaCosmosConfigEntry("azure.cosmos.throughputControl.auth.type", CosmosAuthType.MASTER_KEY.getName(), true), @@ -487,6 +475,19 @@ public static class SinkConfigs { "azure.cosmos.sink.write.strategy", ItemWriteStrategy.ITEM_OVERWRITE.getName(), true), + new KafkaCosmosConfigEntry( + "azure.cosmos.sink.write.patch.operationType.default", + KafkaCosmosPatchOperationType.SET.getName(), + true), + new KafkaCosmosConfigEntry( + "azure.cosmos.sink.write.patch.property.configs", + Strings.Emtpy, + true), + new KafkaCosmosConfigEntry( + "azure.cosmos.sink.write.patch.filter", + Strings.Emtpy, + true), + new KafkaCosmosConfigEntry("azure.cosmos.sink.maxRetryCount", 10, true), new KafkaCosmosConfigEntry("azure.cosmos.sink.database.name", null, false), new KafkaCosmosConfigEntry("azure.cosmos.sink.containers.topicMap", null, false), new KafkaCosmosConfigEntry( From a9005a8a3744c734eb555c7e558024fe0265ee1d Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Dec 2025 14:35:03 -0500 Subject: [PATCH 4/8] revert changes --- .../com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java index 3c52549f49f7..e4488339a55c 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSinkConnectorTest.java @@ -450,7 +450,6 @@ public static class SinkConfigs { new KafkaCosmosConfigEntry("azure.cosmos.application.name", Strings.Emtpy, true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.enabled", false, true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.account.endpoint", Strings.Emtpy, true), - new KafkaCosmosConfigEntry("azure.cosmos.sink.maxRetryCount", 10, true), new KafkaCosmosConfigEntry("azure.cosmos.throughputControl.account.tenantId", Strings.Emtpy, true), new KafkaCosmosConfigEntry("azure.cosmos.throughputControl.auth.type", CosmosAuthType.MASTER_KEY.getName(), true), new KafkaCosmosConfigEntry<>("azure.cosmos.throughputControl.account.key", Strings.Emtpy, true, true), From faaf95715dae4203cb2784cea7043109a2275677 Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Dec 2025 14:51:17 -0500 Subject: [PATCH 5/8] change Allowed Classes to not be hard coded strings --- .../kafka/connect/implementation/KafkaCosmosUtils.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java index a1d9bc85fe8d..1ae03ae3e0a5 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java @@ -17,10 +17,15 @@ import java.io.ObjectOutputStream; import java.io.ObjectStreamClass; import java.util.Base64; +import java.util.HashSet; +import java.util.Set; public class KafkaCosmosUtils { private static final Logger LOGGER = LoggerFactory.getLogger(KafkaCosmosUtils.class); - private static final String ALLOWED_CLASSES = "CosmosClientMetadataCachesSnapshot"; + private static final Set ALLOWED_CLASSES = new HashSet<>(); + static { + ALLOWED_CLASSES.add(CosmosClientMetadataCachesSnapshot.class.getName()); + } public static CosmosClientMetadataCachesSnapshot getCosmosClientMetadataFromString(String metadataCacheString) { if (StringUtils.isNotEmpty(metadataCacheString)) { From 21daf8043cefb33277135819a2b9301a7fb5aec2 Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Dec 2025 15:38:36 -0500 Subject: [PATCH 6/8] fix tests and filter on byte array as well --- .../implementation/KafkaCosmosUtils.java | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java index 1ae03ae3e0a5..1dcc74777ebc 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java @@ -25,6 +25,7 @@ public class KafkaCosmosUtils { private static final Set ALLOWED_CLASSES = new HashSet<>(); static { ALLOWED_CLASSES.add(CosmosClientMetadataCachesSnapshot.class.getName()); + ALLOWED_CLASSES.add(byte[].class.getName()); } public static CosmosClientMetadataCachesSnapshot getCosmosClientMetadataFromString(String metadataCacheString) { @@ -32,16 +33,17 @@ public static CosmosClientMetadataCachesSnapshot getCosmosClientMetadataFromStri byte[] inputByteArray = Base64.getDecoder().decode(metadataCacheString); try (ObjectInputStream objectInputStream = new ObjectInputStream(new ByteArrayInputStream(inputByteArray)) { - @Override - protected Class resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException { - // Whitelist only allowed classes to prevent rce from arbitrary classes - if (!ALLOWED_CLASSES.contains(desc.getName())) { - throw new InvalidClassException("Unauthorized deserialization attempt", desc.getName()); + @Override + protected Class resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException { + // Whitelist only allowed classes to prevent rce from arbitrary classes + if (!ALLOWED_CLASSES.contains(desc.getName())) { + LOGGER.error(desc.getName()); + throw new InvalidClassException("Unauthorized deserialization attempt", desc.getName()); + } + return super.resolveClass(desc); } - return super.resolveClass(desc); - } - }){ - return (CosmosClientMetadataCachesSnapshot) objectInputStream.readObject(); + }) { + return (CosmosClientMetadataCachesSnapshot) objectInputStream.readObject(); } catch (IOException | ClassNotFoundException e) { LOGGER.warn("Failed to deserialize cosmos client metadata cache snapshot"); return null; From 0c1e7b764d211d24d09178122b4cb11883dc4d69 Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Dec 2025 17:43:51 -0500 Subject: [PATCH 7/8] fix indentation --- .../cosmos/kafka/connect/implementation/KafkaCosmosUtils.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java index 1dcc74777ebc..c0b7daca499a 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java @@ -42,7 +42,7 @@ protected Class resolveClass(ObjectStreamClass desc) throws IOException, Clas } return super.resolveClass(desc); } - }) { + }) { return (CosmosClientMetadataCachesSnapshot) objectInputStream.readObject(); } catch (IOException | ClassNotFoundException e) { LOGGER.warn("Failed to deserialize cosmos client metadata cache snapshot"); From 05562bec1da7ef93fab8ffdf050c018617a1ca57 Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 5 Jan 2026 10:06:15 -0500 Subject: [PATCH 8/8] react to comments --- .../cosmos/kafka/connect/implementation/KafkaCosmosUtils.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java index c0b7daca499a..6246b9a4566c 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosUtils.java @@ -37,7 +37,7 @@ public static CosmosClientMetadataCachesSnapshot getCosmosClientMetadataFromStri protected Class resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException { // Whitelist only allowed classes to prevent rce from arbitrary classes if (!ALLOWED_CLASSES.contains(desc.getName())) { - LOGGER.error(desc.getName()); + LOGGER.error("Invalid class type for deserialization {}", desc.getName()); throw new InvalidClassException("Unauthorized deserialization attempt", desc.getName()); } return super.resolveClass(desc);