[CALCITE-7773] Kafka adapter should validate class-naming keys in the "consumer.params" operand

core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java

@@ -509,6 +509,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.

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 @@
     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 @@
 
     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());
+    }
+  }
 }

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())));
+  }
+}

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