From 9dbac2e57d09c11185db12974fd0543ca3381fd7 Mon Sep 17 00:00:00 2001 From: Ruben Quesada Lopez Date: Fri, 11 Sep 2026 17:25:01 +0100 Subject: [PATCH] [CALCITE-7773] Kafka adapter should validate class-naming keys in the "consumer.params" operand --- .../calcite/config/CalciteSystemProperty.java | 23 ++++ .../adapter/kafka/KafkaTableFactory.java | 66 ++++++++++- .../kafka/KafkaConsumerParamsTest.java | 109 ++++++++++++++++++ site/_docs/history.md | 11 ++ 4 files changed, 207 insertions(+), 2 deletions(-) create mode 100644 kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaConsumerParamsTest.java 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 9c039437474..ac3aa883ee6 100644 --- a/core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java +++ b/core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java @@ -509,6 +509,29 @@ public final class CalciteSystemProperty { public static final CalciteSystemProperty 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. + * + *

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. + * + *

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 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 2d6c3a2a103..ca2cce7c484 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) operand.get(KafkaTableConstants.SCHEMA_CONSUMER_PARAMS)); + final Map consumerParams = + (Map) 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 consumerParams) { + if (CalciteSystemProperty.KAFKA_CONSUMER_PARAMS_TRUSTED.value()) { + return; + } + for (Map.Entry 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 00000000000..8f24c70c12f --- /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 params(String key, String value) { + final Map params = new HashMap<>(); + params.put(key, value); + return params; + } + + @Test void testHarmlessConnectivityParamsAllowed() { + final Map 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 fabe61ed049..64afcdc0354 100644 --- a/site/_docs/history.md +++ b/site/_docs/history.md @@ -63,6 +63,17 @@ creation will fail with `IllegalArgumentException: class X is not annotated @Inp `KafkaTableFactory` row converter operand now requires the presence of a `public` constructor for instantiating the class. +* [CALCITE-7773] +`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. + * [CALCITE-7713] 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