TLS: allow protocol allowlist; harden JVM SSL init (#17776)

* Made TLS configurable

* Made TLS configurable

* Address PR feedback: trim protocols, restore diagnostics, add tests

* Address PR feedback: trim protocols, restore diagnostics, add tests

* Address PR feedback: trim protocols, restore diagnostics, add tests

---------

Co-authored-by: Shivam Phutela <shvmphutela05@gmail.com>

Author: shivam-startree

pinot-common/src/main/java/org/apache/pinot/common/config/TlsConfig.java

@@ -20,7 +20,9 @@
 
 import io.netty.handler.ssl.SslProvider;
 import java.security.KeyStore;
+import java.util.Arrays;
 import java.util.Objects;
+import javax.annotation.Nullable;
 import org.apache.commons.lang3.StringUtils;
 
 
@@ -38,6 +40,9 @@ public class TlsConfig {
   private String _sslProvider = SslProvider.JDK.toString();
   // If true, the client will not verify the server's certificate
   private boolean _insecure = false;
+  // Allowed TLS protocols (optional, if not set uses default)
+  @Nullable
+  private String[] _allowedProtocols = null;
 
   public TlsConfig() {
     // left blank
@@ -52,6 +57,10 @@ public TlsConfig(TlsConfig tlsConfig) {
     _trustStorePath = tlsConfig._trustStorePath;
     _trustStorePassword = tlsConfig._trustStorePassword;
     _sslProvider = tlsConfig._sslProvider;
+    _insecure = tlsConfig._insecure;
+    _allowedProtocols = tlsConfig._allowedProtocols != null
+        ? Arrays.copyOf(tlsConfig._allowedProtocols, tlsConfig._allowedProtocols.length)
+        : null;
   }
 
   public boolean isClientAuthEnabled() {
@@ -130,6 +139,17 @@ public void setInsecure(boolean insecure) {
     _insecure = insecure;
   }
 
+  @Nullable
+  public String[] getAllowedProtocols() {
+    return _allowedProtocols != null ? Arrays.copyOf(_allowedProtocols, _allowedProtocols.length) : null;
+  }
+
+  public void setAllowedProtocols(@Nullable String[] allowedProtocols) {
+    _allowedProtocols = allowedProtocols != null
+        ? Arrays.copyOf(allowedProtocols, allowedProtocols.length)
+        : null;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -139,16 +159,23 @@ public boolean equals(Object o) {
       return false;
     }
     TlsConfig tlsConfig = (TlsConfig) o;
-    return _clientAuthEnabled == tlsConfig._clientAuthEnabled && _insecure == tlsConfig._insecure && Objects.equals(
-        _keyStoreType, tlsConfig._keyStoreType) && Objects.equals(_keyStorePath, tlsConfig._keyStorePath)
-        && Objects.equals(_keyStorePassword, tlsConfig._keyStorePassword) && Objects.equals(_trustStoreType,
-        tlsConfig._trustStoreType) && Objects.equals(_trustStorePath, tlsConfig._trustStorePath) && Objects.equals(
-        _trustStorePassword, tlsConfig._trustStorePassword) && Objects.equals(_sslProvider, tlsConfig._sslProvider);
+    return _clientAuthEnabled == tlsConfig._clientAuthEnabled
+        && _insecure == tlsConfig._insecure
+        && Objects.equals(_keyStoreType, tlsConfig._keyStoreType)
+        && Objects.equals(_keyStorePath, tlsConfig._keyStorePath)
+        && Objects.equals(_keyStorePassword, tlsConfig._keyStorePassword)
+        && Objects.equals(_trustStoreType, tlsConfig._trustStoreType)
+        && Objects.equals(_trustStorePath, tlsConfig._trustStorePath)
+        && Objects.equals(_trustStorePassword, tlsConfig._trustStorePassword)
+        && Objects.equals(_sslProvider, tlsConfig._sslProvider)
+        && Arrays.equals(_allowedProtocols, tlsConfig._allowedProtocols);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(_clientAuthEnabled, _keyStoreType, _keyStorePath, _keyStorePassword, _trustStoreType,
+    int result = Objects.hash(_clientAuthEnabled, _keyStoreType, _keyStorePath, _keyStorePassword, _trustStoreType,
         _trustStorePath, _trustStorePassword, _sslProvider, _insecure);
+    result = 31 * result + Arrays.hashCode(_allowedProtocols);
+    return result;
   }
 }

pinot-common/src/main/java/org/apache/pinot/common/utils/tls/JvmDefaultSslContext.java

@@ -49,6 +49,12 @@ private JvmDefaultSslContext() {
    * system property and they are files:
    * set the default SSL context to the default SSL context created by SSLFactory, and enable auto renewal of
    * SSLFactory when either key store or trust store file changes.
+   *
+   * <p>This method is called from a static block in {@code PinotAdministrator}. Failures here must not
+   * prevent the JVM from starting — for example when security providers are not yet registered or
+   * the keystore/truststore is temporarily unavailable. In such cases, a warning is logged and Pinot
+   * components will configure TLS individually during their own startup.</p>
+   *
    * TODO: need to support "javax.net.ssl.keyStoreProvider", "javax.net.ssl.trustStoreProvider", "https.protocols" and
    *  "https.cipherSuites" system properties.
    */
@@ -59,6 +65,21 @@ public static synchronized void initDefaultSslContext() {
       return;
     }
 
+    try {
+      initDefaultSslContextInternal();
+    } catch (Exception e) {
+      // Do NOT rethrow — this runs in PinotAdministrator's static initializer.
+      // Rethrowing would cause ExceptionInInitializerError and prevent the JVM from starting.
+      // Pinot components re-initialize TLS individually during their own startup.
+      LOGGER.warn("Failed to initialize JVM default SSL context. This is non-fatal — Pinot components "
+          + "will configure TLS individually during startup. trustStoreType='{}', trustStore='{}'. Error: {}",
+          System.getProperty(JVM_TRUST_STORE_TYPE), System.getProperty(JVM_TRUST_STORE), e.getMessage());
+      LOGGER.debug("Full stack trace for SSL context initialization failure:", e);
+    }
+    _initialized = true;
+  }
+
+  private static void initDefaultSslContextInternal() {
     String jvmKeyStorePath = System.getProperty(JVM_KEY_STORE);
     String jvmTrustStorePath = System.getProperty(JVM_TRUST_STORE);
 
@@ -101,8 +122,10 @@ public static synchronized void initDefaultSslContext() {
               .map(String::trim).filter(StringUtils::isNotBlank).orElse(null);
       RenewableTlsUtils.enableAutoRenewalFromFileStoreForSSLFactory(jvmSslFactory, jvmKeystoreType, jvmKeyStorePath,
           jvmKeystorePassword, jvmTrustStoreType, jvmTrustStorePath, jvmTrustStorePassword, null, null, () -> false);
+      LOGGER.info("Successfully initialized JVM default SSL context");
+    } else {
+      LOGGER.info("No key store or trust store specified via system properties, "
+          + "skipping JVM default SSL context setup");
     }
-    _initialized = true;
-    LOGGER.info("Successfully initialized mvm default SSL context");
   }
 }

pinot-common/src/main/java/org/apache/pinot/common/utils/tls/TlsUtils.java

@@ -72,6 +72,7 @@ public final class TlsUtils {
   private static final String FILE_SCHEME_PREFIX = FILE_SCHEME + "://";
   private static final String FILE_SCHEME_PREFIX_WITHOUT_SLASH = FILE_SCHEME + ":";
   private static final String INSECURE = "insecure";
+  private static final String PROTOCOLS = "protocols";
 
   private static final AtomicReference<SSLContext> SSL_CONTEXT_REF = new AtomicReference<>();
   private static final Set<String> LOGGED_TLS_DIAGNOSTICS_KEYS = ConcurrentHashMap.newKeySet();
@@ -122,6 +123,22 @@ public static TlsConfig extractTlsConfig(PinotConfiguration pinotConfig, String
     tlsConfig.setInsecure(
         pinotConfig.getProperty(key(namespace, INSECURE), defaultConfig.isInsecure()));
 
+    // Read allowed TLS protocols from config (e.g., "TLSv1.2,TLSv1.3")
+    String protocolsConfig = pinotConfig.getProperty(key(namespace, PROTOCOLS));
+    if (StringUtils.isNotBlank(protocolsConfig)) {
+      String[] protocols = Arrays.stream(protocolsConfig.split(","))
+          .map(String::trim)
+          .filter(StringUtils::isNotBlank)
+          .toArray(String[]::new);
+      if (protocols.length > 0) {
+        tlsConfig.setAllowedProtocols(protocols);
+      } else if (defaultConfig.getAllowedProtocols() != null) {
+        tlsConfig.setAllowedProtocols(defaultConfig.getAllowedProtocols());
+      }
+    } else if (defaultConfig.getAllowedProtocols() != null) {
+      tlsConfig.setAllowedProtocols(defaultConfig.getAllowedProtocols());
+    }
+
     return tlsConfig;
   }
 
@@ -326,6 +343,14 @@ public static SslContext buildClientContext(TlsConfig tlsConfig) {
         SslContextBuilder.forClient().sslProvider(SslProvider.valueOf(tlsConfig.getSslProvider()));
     sslFactory.getKeyManagerFactory().ifPresent(sslContextBuilder::keyManager);
     sslFactory.getTrustManagerFactory().ifPresent(sslContextBuilder::trustManager);
+
+    // Apply protocol restrictions if configured
+    String[] allowedProtocols = tlsConfig.getAllowedProtocols();
+    if (allowedProtocols != null && allowedProtocols.length > 0) {
+      sslContextBuilder.protocols(allowedProtocols);
+      LOGGER.debug("TLS client context restricted to protocols: {}", Arrays.toString(allowedProtocols));
+    }
+
     try {
       warnIfNonJdkProviderConfiguredInternal("netty.client", tlsConfig);
       return sslContextBuilder.build();
@@ -352,6 +377,14 @@ public static SslContext buildServerContext(TlsConfig tlsConfig) {
     if (tlsConfig.isClientAuthEnabled()) {
       sslContextBuilder.clientAuth(ClientAuth.REQUIRE);
     }
+
+    // Apply protocol restrictions if configured
+    String[] allowedProtocols = tlsConfig.getAllowedProtocols();
+    if (allowedProtocols != null && allowedProtocols.length > 0) {
+      sslContextBuilder.protocols(allowedProtocols);
+      LOGGER.debug("TLS server context restricted to protocols: {}", Arrays.toString(allowedProtocols));
+    }
+
     try {
       warnIfNonJdkProviderConfiguredInternal("netty.server", tlsConfig);
       return sslContextBuilder.build();
@@ -435,7 +468,7 @@ private static void logTlsDiagnosticsOnce(String contextName, SSLContext sslCont
       return;
     }
 
-    // Basic “what are we actually using at runtime?” visibility.
+    // Basic "what are we actually using at runtime?" visibility.
     LOGGER.info(
         "TLS diagnostics ({}): SSLContext protocol='{}', provider='{}', configuredSslProvider='{}', insecure={}",
         contextName, protocol, providerName, configuredSslProvider, insecure);

pinot-common/src/test/java/org/apache/pinot/common/config/TlsConfigTest.java

@@ -0,0 +1,57 @@
+/**
+ * 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.pinot.common.config;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TlsConfigTest {
+  @Test
+  public void copyConstructorShouldCopyInsecureAndAllowedProtocols() {
+    TlsConfig original = new TlsConfig();
+    original.setInsecure(true);
+    original.setAllowedProtocols(new String[]{"TLSv1.2", "TLSv1.3"});
+
+    TlsConfig copy = new TlsConfig(original);
+    Assert.assertTrue(copy.isInsecure());
+    Assert.assertEquals(copy.getAllowedProtocols(), new String[]{"TLSv1.2", "TLSv1.3"});
+
+    // Ensure returned array is a defensive copy
+    String[] protocols = copy.getAllowedProtocols();
+    protocols[0] = "MUTATED";
+    Assert.assertEquals(copy.getAllowedProtocols(), new String[]{"TLSv1.2", "TLSv1.3"});
+  }
+
+  @Test
+  public void equalsAndHashCodeShouldIncludeAllowedProtocols() {
+    TlsConfig a = new TlsConfig();
+    a.setInsecure(true);
+    a.setAllowedProtocols(new String[]{"TLSv1.2"});
+
+    TlsConfig b = new TlsConfig();
+    b.setInsecure(true);
+    b.setAllowedProtocols(new String[]{"TLSv1.2"});
+
+    Assert.assertEquals(a, b);
+    Assert.assertEquals(a.hashCode(), b.hashCode());
+
+    b.setAllowedProtocols(new String[]{"TLSv1.3"});
+    Assert.assertNotEquals(a, b);
+  }
+}

pinot-common/src/test/java/org/apache/pinot/common/utils/tls/JvmDefaultSslContextTest.java

@@ -0,0 +1,60 @@
+/**
+ * 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.pinot.common.utils.tls;
+
+import java.lang.reflect.Field;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+public class JvmDefaultSslContextTest {
+  private static final String JVM_TRUST_STORE = "javax.net.ssl.trustStore";
+
+  @AfterMethod
+  public void cleanup() throws Exception {
+    System.clearProperty(JVM_TRUST_STORE);
+    setInitialized(false);
+  }
+
+  @Test
+  public void initDefaultSslContextShouldNotThrowWhenInitializationFails() throws Exception {
+    setInitialized(false);
+    System.setProperty(JVM_TRUST_STORE, "/does/not/exist");
+
+    // Should not throw (used from PinotAdministrator static initializer)
+    JvmDefaultSslContext.initDefaultSslContext();
+    Assert.assertTrue(isInitialized());
+
+    // Subsequent calls should be short-circuited
+    JvmDefaultSslContext.initDefaultSslContext();
+    Assert.assertTrue(isInitialized());
+  }
+
+  private static boolean isInitialized() throws Exception {
+    Field f = JvmDefaultSslContext.class.getDeclaredField("_initialized");
+    f.setAccessible(true);
+    return (boolean) f.get(null);
+  }
+
+  private static void setInitialized(boolean value) throws Exception {
+    Field f = JvmDefaultSslContext.class.getDeclaredField("_initialized");
+    f.setAccessible(true);
+    f.set(null, value);
+  }
+}

pinot-common/src/test/java/org/apache/pinot/common/utils/tls/TlsUtilsTest.java

@@ -18,6 +18,11 @@
  */
 package org.apache.pinot.common.utils.tls;
 
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.config.TlsConfig;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.testng.Assert;
 import org.testng.annotations.Test;
 
 
@@ -26,4 +31,22 @@ public class TlsUtilsTest {
   public void installDefaultSSLSocketFactoryWhenNoKeyOrTrustStore() {
     TlsUtils.installDefaultSSLSocketFactory(null, null, null, null, null, null);
   }
+
+  @Test
+  public void extractTlsConfigShouldTrimProtocols() {
+    Map<String, Object> props = new HashMap<>();
+    props.put("tls.protocols", " TLSv1.2, TLSv1.3 ,, ");
+    TlsConfig tlsConfig = TlsUtils.extractTlsConfig(new PinotConfiguration(props), "tls");
+    Assert.assertEquals(tlsConfig.getAllowedProtocols(), new String[]{"TLSv1.2", "TLSv1.3"});
+  }
+
+  @Test
+  public void extractTlsConfigShouldFallbackToDefaultProtocolsWhenEmpty() {
+    Map<String, Object> props = new HashMap<>();
+    props.put("tls.protocols", " , ");
+    TlsConfig defaultConfig = new TlsConfig();
+    defaultConfig.setAllowedProtocols(new String[]{"TLSv1.2"});
+    TlsConfig tlsConfig = TlsUtils.extractTlsConfig(new PinotConfiguration(props), "tls", defaultConfig);
+    Assert.assertEquals(tlsConfig.getAllowedProtocols(), new String[]{"TLSv1.2"});
+  }
 }

pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtils.java

@@ -59,7 +59,8 @@ private KafkaSSLUtils() {
   private static final String DEFAULT_KEY_ALGORITHM = "RSA";
   private static final String DEFAULT_KEYSTORE_TYPE = "PKCS12";
   private static final String DEFAULT_SECURITY_PROTOCOL = "SSL";
-  private static final String DEFAULT_TRUSTSTORE_TYPE = "jks";
+  // Follow the JVM default keystore type (typically "jks") unless explicitly configured.
+  private static final String DEFAULT_TRUSTSTORE_TYPE = KeyStore.getDefaultType();
   private static final String DEFAULT_SERVER_ALIAS = "ServerAlias";
   private static final String DEFAULT_CLIENT_ALIAS = "ClientAlias";
   // Key constants