This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch site
in repository https://gitbox.apache.org/repos/asf/calcite.git
The following commit(s) were added to refs/heads/site by this push:
new 746232a52f [CALCITE-7773] Kafka adapter should validate class-naming
keys in the "consumer.params" operand
746232a52f is described below
commit 746232a52fefe96bed30c533491557970a528f43
Author: Ruben Quesada Lopez <[email protected]>
AuthorDate: Fri Sep 11 17:25:01 2026 +0100
[CALCITE-7773] Kafka adapter should validate class-naming keys in the
"consumer.params" operand
---
.../calcite/config/CalciteSystemProperty.java | 23 +++++
.../calcite/adapter/kafka/KafkaTableFactory.java | 66 ++++++++++++-
.../adapter/kafka/KafkaConsumerParamsTest.java | 109 +++++++++++++++++++++
site/_docs/history.md | 11 +++
4 files changed, 207 insertions(+), 2 deletions(-)
diff --git
a/core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java
b/core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java
index b5f9407509..6b3adc32de 100644
--- a/core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java
+++ b/core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java
@@ -497,6 +497,29 @@ public final class CalciteSystemProperty<T> {
public static final CalciteSystemProperty<String> MODEL_BASE_DIRECTORY =
stringProperty("calcite.model.baseDirectory", "");
+ /**
+ * Whether the Kafka adapter forwards the {@code consumer.params} model
+ * operand to the Kafka client without filtering it.
+ *
+ * <p>The operand comes from the (query-author-supplied) model, and several
+ * Kafka consumer config keys ({@code key.deserializer},
+ * {@code value.deserializer}, {@code interceptor.classes},
+ * {@code metric.reporters}, {@code partition.assignment.strategy},
+ * {@code sasl.jaas.config}, {@code security.providers}, and any key
+ * ending in {@code .class} or {@code .classes}) cause the Kafka client
+ * to load and initialize classes named in the operand.
+ *
+ * <p>By default (empty / {@code false}) the Kafka adapter rejects those
+ * keys in {@code consumer.params}, and only accepts
+ * {@code key.deserializer}/{@code value.deserializer} after checking that
+ * the named class implements
+ * {@code org.apache.kafka.common.serialization.Deserializer} without
+ * initializing it. Set this property to {@code true} only when models
+ * come exclusively from trusted operators.
+ */
+ public static final CalciteSystemProperty<Boolean>
KAFKA_CONSUMER_PARAMS_TRUSTED =
+ booleanProperty("calcite.kafka.consumer.params.trusted", false);
+
/**
* Maximum number of decimal digits that the plain-notation expansion of a
{@code DECIMAL}
* literal may contain.
diff --git
a/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
b/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
index 2d6c3a2a10..ca2cce7c48 100644
---
a/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
+++
b/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
@@ -17,12 +17,15 @@
package org.apache.calcite.adapter.kafka;
import org.apache.calcite.avatica.AvaticaUtils;
+import org.apache.calcite.config.CalciteSystemProperty;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.TableFactory;
import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.common.serialization.Deserializer;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -59,8 +62,10 @@ public KafkaTableFactory() {
tableOptionBuilder.setRowConverter(rowConverter);
if (operand.containsKey(KafkaTableConstants.SCHEMA_CONSUMER_PARAMS)) {
- tableOptionBuilder.setConsumerParams(
- (Map<String, String>)
operand.get(KafkaTableConstants.SCHEMA_CONSUMER_PARAMS));
+ final Map<String, String> consumerParams =
+ (Map<String, String>)
operand.get(KafkaTableConstants.SCHEMA_CONSUMER_PARAMS);
+ checkConsumerParams(consumerParams);
+ tableOptionBuilder.setConsumerParams(consumerParams);
}
if (operand.containsKey(KafkaTableConstants.SCHEMA_CUST_CONSUMER)) {
String custConsumerClass = (String)
operand.get(KafkaTableConstants.SCHEMA_CUST_CONSUMER);
@@ -84,4 +89,61 @@ public KafkaTableFactory() {
return new KafkaStreamTable(tableOptionBuilder);
}
+
+ /** Rejects entries of the {@code consumer.params} operand that would make
+ * the Kafka client load classes named by the model author, unless system
+ * property
+ * {@link CalciteSystemProperty#KAFKA_CONSUMER_PARAMS_TRUSTED} is set. Key
+ * and value deserializers are allowed if the named class implements
+ * {@link Deserializer}. */
+ static void checkConsumerParams(Map<String, String> consumerParams) {
+ if (CalciteSystemProperty.KAFKA_CONSUMER_PARAMS_TRUSTED.value()) {
+ return;
+ }
+ for (Map.Entry<String, String> entry : consumerParams.entrySet()) {
+ final String key = entry.getKey();
+ if (key.equals(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+ || key.equals(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)) {
+ checkDeserializer(key, entry.getValue());
+ } else if (isClassLoadingParam(key)) {
+ throw new SecurityException("Consumer parameter '" + key
+ + "' can make the Kafka client load and run classes named in"
+ + " the model, so it is not allowed in the 'consumer.params'"
+ + " operand; configure it on the operator side, or set system"
+ + " property 'calcite.kafka.consumer.params.trusted' to 'true'"
+ + " if models are trusted");
+ }
+ }
+ }
+
+ /** Returns whether a Kafka consumer configuration key causes classes
+ * named in its value (or, for JAAS, in the login-module configuration
+ * text) to be loaded and instantiated by the Kafka client. */
+ private static boolean isClassLoadingParam(String key) {
+ return key.endsWith(".class")
+ || key.endsWith(".classes")
+ || key.equals("sasl.jaas.config")
+ || key.equals("security.providers")
+ || key.equals("metric.reporters")
+ || key.equals("partition.assignment.strategy");
+ }
+
+ /** Checks that a deserializer named in {@code consumer.params} implements
+ * {@link Deserializer} before the Kafka client loads (and initializes) it.
*/
+ private static void checkDeserializer(String key, String className) {
+ final Class<?> klass;
+ try {
+ klass =
+ Class.forName(className, false,
+ KafkaTableFactory.class.getClassLoader());
+ } catch (ClassNotFoundException e) {
+ throw new SecurityException("Deserializer class '" + className
+ + "' given as consumer parameter '" + key + "' not found", e);
+ }
+ if (!Deserializer.class.isAssignableFrom(klass)) {
+ throw new SecurityException("Class '" + className
+ + "' given as consumer parameter '" + key
+ + "' does not implement " + Deserializer.class.getName());
+ }
+ }
}
diff --git
a/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaConsumerParamsTest.java
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaConsumerParamsTest.java
new file mode 100644
index 0000000000..8f24c70c12
--- /dev/null
+++
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaConsumerParamsTest.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.adapter.kafka;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Tests for the {@code consumer.params} filtering in
+ * {@link KafkaTableFactory}: parameters that make the Kafka client load
+ * classes named in the model must be rejected by default.
+ */
+class KafkaConsumerParamsTest {
+ /** Set by {@link NotADeserializer}'s static initializer. A class named as
+ * a deserializer in {@code consumer.params} must not have any of its code
+ * run unless it implements {@code Deserializer}. */
+ static final AtomicBoolean NOT_A_DESERIALIZER_INITIALIZED = new
AtomicBoolean(false);
+
+ /** Stands in for an arbitrary untrusted class named as a deserializer. */
+ public static class NotADeserializer {
+ static {
+ NOT_A_DESERIALIZER_INITIALIZED.set(true);
+ }
+ }
+
+ private static Map<String, String> params(String key, String value) {
+ final Map<String, String> params = new HashMap<>();
+ params.put(key, value);
+ return params;
+ }
+
+ @Test void testHarmlessConnectivityParamsAllowed() {
+ final Map<String, String> params = new HashMap<>();
+ params.put("group.id", "g");
+ params.put("max.poll.records", "100");
+ params.put("security.protocol", "PLAINTEXT");
+ assertDoesNotThrow(() -> KafkaTableFactory.checkConsumerParams(params));
+ }
+
+ @Test void testJaasConfigRejected() {
+ SecurityException e =
+ assertThrows(SecurityException.class, () ->
+ KafkaTableFactory.checkConsumerParams(
+ params("sasl.jaas.config",
+ "com.sun.security.auth.module.JndiLoginModule required"
+ + " user.provider.url=\"ldap://example/o\";";)));
+ assertThat(e.getMessage(), containsString("sasl.jaas.config"));
+ }
+
+ @Test void testInterceptorClassesRejected() {
+ SecurityException e =
+ assertThrows(SecurityException.class, () ->
+ KafkaTableFactory.checkConsumerParams(
+ params("interceptor.classes", "com.bad.Interceptor")));
+ assertThat(e.getMessage(), containsString("interceptor.classes"));
+ }
+
+ @Test void testCallbackHandlerClassRejected() {
+ SecurityException e =
+ assertThrows(SecurityException.class, () ->
+ KafkaTableFactory.checkConsumerParams(
+ params("sasl.client.callback.handler.class",
"com.bad.Handler")));
+ assertThat(e.getMessage(),
containsString("sasl.client.callback.handler.class"));
+ }
+
+ /** A deserializer that is not actually a {@code Deserializer} is rejected
+ * without being initialized. */
+ @Test void testNonDeserializerClassRejectedUninitialized() {
+ SecurityException e =
+ assertThrows(SecurityException.class, () ->
+ KafkaTableFactory.checkConsumerParams(
+ params("value.deserializer",
NotADeserializer.class.getName())));
+ assertThat(e.getMessage(),
containsString(NotADeserializer.class.getName()));
+ assertThat("static initializer of a rejected class must not run",
+ NOT_A_DESERIALIZER_INITIALIZED.get(), is(false));
+ }
+
+ /** A genuine {@code Deserializer} implementation is still accepted. */
+ @Test void testRealDeserializerAllowed() {
+ assertDoesNotThrow(() ->
+ KafkaTableFactory.checkConsumerParams(
+ params("key.deserializer", StringDeserializer.class.getName())));
+ }
+}
diff --git a/site/_docs/history.md b/site/_docs/history.md
index 32a7f0d4fb..a2c769c234 100644
--- a/site/_docs/history.md
+++ b/site/_docs/history.md
@@ -63,6 +63,17 @@ #### Breaking Changes
`KafkaTableFactory` row converter operand now requires the presence of a
`public`
constructor for instantiating the class.
+* [<a
href="https://issues.apache.org/jira/browse/CALCITE-7773";>CALCITE-7773</a>]
+`KafkaTableFactory` now validates the `consumer.params` operand at table
+creation. Keys whose values name a class the Kafka client would load and
+initialize (deserializers, interceptors, metric reporters, partition
+assignment strategies, callback handlers, JAAS/security providers, and any
+key ending in `.class` or `.classes`) are rejected with `SecurityException`.
+`key.deserializer` and `value.deserializer` remain accepted after an
+interface check. Set `-Dcalcite.kafka.consumer.params.trusted=true` to
+restore the previous forwarding behavior when models are entirely
+operator-supplied.
+
* [<a
href="https://issues.apache.org/jira/browse/CALCITE-7713";>CALCITE-7713</a>]
Class loading from model files has been disabled by default. Any attempt to
load
classes from model files will lead to `SecurityException` unless an appropriate
