You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/10/06 15:45:56 UTC
(fluss) 01/02: [FIP-29] Add SSL config parsing and SslContext/SslHandler factory
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 2f1d36cff07f29b25d9db06fa7ad9a7263c51273
Author: MicheleGuerriero <mi...@ververica.com>
AuthorDate: Tue Oct 6 15:02:10 2026 +0200
[FIP-29] Add SSL config parsing and SslContext/SslHandler factory
---
.../org/apache/fluss/config/ConfigOptions.java | 164 +++++
fluss-rpc/pom.xml | 33 +
.../org/apache/fluss/rpc/netty/ssl/SslConfig.java | 524 ++++++++++++++
.../fluss/rpc/netty/ssl/SslContextFactory.java | 379 ++++++++++
.../fluss/rpc/netty/ssl/SslContextFactoryTest.java | 800 +++++++++++++++++++++
.../apache/fluss/rpc/netty/ssl/TestSslUtils.java | 138 ++++
6 files changed, 2038 insertions(+)
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index d536e101ab..e497ea0703 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -656,6 +656,89 @@ public class ConfigOptions {
+ "from 'security.sasl.plain.credentials' when that credential map is set, "
+ "and can also be configured directly for compatibility.");
+ // ------------------------------------------------------------------------
+ // Server-side TLS (transport encryption + mTLS) options
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<List<String>> SERVER_SSL_ENABLED_LISTENERS =
+ key("security.ssl.enabled.listeners")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "Listener names with TLS enabled, e.g. `CLIENT,INTERNAL`; others "
+ + "accept plaintext. Requires `security.ssl.keystore.path`. A "
+ + "listener whose `security.protocol.map` entry is `mTLS` must "
+ + "be listed here, since it authenticates by certificate. Names "
+ + "match exactly: spell and case them as in `bind.listeners` "
+ + "and `security.protocol.map`, or no listener is selected.");
+
+ public static final ConfigOption<List<String>> SERVER_SSL_ENABLED_PROTOCOLS =
+ key("security.ssl.enabled.protocols")
+ .stringType()
+ .asList()
+ .defaultValues("TLSv1.2", "TLSv1.3")
+ .withDescription(
+ "TLS protocols enabled for incoming connections. `TLSv1.2` is kept "
+ + "by default for JDK 8 builds older than 8u261.");
+
+ public static final ConfigOption<List<String>> SERVER_SSL_CIPHER_SUITES =
+ key("security.ssl.cipher.suites")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "Cipher suites enabled for TLS connections. Empty uses the provider "
+ + "defaults.");
+
+ public static final ConfigOption<String> SERVER_SSL_KEYSTORE_PATH =
+ key("security.ssl.keystore.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Keystore file with the server certificate and key. Required with "
+ + "`security.ssl.enabled.listeners`.");
+
+ public static final ConfigOption<Password> SERVER_SSL_KEYSTORE_PASSWORD =
+ key("security.ssl.keystore.password")
+ .passwordType()
+ .noDefaultValue()
+ .withDescription("The password to access the server keystore.");
+
+ public static final ConfigOption<String> SERVER_SSL_KEYSTORE_TYPE =
+ key("security.ssl.keystore.type")
+ .stringType()
+ .defaultValue("JKS")
+ .withDescription("Format of the server keystore: `JKS` or `PKCS12`.");
+
+ public static final ConfigOption<Password> SERVER_SSL_KEY_PASSWORD =
+ key("security.ssl.key.password")
+ .passwordType()
+ .noDefaultValue()
+ .withDescription(
+ "Password of the server private key. Defaults to "
+ + "`security.ssl.keystore.password`.");
+
+ public static final ConfigOption<String> SERVER_SSL_TRUSTSTORE_PATH =
+ key("security.ssl.truststore.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Truststore file holding the certificates the server trusts. Required "
+ + "for `mTLS` listeners, to validate client certificates.");
+
+ public static final ConfigOption<Password> SERVER_SSL_TRUSTSTORE_PASSWORD =
+ key("security.ssl.truststore.password")
+ .passwordType()
+ .noDefaultValue()
+ .withDescription("The password to access the server truststore.");
+
+ public static final ConfigOption<String> SERVER_SSL_TRUSTSTORE_TYPE =
+ key("security.ssl.truststore.type")
+ .stringType()
+ .defaultValue("JKS")
+ .withDescription("Format of the server truststore: `JKS` or `PKCS12`.");
+
public static final ConfigOption<Integer> TABLET_SERVER_ID =
key("tablet-server.id")
.intType()
@@ -1509,6 +1592,87 @@ public class ConfigOptions {
+ "the server JVM but still need to load authentication plugins "
+ "shipped in plugins/.");
+ // ------------------------------------------------------------------------
+ // Client-side TLS (transport encryption + mTLS) options
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<Boolean> CLIENT_SSL_ENABLED =
+ key("client.security.ssl.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether the client connects over TLS. Must match the listener it "
+ + "connects to (`security.ssl.enabled.listeners`).");
+
+ public static final ConfigOption<List<String>> CLIENT_SSL_ENABLED_PROTOCOLS =
+ key("client.security.ssl.enabled.protocols")
+ .stringType()
+ .asList()
+ .defaultValues("TLSv1.2", "TLSv1.3")
+ .withDescription("TLS protocols enabled for client connections.");
+
+ public static final ConfigOption<List<String>> CLIENT_SSL_CIPHER_SUITES =
+ key("client.security.ssl.cipher.suites")
+ .stringType()
+ .asList()
+ .defaultValues()
+ .withDescription("Cipher suites enabled for client connections.");
+
+ public static final ConfigOption<String> CLIENT_SSL_TRUSTSTORE_PATH =
+ key("client.security.ssl.truststore.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Truststore file holding the certificates the client trusts. If "
+ + "unset, the JVM default truststore is used.");
+
+ public static final ConfigOption<String> CLIENT_SSL_TRUSTSTORE_PASSWORD =
+ key("client.security.ssl.truststore.password")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The password to access the client truststore.");
+
+ public static final ConfigOption<String> CLIENT_SSL_TRUSTSTORE_TYPE =
+ key("client.security.ssl.truststore.type")
+ .stringType()
+ .defaultValue("JKS")
+ .withDescription("Format of the client truststore: `JKS` or `PKCS12`.");
+
+ public static final ConfigOption<String> CLIENT_SSL_KEYSTORE_PATH =
+ key("client.security.ssl.keystore.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Keystore file with the client certificate and key.");
+
+ public static final ConfigOption<String> CLIENT_SSL_KEYSTORE_PASSWORD =
+ key("client.security.ssl.keystore.password")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The password to access the client keystore.");
+
+ public static final ConfigOption<String> CLIENT_SSL_KEYSTORE_TYPE =
+ key("client.security.ssl.keystore.type")
+ .stringType()
+ .defaultValue("JKS")
+ .withDescription("Format of the client keystore: `JKS` or `PKCS12`.");
+
+ public static final ConfigOption<String> CLIENT_SSL_KEY_PASSWORD =
+ key("client.security.ssl.key.password")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Password of the client private key. Defaults to "
+ + "`client.security.ssl.keystore.password`.");
+
+ public static final ConfigOption<String> CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM =
+ key("client.security.ssl.endpoint.identification.algorithm")
+ .stringType()
+ .defaultValue("https")
+ .withDescription(
+ "Algorithm verifying the server hostname against its certificate: "
+ + "`https` or `ldaps`. Empty disables the check, which is not "
+ + "recommended in production.");
+
public static final ConfigOption<MemorySize> CLIENT_SCANNER_LOG_FETCH_MAX_BYTES =
key("client.scanner.log.fetch.max-bytes")
.memoryType()
diff --git a/fluss-rpc/pom.xml b/fluss-rpc/pom.xml
index 369d7ce5ab..f41bbec203 100644
--- a/fluss-rpc/pom.xml
+++ b/fluss-rpc/pom.xml
@@ -33,6 +33,10 @@
The Fluss RPC framework and Netty-based implementation.
</description>
+ <properties>
+ <bouncycastle.version>1.85</bouncycastle.version>
+ </properties>
+
<dependencies>
<dependency>
<groupId>org.apache.fluss</groupId>
@@ -65,6 +69,35 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+
+ <!--
+ Netty's SelfSignedCertificate, used by the TLS tests to generate key material, prefers
+ BouncyCastle and only falls back to sun.security.x509 internals, which are inaccessible
+ on JDK 9+ and unusable from JDK 18 on. Test scope: no BouncyCastle is shipped.
+ bcutil and bcprov are bcpkix's own transitive dependencies, declared here only to pin
+ them: bcpkix depends on them through version ranges, which resolve differently over time
+ and make every build fetch their maven-metadata.xml.
+ -->
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcpkix-jdk18on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcutil-jdk18on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-jdk18on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslConfig.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslConfig.java
new file mode 100644
index 0000000000..95a089a8be
--- /dev/null
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslConfig.java
@@ -0,0 +1,524 @@
+/*
+ * 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.fluss.rpc.netty.ssl;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOption;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.Password;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.IllegalConfigurationException;
+
+import javax.annotation.Nullable;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.Security;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+
+/**
+ * The parsed and validated TLS configuration for one side (server or client) of an RPC connection.
+ *
+ * <p>This is a thin, immutable holder over the {@code security.ssl.*} (server) and {@code
+ * client.security.ssl.*} (client) {@link ConfigOptions}. Whether TLS is enabled at all is also
+ * encoded here: the {@link #fromServerConfig} / {@link #fromClientConfig} factory methods return
+ * {@link Optional#empty()} when TLS is not enabled (no listener listed in {@code
+ * security.ssl.enabled.listeners}, resp. {@code client.security.ssl.enabled} being false), so
+ * callers never have to consult the raw configuration to make that decision. Validation that does
+ * not depend on the actual key material (e.g. "a keystore must be configured for a TLS server",
+ * "every configured TLS protocol and cipher suite is one this JVM supports") happens in the same
+ * factory methods so misconfiguration fails fast with a clear message, instead of surfacing later
+ * as an engine-level error on every connection.
+ *
+ * <p>The per-listener client-certificate requirement <b>is</b> held here: a listener whose {@code
+ * security.protocol.map} entry is {@code mTLS} (matched by an exact listener name) must also be
+ * listed in {@code security.ssl.enabled.listeners}, since certificate authentication has no
+ * transport to run on otherwise. That relationship is validated even when no listener enables TLS,
+ * so {@link #fromServerConfig} can reject a configuration it would otherwise report as having no
+ * TLS at all. Such a listener requires a client certificate, which the server can only validate
+ * against an explicitly configured truststore — so such a listener without {@code
+ * security.ssl.truststore.path} is rejected here rather than silently falling back to the JVM
+ * default truststore. The server pipeline reads the requirement per listener via {@link
+ * #requiresClientAuth(String)} instead of re-deriving it from the raw configuration.
+ */
+@Internal
+public final class SslConfig {
+
+ /**
+ * The {@code security.protocol.map} authentication protocol that requires a client certificate.
+ * Matched case-insensitively, as {@code AuthenticationFactory} matches authentication protocol
+ * names.
+ */
+ private static final String MUTUAL_TLS_AUTH_PROTOCOL = "mTLS";
+
+ /** The endpoint identification algorithms the JDK implements for TLS. */
+ private static final List<String> ENDPOINT_IDENTIFICATION_ALGORITHMS =
+ Arrays.asList("https", "ldaps");
+
+ /** Server-only: the listener names for which TLS is enabled (empty for a client config). */
+ private final List<String> enabledListeners;
+
+ /**
+ * Server-only: the subset of {@link #enabledListeners} that requires a client certificate
+ * (empty for a client config).
+ */
+ private final Set<String> clientAuthListeners;
+
+ private final List<String> enabledProtocols;
+ private final List<String> cipherSuites;
+
+ @Nullable private final String keystorePath;
+ @Nullable private final String keystorePassword;
+ private final String keystoreType;
+ @Nullable private final String keyPassword;
+
+ @Nullable private final String truststorePath;
+ @Nullable private final String truststorePassword;
+ private final String truststoreType;
+
+ /**
+ * Client-only: the endpoint identification algorithm (empty disables hostname verification).
+ */
+ private final String endpointIdentificationAlgorithm;
+
+ private SslConfig(
+ List<String> enabledListeners,
+ Set<String> clientAuthListeners,
+ List<String> enabledProtocols,
+ List<String> cipherSuites,
+ @Nullable String keystorePath,
+ @Nullable String keystorePassword,
+ String keystoreType,
+ @Nullable String keyPassword,
+ @Nullable String truststorePath,
+ @Nullable String truststorePassword,
+ String truststoreType,
+ String endpointIdentificationAlgorithm) {
+ this.enabledListeners = enabledListeners;
+ this.clientAuthListeners = clientAuthListeners;
+ this.enabledProtocols = enabledProtocols;
+ this.cipherSuites = cipherSuites;
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.keystoreType = keystoreType;
+ this.keyPassword = keyPassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ this.truststoreType = truststoreType;
+ this.endpointIdentificationAlgorithm = endpointIdentificationAlgorithm;
+ }
+
+ /**
+ * Build and validate the server-side TLS configuration, or {@link Optional#empty()} when TLS is
+ * not enabled for any listener (i.e. {@code security.ssl.enabled.listeners} is unset or empty).
+ */
+ public static Optional<SslConfig> fromServerConfig(Configuration conf) {
+ List<String> enabledListeners =
+ orEmpty(conf.get(ConfigOptions.SERVER_SSL_ENABLED_LISTENERS));
+
+ // The listener name is looked up exactly and the protocol name compared ignoring case,
+ // because that is how each is resolved at runtime: FlussProtocolPlugin selects a listener's
+ // authenticator with a plain map lookup on security.protocol.map, while
+ // AuthenticationFactory matches a plugin to a protocol name with equalsIgnoreCase.
+ // Matching listener names loosely here would classify a listener as mTLS that the server
+ // then authenticates as PLAINTEXT.
+ Set<String> clientAuthListeners =
+ conf.get(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP).entrySet().stream()
+ .filter(
+ entry ->
+ MUTUAL_TLS_AUTH_PROTOCOL.equalsIgnoreCase(entry.getValue()))
+ .map(Map.Entry::getKey)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+
+ // Checked before TLS is known to be enabled at all: a listener that authenticates clients
+ // by certificate has no way to obtain one without TLS transport underneath it.
+ Set<String> withoutTls = new LinkedHashSet<>(clientAuthListeners);
+ withoutTls.removeAll(enabledListeners);
+ if (!withoutTls.isEmpty()) {
+ throw new IllegalConfigurationException(
+ "Listener(s) %s use the %s authentication protocol but are not listed in '%s'. "
+ + "%s requires TLS transport, so those listeners can neither present "
+ + "nor validate certificates.",
+ withoutTls,
+ MUTUAL_TLS_AUTH_PROTOCOL,
+ ConfigOptions.SERVER_SSL_ENABLED_LISTENERS.key(),
+ MUTUAL_TLS_AUTH_PROTOCOL);
+ }
+
+ if (enabledListeners.isEmpty()) {
+ return Optional.empty();
+ }
+
+ String keystorePath = conf.getString(ConfigOptions.SERVER_SSL_KEYSTORE_PATH);
+ if (keystorePath == null) {
+ throw new IllegalConfigurationException(
+ "'%s' must be configured when any listener enables TLS via '%s'.",
+ ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key(),
+ ConfigOptions.SERVER_SSL_ENABLED_LISTENERS.key());
+ }
+
+ String truststorePath = conf.getString(ConfigOptions.SERVER_SSL_TRUSTSTORE_PATH);
+ if (!clientAuthListeners.isEmpty() && truststorePath == null) {
+ throw new IllegalConfigurationException(
+ "'%s' must be configured to validate client certificates for the %s listener(s) %s. "
+ + "Without it the server would validate client certificates against the JVM "
+ + "default truststore, accepting any certificate issued by a public CA.",
+ ConfigOptions.SERVER_SSL_TRUSTSTORE_PATH.key(),
+ MUTUAL_TLS_AUTH_PROTOCOL,
+ clientAuthListeners);
+ }
+
+ List<String> enabledProtocols =
+ effectiveProtocols(conf, ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS);
+ List<String> cipherSuites = orEmpty(conf.get(ConfigOptions.SERVER_SSL_CIPHER_SUITES));
+ validateProtocolsAndCipherSuites(
+ enabledProtocols,
+ ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS,
+ cipherSuites,
+ ConfigOptions.SERVER_SSL_CIPHER_SUITES);
+
+ String keystoreType = conf.getString(ConfigOptions.SERVER_SSL_KEYSTORE_TYPE);
+ String truststoreType = conf.getString(ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE);
+ validateStoreType(keystorePath, keystoreType, ConfigOptions.SERVER_SSL_KEYSTORE_TYPE);
+ validateStoreType(truststorePath, truststoreType, ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE);
+
+ return Optional.of(
+ new SslConfig(
+ enabledListeners,
+ clientAuthListeners,
+ enabledProtocols,
+ cipherSuites,
+ keystorePath,
+ password(conf.get(ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD)),
+ keystoreType,
+ password(conf.get(ConfigOptions.SERVER_SSL_KEY_PASSWORD)),
+ truststorePath,
+ password(conf.get(ConfigOptions.SERVER_SSL_TRUSTSTORE_PASSWORD)),
+ truststoreType,
+ ""));
+ }
+
+ /**
+ * Build and validate the client-side TLS configuration, or {@link Optional#empty()} when TLS is
+ * disabled (i.e. {@code client.security.ssl.enabled} is false).
+ */
+ public static Optional<SslConfig> fromClientConfig(Configuration conf) {
+ // Mirrors the server side: a client that authenticates with a certificate needs TLS to
+ // present it over, and a keystore to present it from. Both are checked before TLS is known
+ // to be enabled, so the combination that disables TLS outright is rejected too.
+ boolean mutualTls =
+ MUTUAL_TLS_AUTH_PROTOCOL.equalsIgnoreCase(
+ conf.getString(ConfigOptions.CLIENT_SECURITY_PROTOCOL));
+ if (mutualTls && !conf.get(ConfigOptions.CLIENT_SSL_ENABLED)) {
+ throw new IllegalConfigurationException(
+ "'%s' is '%s', which requires TLS transport, but '%s' is false.",
+ ConfigOptions.CLIENT_SECURITY_PROTOCOL.key(),
+ MUTUAL_TLS_AUTH_PROTOCOL,
+ ConfigOptions.CLIENT_SSL_ENABLED.key());
+ }
+ if (mutualTls && conf.getString(ConfigOptions.CLIENT_SSL_KEYSTORE_PATH) == null) {
+ throw new IllegalConfigurationException(
+ "'%s' is '%s', but '%s' is not configured, so the client has no certificate "
+ + "to present to the server.",
+ ConfigOptions.CLIENT_SECURITY_PROTOCOL.key(),
+ MUTUAL_TLS_AUTH_PROTOCOL,
+ ConfigOptions.CLIENT_SSL_KEYSTORE_PATH.key());
+ }
+
+ if (!conf.get(ConfigOptions.CLIENT_SSL_ENABLED)) {
+ return Optional.empty();
+ }
+
+ List<String> enabledProtocols =
+ effectiveProtocols(conf, ConfigOptions.CLIENT_SSL_ENABLED_PROTOCOLS);
+ List<String> cipherSuites = orEmpty(conf.get(ConfigOptions.CLIENT_SSL_CIPHER_SUITES));
+ validateProtocolsAndCipherSuites(
+ enabledProtocols,
+ ConfigOptions.CLIENT_SSL_ENABLED_PROTOCOLS,
+ cipherSuites,
+ ConfigOptions.CLIENT_SSL_CIPHER_SUITES);
+
+ String endpointIdentificationAlgorithm =
+ conf.getString(ConfigOptions.CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM);
+ validateEndpointIdentificationAlgorithm(endpointIdentificationAlgorithm);
+
+ String keystorePath = conf.getString(ConfigOptions.CLIENT_SSL_KEYSTORE_PATH);
+ String keystoreType = conf.getString(ConfigOptions.CLIENT_SSL_KEYSTORE_TYPE);
+ String truststorePath = conf.getString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_PATH);
+ String truststoreType = conf.getString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_TYPE);
+ validateStoreType(keystorePath, keystoreType, ConfigOptions.CLIENT_SSL_KEYSTORE_TYPE);
+ validateStoreType(truststorePath, truststoreType, ConfigOptions.CLIENT_SSL_TRUSTSTORE_TYPE);
+
+ return Optional.of(
+ new SslConfig(
+ Collections.emptyList(),
+ Collections.emptySet(),
+ enabledProtocols,
+ cipherSuites,
+ keystorePath,
+ conf.getString(ConfigOptions.CLIENT_SSL_KEYSTORE_PASSWORD),
+ keystoreType,
+ conf.getString(ConfigOptions.CLIENT_SSL_KEY_PASSWORD),
+ truststorePath,
+ conf.getString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_PASSWORD),
+ truststoreType,
+ endpointIdentificationAlgorithm));
+ }
+
+ /**
+ * The TLS protocols to enable. An explicitly configured list is taken as given, and validated
+ * against this JVM. The default list is narrowed to what this JVM supports instead, so that a
+ * JDK without TLS 1.3 - 8u252 and older - runs on TLS 1.2 rather than failing to start on a
+ * default the operator never chose.
+ */
+ private static List<String> effectiveProtocols(
+ Configuration conf, ConfigOption<List<String>> option) {
+ Optional<List<String>> configured = conf.getOptional(option);
+ if (configured.isPresent()) {
+ return orEmpty(configured.get());
+ }
+ Set<String> supported =
+ new LinkedHashSet<>(
+ Arrays.asList(supportedAlgorithmsProbe().getSupportedProtocols()));
+ return narrowToSupported(orEmpty(option.defaultValue()), supported, option);
+ }
+
+ @VisibleForTesting
+ static List<String> narrowToSupported(
+ List<String> defaults, Set<String> supported, ConfigOption<List<String>> option) {
+ List<String> usable =
+ defaults.stream().filter(supported::contains).collect(Collectors.toList());
+ if (usable.isEmpty()) {
+ throw new IllegalConfigurationException(
+ "This JVM supports none of the default TLS protocols %s; it supports %s. "
+ + "Set '%s' explicitly.",
+ defaults, supported, option.key());
+ }
+ return usable;
+ }
+
+ /**
+ * Reject protocol and cipher suite names this JVM does not support, and an empty protocol list.
+ * Both are otherwise only caught when an {@link SSLEngine} is created, i.e. once per connection
+ * and without naming the option at fault — and an empty protocol list is not caught at all: the
+ * engine then comes up with no protocol enabled and every handshake fails.
+ *
+ * <p>Cipher suites are only checked against what the JVM supports, not against the enabled
+ * protocols: which suites can actually be negotiated depends on the protocol version agreed
+ * during the handshake, so pinning e.g. a TLS 1.3 suite alongside TLS 1.2 is a handshake
+ * concern, not a configuration error. An empty cipher suite list is not an error either — it
+ * selects the provider defaults.
+ */
+ private static void validateProtocolsAndCipherSuites(
+ List<String> enabledProtocols,
+ ConfigOption<List<String>> protocolsOption,
+ List<String> cipherSuites,
+ ConfigOption<List<String>> cipherSuitesOption) {
+ if (enabledProtocols.isEmpty()) {
+ throw new IllegalConfigurationException(
+ "'%s' must list at least one TLS protocol.", protocolsOption.key());
+ }
+
+ SSLEngine probe = supportedAlgorithmsProbe();
+
+ List<String> unsupportedProtocols =
+ unsupported(enabledProtocols, probe.getSupportedProtocols());
+ if (!unsupportedProtocols.isEmpty()) {
+ throw new IllegalConfigurationException(
+ "'%s' contains TLS protocol(s) not supported by this JVM: %s. Supported: %s.",
+ protocolsOption.key(),
+ unsupportedProtocols,
+ Arrays.asList(probe.getSupportedProtocols()));
+ }
+
+ List<String> unsupportedCipherSuites =
+ unsupported(cipherSuites, probe.getSupportedCipherSuites());
+ if (!unsupportedCipherSuites.isEmpty()) {
+ throw new IllegalConfigurationException(
+ "'%s' contains cipher suite(s) not supported by this JVM: %s. Supported: %s.",
+ cipherSuitesOption.key(),
+ unsupportedCipherSuites,
+ Arrays.asList(probe.getSupportedCipherSuites()));
+ }
+ }
+
+ /**
+ * Reject an endpoint identification algorithm the JDK does not implement. An unknown name is
+ * accepted by {@code SSLParameters} and only rejected once a handshake runs, as "Unknown
+ * identification algorithm", i.e. on every connection. An empty value is valid and disables
+ * hostname verification.
+ */
+ private static void validateEndpointIdentificationAlgorithm(@Nullable String algorithm) {
+ if (algorithm == null || algorithm.trim().isEmpty()) {
+ return;
+ }
+ boolean known =
+ ENDPOINT_IDENTIFICATION_ALGORITHMS.stream()
+ .anyMatch(supported -> supported.equalsIgnoreCase(algorithm));
+ if (!known) {
+ throw new IllegalConfigurationException(
+ "'%s' is set to '%s', which the JDK does not implement, so every handshake "
+ + "would fail. Supported: %s, or empty to disable hostname "
+ + "verification.",
+ ConfigOptions.CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM.key(),
+ algorithm,
+ ENDPOINT_IDENTIFICATION_ALGORITHMS);
+ }
+ }
+
+ /**
+ * Reject a keystore or truststore type this JVM has no provider for. Only the type of a store
+ * that is actually configured is checked, since the type of an absent store is never used.
+ * Without this the typo surfaces as a {@code KeyStoreException} while the SSL context is built,
+ * which does not name the option at fault.
+ */
+ private static void validateStoreType(
+ @Nullable String storePath, String storeType, ConfigOption<String> storeTypeOption) {
+ if (storePath == null) {
+ return;
+ }
+ try {
+ KeyStore.getInstance(storeType);
+ } catch (KeyStoreException e) {
+ throw new IllegalConfigurationException(
+ "'%s' is set to '%s', which is not a keystore type supported by this JVM. Supported: %s.",
+ storeTypeOption.key(),
+ storeType,
+ new TreeSet<>(Security.getAlgorithms("KeyStore")));
+ }
+ }
+
+ private static List<String> unsupported(List<String> configured, String[] supported) {
+ Set<String> supportedSet = new HashSet<>(Arrays.asList(supported));
+ return configured.stream()
+ .filter(value -> !supportedSet.contains(value))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * An engine off the default JSSE context, used only for the protocol and cipher suite names it
+ * reports as supported. Those come from the security provider, so they do not depend on the
+ * configured key material.
+ */
+ private static SSLEngine supportedAlgorithmsProbe() {
+ try {
+ SSLContext context = SSLContext.getInstance("TLS");
+ context.init(null, null, null);
+ return context.createSSLEngine();
+ } catch (GeneralSecurityException e) {
+ throw new FlussRuntimeException(
+ "Failed to determine the TLS protocols and cipher suites supported by this JVM.",
+ e);
+ }
+ }
+
+ @Nullable
+ private static String password(@Nullable Password password) {
+ return password == null ? null : password.value();
+ }
+
+ private static List<String> orEmpty(@Nullable List<String> list) {
+ return list == null ? Collections.emptyList() : list;
+ }
+
+ /**
+ * Server-only: the listener names for which TLS is enabled, as configured via {@code
+ * security.ssl.enabled.listeners}. Never empty for a server-side config (an empty list means
+ * TLS is off and {@link #fromServerConfig} returns no config at all); always empty for a
+ * client-side config.
+ */
+ public List<String> enabledListeners() {
+ return enabledListeners;
+ }
+
+ /**
+ * Server-only: the TLS listeners that require a client certificate, i.e. those whose {@code
+ * security.protocol.map} entry is {@code mTLS}. Always empty for a client-side config.
+ */
+ public Set<String> clientAuthListeners() {
+ return clientAuthListeners;
+ }
+
+ /**
+ * Whether {@code listenerName} requires a client certificate during the TLS handshake. A
+ * truststore is guaranteed to be configured when this returns true.
+ */
+ public boolean requiresClientAuth(String listenerName) {
+ return clientAuthListeners.contains(listenerName);
+ }
+
+ public String[] enabledProtocols() {
+ return enabledProtocols.toArray(new String[0]);
+ }
+
+ public List<String> cipherSuites() {
+ return cipherSuites;
+ }
+
+ @Nullable
+ public String keystorePath() {
+ return keystorePath;
+ }
+
+ @Nullable
+ public String keystorePassword() {
+ return keystorePassword;
+ }
+
+ public String keystoreType() {
+ return keystoreType;
+ }
+
+ /** The key password, falling back to the keystore password when not explicitly configured. */
+ @Nullable
+ public String keyPassword() {
+ return keyPassword != null ? keyPassword : keystorePassword;
+ }
+
+ @Nullable
+ public String truststorePath() {
+ return truststorePath;
+ }
+
+ @Nullable
+ public String truststorePassword() {
+ return truststorePassword;
+ }
+
+ public String truststoreType() {
+ return truststoreType;
+ }
+
+ public String endpointIdentificationAlgorithm() {
+ return endpointIdentificationAlgorithm;
+ }
+}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslContextFactory.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslContextFactory.java
new file mode 100644
index 0000000000..818855a417
--- /dev/null
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslContextFactory.java
@@ -0,0 +1,379 @@
+/*
+ * 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.fluss.rpc.netty.ssl;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBufAllocator;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.SslContext;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.SslContextBuilder;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.SslHandler;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.SslProvider;
+
+import javax.annotation.Nullable;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Paths;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.UnrecoverableKeyException;
+import java.util.Enumeration;
+import java.util.Optional;
+
+/**
+ * Builds Netty {@link SslContext}s and {@link SslHandler}s for the Fluss RPC layer from the {@code
+ * security.ssl.*} (server) and {@code client.security.ssl.*} (client) configuration.
+ *
+ * <p>The JDK SSL provider is used for portability (it does not require a native OpenSSL binding to
+ * be present on the host). Transport encryption is orthogonal to the application-level
+ * authentication protocol: a TLS-enabled listener can run any authenticator. Mutual TLS is realized
+ * by a listener whose auth protocol is {@code mTLS}; that per-listener client-certificate
+ * requirement is parsed and validated by {@link SslConfig} and applied on the server handler
+ * ({@link #createServerSslHandler}).
+ */
+@Internal
+public final class SslContextFactory {
+
+ private SslContextFactory() {}
+
+ /**
+ * Build a server {@link SslContext} from the {@code security.ssl.*} configuration, or {@link
+ * Optional#empty()} when TLS is not enabled for any listener.
+ */
+ public static Optional<SslContext> createServerSslContext(Configuration conf) {
+ return SslConfig.fromServerConfig(conf).map(SslContextFactory::createServerSslContext);
+ }
+
+ /** Build a server {@link SslContext} from a parsed {@link SslConfig}. */
+ public static SslContext createServerSslContext(SslConfig config) {
+ KeyManagerFactory kmf =
+ keyManagerFactory(
+ SERVER_KEYSTORE,
+ config.keystorePath(),
+ config.keystoreType(),
+ config.keystorePassword(),
+ config.keyPassword());
+ SslContextBuilder builder =
+ SslContextBuilder.forServer(kmf)
+ .sslProvider(SslProvider.JDK)
+ .protocols(config.enabledProtocols());
+ if (!config.cipherSuites().isEmpty()) {
+ builder.ciphers(config.cipherSuites());
+ }
+ if (config.truststorePath() != null) {
+ builder.trustManager(
+ trustManagerFactory(
+ SERVER_TRUSTSTORE,
+ config.truststorePath(),
+ config.truststoreType(),
+ config.truststorePassword()));
+ }
+ try {
+ return builder.build();
+ } catch (Exception e) {
+ throw new FlussRuntimeException("Failed to build the server SSL context.", e);
+ }
+ }
+
+ /**
+ * Build a client {@link SslContext} from the {@code client.security.ssl.*} configuration, or
+ * {@link Optional#empty()} when TLS is disabled on the client.
+ */
+ public static Optional<SslContext> createClientSslContext(Configuration conf) {
+ return SslConfig.fromClientConfig(conf).map(SslContextFactory::createClientSslContext);
+ }
+
+ /** Build a client {@link SslContext} from a parsed {@link SslConfig}. */
+ public static SslContext createClientSslContext(SslConfig config) {
+ SslContextBuilder builder =
+ SslContextBuilder.forClient()
+ .sslProvider(SslProvider.JDK)
+ .protocols(config.enabledProtocols());
+ if (!config.cipherSuites().isEmpty()) {
+ builder.ciphers(config.cipherSuites());
+ }
+ if (config.truststorePath() != null) {
+ builder.trustManager(
+ trustManagerFactory(
+ CLIENT_TRUSTSTORE,
+ config.truststorePath(),
+ config.truststoreType(),
+ config.truststorePassword()));
+ }
+ // Present a client certificate when a keystore is configured (required for mutual TLS).
+ if (config.keystorePath() != null) {
+ builder.keyManager(
+ keyManagerFactory(
+ CLIENT_KEYSTORE,
+ config.keystorePath(),
+ config.keystoreType(),
+ config.keystorePassword(),
+ config.keyPassword()));
+ }
+ try {
+ return builder.build();
+ } catch (Exception e) {
+ throw new FlussRuntimeException("Failed to build the client SSL context.", e);
+ }
+ }
+
+ /**
+ * Create a server-side {@link SslHandler} for a newly accepted channel. When {@code
+ * requireClientAuth} is true the engine demands a client certificate during the handshake;
+ * callers derive that per listener from {@link SslConfig#requiresClientAuth(String)}, which
+ * only returns true when a truststore is configured to validate those certificates against.
+ */
+ public static SslHandler createServerSslHandler(
+ SslContext sslContext, ByteBufAllocator alloc, boolean requireClientAuth) {
+ SslHandler handler = sslContext.newHandler(alloc);
+ if (requireClientAuth) {
+ handler.engine().setNeedClientAuth(true);
+ }
+ return handler;
+ }
+
+ /**
+ * Create a client-side {@link SslHandler} for a connection to {@code host:port}, configuring
+ * SNI and (optionally) hostname verification via the endpoint identification algorithm.
+ */
+ public static SslHandler createClientSslHandler(
+ SslContext sslContext,
+ ByteBufAllocator alloc,
+ String host,
+ int port,
+ String endpointIdentificationAlgorithm) {
+ SslHandler handler = sslContext.newHandler(alloc, host, port);
+ SSLEngine engine = handler.engine();
+ SSLParameters parameters = engine.getSSLParameters();
+ // An empty algorithm disables hostname verification (Kafka semantics).
+ parameters.setEndpointIdentificationAlgorithm(
+ endpointIdentificationAlgorithm == null
+ || endpointIdentificationAlgorithm.trim().isEmpty()
+ ? null
+ : endpointIdentificationAlgorithm);
+ engine.setSSLParameters(parameters);
+ return handler;
+ }
+
+ private static KeyManagerFactory keyManagerFactory(
+ Store store,
+ String path,
+ String type,
+ @Nullable String storePassword,
+ @Nullable String keyPassword) {
+ KeyStore keyStore = loadKeyStore(store, path, type, storePassword);
+ checkHoldsPrivateKey(store, keyStore, path);
+ try {
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keyPassword == null ? null : keyPassword.toCharArray());
+ return kmf;
+ } catch (UnrecoverableKeyException e) {
+ throw new FlussRuntimeException(
+ String.format(
+ "Cannot recover the private key from the %s at '%s'. Check '%s', which "
+ + "defaults to '%s' when not set.",
+ store.what, path, store.keyPasswordKey, store.passwordKey),
+ e);
+ } catch (GeneralSecurityException e) {
+ throw new FlussRuntimeException(
+ String.format(
+ "Failed to build a key manager from the %s at '%s'.", store.what, path),
+ e);
+ }
+ }
+
+ private static TrustManagerFactory trustManagerFactory(
+ Store store, String path, String type, @Nullable String storePassword) {
+ KeyStore trustStore = loadKeyStore(store, path, type, storePassword);
+ checkHoldsCertificate(store, trustStore, path);
+ try {
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ return tmf;
+ } catch (GeneralSecurityException e) {
+ throw new FlussRuntimeException(
+ String.format(
+ "Failed to build a trust manager from the %s at '%s'.",
+ store.what, path),
+ e);
+ }
+ }
+
+ /**
+ * Reject a keystore that carries no private key, which a truststore pointed at by mistake does.
+ * Such a store loads and builds a context without complaint, and then fails every handshake
+ * with "No available authentication scheme" - on the server side only, so the client sees a
+ * dropped connection with no cause at all.
+ */
+ private static void checkHoldsPrivateKey(Store store, KeyStore keyStore, String path) {
+ try {
+ Enumeration<String> aliases = keyStore.aliases();
+ while (aliases.hasMoreElements()) {
+ if (keyStore.isKeyEntry(aliases.nextElement())) {
+ return;
+ }
+ }
+ } catch (KeyStoreException e) {
+ throw new FlussRuntimeException(
+ String.format("Failed to read the %s at '%s'.", store.what, path), e);
+ }
+ throw new FlussRuntimeException(
+ String.format(
+ "The %s at '%s', configured by '%s', holds no private key, only trusted "
+ + "certificates. A truststore is the usual thing to find there by "
+ + "mistake. Every TLS handshake would fail with \"No available "
+ + "authentication scheme\".",
+ store.what, path, store.pathKey));
+ }
+
+ /**
+ * Reject a truststore the trust manager would find empty. A PKCS12 store keeps its contents
+ * under the store password and loads with no entries at all when that password is missing,
+ * without reporting anything; JKS reads its certificates either way, so the mistake only shows
+ * on PKCS12. The handshake failure it produces - "the trustAnchors parameter must be non-empty"
+ * - names neither the file nor the option.
+ */
+ private static void checkHoldsCertificate(Store store, KeyStore trustStore, String path) {
+ try {
+ Enumeration<String> aliases = trustStore.aliases();
+ while (aliases.hasMoreElements()) {
+ if (trustStore.isCertificateEntry(aliases.nextElement())) {
+ return;
+ }
+ }
+ } catch (KeyStoreException e) {
+ throw new FlussRuntimeException(
+ String.format("Failed to read the %s at '%s'.", store.what, path), e);
+ }
+ throw new FlussRuntimeException(
+ String.format(
+ "The %s at '%s', configured by '%s', holds no trusted certificates. A "
+ + "PKCS12 store loads empty when '%s' is not set. Every TLS "
+ + "handshake would fail with \"the trustAnchors parameter must be "
+ + "non-empty\".",
+ store.what, path, store.pathKey, store.passwordKey));
+ }
+
+ /**
+ * Load a keystore or truststore, reporting the failures an operator can actually cause - a path
+ * that is not there, and a store password that does not open the file - with the file and the
+ * option that configures it, rather than letting the bare JCA exception surface.
+ */
+ private static KeyStore loadKeyStore(
+ Store store, String path, String type, @Nullable String password) {
+ KeyStore keyStore;
+ try {
+ keyStore = KeyStore.getInstance(type);
+ } catch (GeneralSecurityException e) {
+ // unreachable for a config parsed by SslConfig, which validates the type up front.
+ throw new FlussRuntimeException(
+ String.format(
+ "'%s' is set to the unsupported store type '%s'.", store.typeKey, type),
+ e);
+ }
+ try (InputStream in = Files.newInputStream(Paths.get(path))) {
+ keyStore.load(in, password == null ? null : password.toCharArray());
+ return keyStore;
+ } catch (NoSuchFileException e) {
+ throw new FlussRuntimeException(
+ String.format(
+ "No %s at '%s', configured by '%s'.", store.what, path, store.pathKey),
+ e);
+ } catch (IOException e) {
+ throw new FlussRuntimeException(
+ String.format(
+ "Failed to read the %s at '%s' as a '%s' store. A wrong '%s' is the "
+ + "usual cause; the file may also be unreadable or not a "
+ + "keystore at all.",
+ store.what, path, type, store.passwordKey),
+ e);
+ } catch (GeneralSecurityException e) {
+ throw new FlussRuntimeException(
+ String.format(
+ "Failed to load the %s at '%s' as a '%s' store.",
+ store.what, path, type),
+ e);
+ }
+ }
+
+ /** The options that configure one keystore or truststore, so failures can name them. */
+ private static final class Store {
+ private final String what;
+ private final String pathKey;
+ private final String passwordKey;
+ private final String typeKey;
+ @Nullable private final String keyPasswordKey;
+
+ private Store(
+ String what,
+ String pathKey,
+ String passwordKey,
+ String typeKey,
+ @Nullable String keyPasswordKey) {
+ this.what = what;
+ this.pathKey = pathKey;
+ this.passwordKey = passwordKey;
+ this.typeKey = typeKey;
+ this.keyPasswordKey = keyPasswordKey;
+ }
+ }
+
+ private static final Store SERVER_KEYSTORE =
+ new Store(
+ "server keystore",
+ ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key(),
+ ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD.key(),
+ ConfigOptions.SERVER_SSL_KEYSTORE_TYPE.key(),
+ ConfigOptions.SERVER_SSL_KEY_PASSWORD.key());
+
+ private static final Store SERVER_TRUSTSTORE =
+ new Store(
+ "server truststore",
+ ConfigOptions.SERVER_SSL_TRUSTSTORE_PATH.key(),
+ ConfigOptions.SERVER_SSL_TRUSTSTORE_PASSWORD.key(),
+ ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE.key(),
+ null);
+
+ private static final Store CLIENT_KEYSTORE =
+ new Store(
+ "client keystore",
+ ConfigOptions.CLIENT_SSL_KEYSTORE_PATH.key(),
+ ConfigOptions.CLIENT_SSL_KEYSTORE_PASSWORD.key(),
+ ConfigOptions.CLIENT_SSL_KEYSTORE_TYPE.key(),
+ ConfigOptions.CLIENT_SSL_KEY_PASSWORD.key());
+
+ private static final Store CLIENT_TRUSTSTORE =
+ new Store(
+ "client truststore",
+ ConfigOptions.CLIENT_SSL_TRUSTSTORE_PATH.key(),
+ ConfigOptions.CLIENT_SSL_TRUSTSTORE_PASSWORD.key(),
+ ConfigOptions.CLIENT_SSL_TRUSTSTORE_TYPE.key(),
+ null);
+}
diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/SslContextFactoryTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/SslContextFactoryTest.java
new file mode 100644
index 0000000000..2da09e319e
--- /dev/null
+++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/SslContextFactoryTest.java
@@ -0,0 +1,800 @@
+/*
+ * 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.fluss.rpc.netty.ssl;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.IllegalConfigurationException;
+import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBufAllocator;
+import org.apache.fluss.shaded.netty4.io.netty.channel.Channel;
+import org.apache.fluss.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.SslContext;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.SslHandler;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.util.SelfSignedCertificate;
+import org.apache.fluss.shaded.netty4.io.netty.util.concurrent.Future;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link SslContextFactory} and {@link SslConfig}. */
+class SslContextFactoryTest {
+
+ @TempDir private Path tempDir;
+
+ private Path keyStore;
+ private Path trustStore;
+
+ @BeforeEach
+ void setup() throws Exception {
+ SelfSignedCertificate cert = TestSslUtils.generateCertificate("localhost");
+ keyStore = TestSslUtils.createKeyStore(tempDir, "keystore.jks", cert);
+ trustStore = TestSslUtils.createTrustStore(tempDir, "truststore.jks", cert);
+ }
+
+ @Test
+ void testCreateServerSslContext() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, trustStore);
+
+ SslContext sslContext = SslContextFactory.createServerSslContext(conf).get();
+ assertThat(sslContext.isServer()).isTrue();
+ assertThat(sslContext.newEngine(ByteBufAllocator.DEFAULT).getEnabledProtocols())
+ .contains("TLSv1.2", "TLSv1.3");
+ }
+
+ @Test
+ void testCreateClientSslContext() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, keyStore);
+
+ SslContext sslContext = SslContextFactory.createClientSslContext(conf).get();
+ assertThat(sslContext.isClient()).isTrue();
+ }
+
+ @Test
+ void testProtocolFiltering() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS.key(), "TLSv1.2");
+
+ SslContext sslContext = SslContextFactory.createServerSslContext(conf).get();
+ assertThat(sslContext.newEngine(ByteBufAllocator.DEFAULT).getEnabledProtocols())
+ .containsExactly("TLSv1.2");
+ }
+
+ @Test
+ void testCipherSuiteFiltering() {
+ String pinned = "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256";
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS.key(), "TLSv1.2");
+ conf.setString(ConfigOptions.SERVER_SSL_CIPHER_SUITES.key(), pinned);
+
+ SslContext sslContext = SslContextFactory.createServerSslContext(conf).get();
+ assertThat(sslContext.newEngine(ByteBufAllocator.DEFAULT).getEnabledCipherSuites())
+ .contains(pinned)
+ .doesNotContain("TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384");
+ }
+
+ @Test
+ void testKeyPasswordFallsBackToKeystorePassword() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+
+ SslConfig config = SslConfig.fromServerConfig(conf).get();
+ assertThat(config.keyPassword()).isEqualTo(TestSslUtils.PASSWORD);
+
+ conf.setString(ConfigOptions.SERVER_SSL_KEY_PASSWORD.key(), "key-only-password");
+ config = SslConfig.fromServerConfig(conf).get();
+ assertThat(config.keyPassword()).isEqualTo("key-only-password");
+ }
+
+ @Test
+ void testClientSslHandlerEndpointIdentification() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, null);
+ SslContext sslContext = SslContextFactory.createClientSslContext(conf).get();
+
+ SslHandler httpsHandler =
+ SslContextFactory.createClientSslHandler(
+ sslContext, ByteBufAllocator.DEFAULT, "localhost", 9123, "https");
+ assertThat(httpsHandler.engine().getSSLParameters().getEndpointIdentificationAlgorithm())
+ .isEqualTo("https");
+
+ SslHandler noVerifyHandler =
+ SslContextFactory.createClientSslHandler(
+ sslContext, ByteBufAllocator.DEFAULT, "localhost", 9123, "");
+ assertThat(noVerifyHandler.engine().getSSLParameters().getEndpointIdentificationAlgorithm())
+ .isNull();
+ }
+
+ @Test
+ void testServerSslHandlerClientAuthRequirement() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, trustStore);
+ SslContext sslContext = SslContextFactory.createServerSslContext(conf).get();
+
+ SslHandler noClientAuth =
+ SslContextFactory.createServerSslHandler(
+ sslContext, ByteBufAllocator.DEFAULT, false);
+ assertThat(noClientAuth.engine().getNeedClientAuth()).isFalse();
+
+ SslHandler requireClientAuth =
+ SslContextFactory.createServerSslHandler(
+ sslContext, ByteBufAllocator.DEFAULT, true);
+ assertThat(requireClientAuth.engine().getNeedClientAuth()).isTrue();
+ }
+
+ @Test
+ void testServerAndClientNegotiateTls() throws Exception {
+ Configuration serverConf = new Configuration();
+ TestSslUtils.setServerSslConfig(serverConf, keyStore, null);
+ Configuration clientConf = new Configuration();
+ TestSslUtils.setClientSslConfig(clientConf, trustStore, null);
+
+ HandshakeResult result = handshake(serverConf, clientConf, false);
+
+ assertThat(result.clientHandshake.isSuccess()).isTrue();
+ assertThat(result.serverHandshake.isSuccess()).isTrue();
+ // Proves real encryption was negotiated (not a NULL/plaintext cipher).
+ assertThat(result.clientHandler.engine().getSession().getProtocol()).startsWith("TLS");
+ assertThat(result.clientHandler.engine().getSession().getCipherSuite())
+ .doesNotContain("NULL");
+ }
+
+ @Test
+ void testHostnameVerificationRejectsCertificateForAnotherHost() throws Exception {
+ SelfSignedCertificate otherHost = TestSslUtils.generateCertificate("other.example.com");
+ Path otherKeyStore = TestSslUtils.createKeyStore(tempDir, "other-keystore.jks", otherHost);
+ Path otherTrustStore =
+ TestSslUtils.createTrustStore(tempDir, "other-truststore.jks", otherHost);
+
+ Configuration serverConf = new Configuration();
+ TestSslUtils.setServerSslConfig(serverConf, otherKeyStore, null);
+ Configuration clientConf = new Configuration();
+ // the certificate is trusted; only the name it was issued for does not match the host.
+ TestSslUtils.setClientSslConfig(clientConf, otherTrustStore, null);
+
+ HandshakeResult result = handshake(serverConf, clientConf, false, "localhost", "https");
+
+ assertThat(result.clientHandshake.isSuccess()).isFalse();
+ assertThat(result.clientHandshake.cause()).hasMessageContaining("localhost");
+ }
+
+ @Test
+ void testHostnameVerificationAcceptsMatchingCertificate() throws Exception {
+ Configuration serverConf = new Configuration();
+ TestSslUtils.setServerSslConfig(serverConf, keyStore, null);
+ Configuration clientConf = new Configuration();
+ TestSslUtils.setClientSslConfig(clientConf, trustStore, null);
+
+ // same setup as the rejection above, with a certificate issued for the host dialled.
+ HandshakeResult result = handshake(serverConf, clientConf, false, "localhost", "https");
+
+ assertThat(result.clientHandshake.isSuccess()).isTrue();
+ }
+
+ @Test
+ void testClientRejectsServerCertificateOutsideItsTruststore() throws Exception {
+ SelfSignedCertificate untrusted = TestSslUtils.generateCertificate("localhost");
+ Path untrustedTrustStore =
+ TestSslUtils.createTrustStore(tempDir, "untrusted-truststore.jks", untrusted);
+
+ Configuration serverConf = new Configuration();
+ TestSslUtils.setServerSslConfig(serverConf, keyStore, null);
+ Configuration clientConf = new Configuration();
+ // trusts a different self-signed certificate than the one the server presents.
+ TestSslUtils.setClientSslConfig(clientConf, untrustedTrustStore, null);
+
+ HandshakeResult result = handshake(serverConf, clientConf, false);
+
+ assertThat(result.clientHandshake.isSuccess()).isFalse();
+ // both certificates are issued for localhost, so the JDK finds a trust anchor by name and
+ // then rejects it on the signature: trust is decided by key, not by subject.
+ assertThat(result.clientHandshake.cause())
+ .hasMessageContaining("PKIX path validation failed");
+ }
+
+ @Test
+ void testClientAuthRejectsClientWithoutCertificate() throws Exception {
+ Configuration serverConf = new Configuration();
+ TestSslUtils.setServerSslConfig(serverConf, keyStore, trustStore);
+ TestSslUtils.setMutualTlsProtocolMap(serverConf);
+ Configuration clientConf = new Configuration();
+ // trusts the server, but presents no certificate of its own.
+ TestSslUtils.setClientSslConfig(clientConf, trustStore, null);
+
+ HandshakeResult result = handshake(serverConf, clientConf, true);
+
+ // The server is the side that enforces client auth. Under TLS 1.3 the client sends its
+ // Finished before the server validates the (missing) certificate, so the client's
+ // handshake future completes successfully and only learns of the rejection from the
+ // alert that follows.
+ assertThat(result.serverHandshake.isSuccess()).isFalse();
+ assertThat(result.serverHandshake.cause())
+ .hasMessageContaining("Empty client certificate chain");
+ }
+
+ @Test
+ void testClientAuthAcceptsClientWithCertificate() throws Exception {
+ Configuration serverConf = new Configuration();
+ TestSslUtils.setServerSslConfig(serverConf, keyStore, trustStore);
+ TestSslUtils.setMutualTlsProtocolMap(serverConf);
+ Configuration clientConf = new Configuration();
+ TestSslUtils.setClientSslConfig(clientConf, trustStore, keyStore);
+
+ HandshakeResult result = handshake(serverConf, clientConf, true);
+
+ assertThat(result.clientHandshake.isSuccess()).isTrue();
+ assertThat(result.serverHandshake.isSuccess()).isTrue();
+ }
+
+ @Test
+ void testPkcs12KeyStoreAndTrustStore() throws Exception {
+ SelfSignedCertificate cert = TestSslUtils.generateCertificate("localhost");
+ Path p12KeyStore = TestSslUtils.createKeyStore(tempDir, "keystore.p12", "PKCS12", cert);
+ Path p12TrustStore =
+ TestSslUtils.createTrustStore(tempDir, "truststore.p12", "PKCS12", cert);
+
+ Configuration serverConf = new Configuration();
+ TestSslUtils.setServerSslConfig(serverConf, p12KeyStore, p12TrustStore);
+ serverConf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_TYPE.key(), "PKCS12");
+ serverConf.setString(ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE.key(), "PKCS12");
+ Configuration clientConf = new Configuration();
+ TestSslUtils.setClientSslConfig(clientConf, p12TrustStore, p12KeyStore);
+ clientConf.setString(ConfigOptions.CLIENT_SSL_KEYSTORE_TYPE.key(), "PKCS12");
+ clientConf.setString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_TYPE.key(), "PKCS12");
+
+ HandshakeResult result = handshake(serverConf, clientConf, true);
+
+ assertThat(result.clientHandshake.isSuccess()).isTrue();
+ assertThat(result.serverHandshake.isSuccess()).isTrue();
+ }
+
+ /** The outcome of a full embedded-channel handshake between a server and a client handler. */
+ private static class HandshakeResult {
+ private final Future<Channel> serverHandshake;
+ private final Future<Channel> clientHandshake;
+ private final SslHandler clientHandler;
+
+ private HandshakeResult(
+ Future<Channel> serverHandshake,
+ Future<Channel> clientHandshake,
+ SslHandler clientHandler) {
+ this.serverHandshake = serverHandshake;
+ this.clientHandshake = clientHandshake;
+ this.clientHandler = clientHandler;
+ }
+ }
+
+ /**
+ * Pump a TLS handshake between a server and a client handler built from the given
+ * configurations, and return both handshake futures once they settle.
+ */
+ private static HandshakeResult handshake(
+ Configuration serverConf, Configuration clientConf, boolean requireClientAuth) {
+ // endpoint identification disabled: these tests are about the certificate exchange.
+ return handshake(serverConf, clientConf, requireClientAuth, "localhost", "");
+ }
+
+ /**
+ * Pump a TLS handshake, with the client dialling {@code host} and applying {@code
+ * endpointIdentificationAlgorithm} to the server certificate it receives.
+ */
+ private static HandshakeResult handshake(
+ Configuration serverConf,
+ Configuration clientConf,
+ boolean requireClientAuth,
+ String host,
+ String endpointIdentificationAlgorithm) {
+ SslHandler serverHandler =
+ SslContextFactory.createServerSslHandler(
+ SslContextFactory.createServerSslContext(serverConf).get(),
+ ByteBufAllocator.DEFAULT,
+ requireClientAuth);
+ SslHandler clientHandler =
+ SslContextFactory.createClientSslHandler(
+ SslContextFactory.createClientSslContext(clientConf).get(),
+ ByteBufAllocator.DEFAULT,
+ host,
+ 9123,
+ endpointIdentificationAlgorithm);
+
+ EmbeddedChannel serverChannel = new EmbeddedChannel(serverHandler);
+ EmbeddedChannel clientChannel = new EmbeddedChannel(clientHandler);
+ try {
+ for (int i = 0;
+ i < 20
+ && !(clientHandler.handshakeFuture().isDone()
+ && serverHandler.handshakeFuture().isDone());
+ i++) {
+ transferOutbound(clientChannel, serverChannel);
+ transferOutbound(serverChannel, clientChannel);
+ }
+ return new HandshakeResult(
+ serverHandler.handshakeFuture(),
+ clientHandler.handshakeFuture(),
+ clientHandler);
+ } catch (Throwable rejected) {
+ // A rejected handshake also propagates through the channel that decoded the alert.
+ // The handshake futures already carry the outcome, which is what callers assert on.
+ return new HandshakeResult(
+ serverHandler.handshakeFuture(),
+ clientHandler.handshakeFuture(),
+ clientHandler);
+ } finally {
+ releaseQuietly(clientChannel);
+ releaseQuietly(serverChannel);
+ }
+ }
+
+ /** {@link EmbeddedChannel#finishAndReleaseAll()} rethrows a failed handshake; ignore it. */
+ private static void releaseQuietly(EmbeddedChannel channel) {
+ try {
+ channel.finishAndReleaseAll();
+ } catch (Throwable ignored) {
+ // asserted through the handshake futures instead.
+ }
+ }
+
+ private static void transferOutbound(EmbeddedChannel from, EmbeddedChannel to) {
+ Object msg;
+ while ((msg = from.readOutbound()) != null) {
+ to.writeInbound(msg);
+ }
+ }
+
+ @Test
+ void testServerConfigRequiresKeystore() {
+ Configuration conf = new Configuration();
+ conf.setString(ConfigOptions.SERVER_SSL_ENABLED_LISTENERS.key(), TestSslUtils.TLS_LISTENER);
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key());
+ }
+
+ @Test
+ void testServerConfigRequiresTruststoreForMtlsListener() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ TestSslUtils.setMutualTlsProtocolMap(conf);
+
+ // without a truststore the server would validate client certificates against the JVM
+ // default truststore, accepting anything issued by a public CA.
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_TRUSTSTORE_PATH.key())
+ .hasMessageContaining(TestSslUtils.TLS_LISTENER);
+ }
+
+ @Test
+ void testMtlsListenerWithTruststoreRequiresClientAuth() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, trustStore);
+ TestSslUtils.setMutualTlsProtocolMap(conf);
+
+ SslConfig config = SslConfig.fromServerConfig(conf).get();
+ assertThat(config.clientAuthListeners()).containsExactly(TestSslUtils.TLS_LISTENER);
+ assertThat(config.requiresClientAuth(TestSslUtils.TLS_LISTENER)).isTrue();
+ assertThat(config.truststorePath()).isEqualTo(trustStore.toString());
+ }
+
+ @Test
+ void testNonMtlsListenerNeedsNoTruststore() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(
+ ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
+ TestSslUtils.TLS_LISTENER + ":PLAINTEXT");
+
+ SslConfig config = SslConfig.fromServerConfig(conf).get();
+ assertThat(config.clientAuthListeners()).isEmpty();
+ assertThat(config.requiresClientAuth(TestSslUtils.TLS_LISTENER)).isFalse();
+ }
+
+ @Test
+ void testMtlsListenerWithoutTlsIsRejected() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ // INTERNAL authenticates clients by certificate but carries no TLS to present one over.
+ conf.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), "INTERNAL:mTLS");
+
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("INTERNAL")
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_ENABLED_LISTENERS.key());
+ }
+
+ @Test
+ void testMtlsListenerWithoutAnyTlsAtAllIsRejected() {
+ Configuration conf = new Configuration();
+ // no TLS anywhere: the config would otherwise be reported as simply having no TLS.
+ conf.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), "CLIENT:mTLS");
+
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("CLIENT")
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_ENABLED_LISTENERS.key());
+ }
+
+ @Test
+ void testMtlsProtocolNameIsCaseInsensitive() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, trustStore);
+ conf.setString(
+ ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
+ TestSslUtils.TLS_LISTENER + ":MTLS");
+
+ // AuthenticationFactory matches authentication protocol names with equalsIgnoreCase.
+ assertThat(
+ SslConfig.fromServerConfig(conf)
+ .get()
+ .requiresClientAuth(TestSslUtils.TLS_LISTENER))
+ .isTrue();
+ }
+
+ @Test
+ void testMissingKeystoreNamesTheFileAndTheOption() {
+ Configuration conf = new Configuration();
+ Path missing = tempDir.resolve("absent.jks");
+ TestSslUtils.setServerSslConfig(conf, missing, null);
+
+ assertThatThrownBy(() -> SslContextFactory.createServerSslContext(conf))
+ .isInstanceOf(FlussRuntimeException.class)
+ .hasMessageContaining("server keystore")
+ .hasMessageContaining(missing.toString())
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key());
+ }
+
+ @Test
+ void testPkcs12TruststoreWithoutPasswordIsRejected() throws Exception {
+ SelfSignedCertificate cert = TestSslUtils.generateCertificate("localhost");
+ Path p12TrustStore = TestSslUtils.createTrustStore(tempDir, "ts.p12", "PKCS12", cert);
+
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, p12TrustStore);
+ conf.setString(ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE.key(), "PKCS12");
+ // PKCS12 keeps its contents under the store password and loads empty without it.
+ conf.removeConfig(ConfigOptions.SERVER_SSL_TRUSTSTORE_PASSWORD);
+
+ assertThatThrownBy(() -> SslContextFactory.createServerSslContext(conf))
+ .isInstanceOf(FlussRuntimeException.class)
+ .hasMessageContaining("no trusted certificates")
+ .hasMessageContaining(p12TrustStore.toString())
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_TRUSTSTORE_PASSWORD.key());
+ }
+
+ @Test
+ void testJksTruststoreWithoutPasswordStillWorks() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, trustStore);
+ // JKS reads its certificates without the store password, so this stays valid.
+ conf.removeConfig(ConfigOptions.SERVER_SSL_TRUSTSTORE_PASSWORD);
+
+ assertThat(SslContextFactory.createServerSslContext(conf)).isPresent();
+ }
+
+ @Test
+ void testKeystoreWithoutPrivateKeyIsRejected() {
+ Configuration conf = new Configuration();
+ // the truststore holds a trusted certificate and no key, a plausible copy-paste mistake.
+ TestSslUtils.setServerSslConfig(conf, trustStore, null);
+
+ assertThatThrownBy(() -> SslContextFactory.createServerSslContext(conf))
+ .isInstanceOf(FlussRuntimeException.class)
+ .hasMessageContaining("no private key")
+ .hasMessageContaining(trustStore.toString())
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key());
+ }
+
+ @Test
+ void testClientKeystoreWithoutPrivateKeyIsRejected() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, trustStore);
+
+ assertThatThrownBy(() -> SslContextFactory.createClientSslContext(conf))
+ .isInstanceOf(FlussRuntimeException.class)
+ .hasMessageContaining("no private key")
+ .hasMessageContaining(ConfigOptions.CLIENT_SSL_KEYSTORE_PATH.key());
+ }
+
+ @Test
+ void testWrongKeystorePasswordNamesThePasswordOption() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD.key(), "not-the-password");
+
+ assertThatThrownBy(() -> SslContextFactory.createServerSslContext(conf))
+ .isInstanceOf(FlussRuntimeException.class)
+ .hasMessageContaining(keyStore.toString())
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD.key());
+ }
+
+ @Test
+ void testWrongKeyPasswordNamesTheKeyPasswordOption() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ // the store itself opens; only the private key inside it does not.
+ conf.setString(ConfigOptions.SERVER_SSL_KEY_PASSWORD.key(), "not-the-key-password");
+
+ assertThatThrownBy(() -> SslContextFactory.createServerSslContext(conf))
+ .isInstanceOf(FlussRuntimeException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_KEY_PASSWORD.key())
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD.key());
+ }
+
+ @Test
+ void testClientStoreFailureNamesTheClientOption() {
+ Configuration conf = new Configuration();
+ Path missing = tempDir.resolve("absent-truststore.jks");
+ TestSslUtils.setClientSslConfig(conf, missing, null);
+
+ assertThatThrownBy(() -> SslContextFactory.createClientSslContext(conf))
+ .isInstanceOf(FlussRuntimeException.class)
+ .hasMessageContaining("client truststore")
+ .hasMessageContaining(ConfigOptions.CLIENT_SSL_TRUSTSTORE_PATH.key());
+ }
+
+ @Test
+ void testServerConfigRejectsUnsupportedKeystoreType() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_TYPE.key(), "JKS2");
+
+ // caught here rather than as a KeyStoreException while the SSL context is built.
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_KEYSTORE_TYPE.key())
+ .hasMessageContaining("JKS2");
+ }
+
+ @Test
+ void testServerConfigRejectsUnsupportedTruststoreType() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, trustStore);
+ conf.setString(ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE.key(), "NOPE");
+
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE.key())
+ .hasMessageContaining("NOPE");
+ }
+
+ @Test
+ void testStoreTypeOfAbsentStoreIsNotValidated() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ // no truststore is configured, so its type is never used to load anything.
+ conf.setString(ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE.key(), "NOPE");
+
+ assertThat(SslConfig.fromServerConfig(conf)).isPresent();
+ }
+
+ @Test
+ void testStoreTypeIsCaseInsensitive() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ // JCA looks providers up case-insensitively, so accept what it accepts.
+ conf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_TYPE.key(), "jks");
+
+ assertThat(SslConfig.fromServerConfig(conf)).isPresent();
+ }
+
+ @Test
+ void testClientMutualTlsRequiresTlsEnabled() {
+ Configuration conf = new Configuration();
+ conf.setString(ConfigOptions.CLIENT_SECURITY_PROTOCOL.key(), "mTLS");
+ conf.setString(ConfigOptions.CLIENT_SSL_KEYSTORE_PATH.key(), keyStore.toString());
+
+ assertThatThrownBy(() -> SslConfig.fromClientConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.CLIENT_SECURITY_PROTOCOL.key())
+ .hasMessageContaining(ConfigOptions.CLIENT_SSL_ENABLED.key());
+ }
+
+ @Test
+ void testClientMutualTlsRequiresKeystore() {
+ Configuration conf = new Configuration();
+ // trusts the server and speaks TLS, but has no certificate of its own to present.
+ TestSslUtils.setClientSslConfig(conf, trustStore, null);
+ conf.setString(ConfigOptions.CLIENT_SECURITY_PROTOCOL.key(), "mTLS");
+
+ assertThatThrownBy(() -> SslConfig.fromClientConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.CLIENT_SSL_KEYSTORE_PATH.key());
+ }
+
+ @Test
+ void testClientMutualTlsWithTlsAndKeystoreIsAccepted() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, keyStore);
+ conf.setString(ConfigOptions.CLIENT_SECURITY_PROTOCOL.key(), "MTLS");
+
+ // protocol names are compared ignoring case, as AuthenticationFactory compares them.
+ assertThat(SslConfig.fromClientConfig(conf)).isPresent();
+ }
+
+ @Test
+ void testClientConfigRejectsUnknownEndpointIdentificationAlgorithm() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, null);
+ conf.setString(ConfigOptions.CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM.key(), "htps");
+
+ // the JDK accepts the name and then fails every handshake with
+ // "Unknown identification algorithm: htps".
+ assertThatThrownBy(() -> SslConfig.fromClientConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(
+ ConfigOptions.CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM.key())
+ .hasMessageContaining("htps");
+ }
+
+ @Test
+ void testClientConfigAcceptsKnownEndpointIdentificationAlgorithms() {
+ for (String algorithm : new String[] {"https", "HTTPS", "ldaps", ""}) {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, null);
+ conf.setString(
+ ConfigOptions.CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM.key(), algorithm);
+
+ assertThat(SslConfig.fromClientConfig(conf)).isPresent();
+ }
+ }
+
+ @Test
+ void testClientConfigRejectsUnsupportedTruststoreType() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, null);
+ conf.setString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_TYPE.key(), "NOPE");
+
+ assertThatThrownBy(() -> SslConfig.fromClientConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.CLIENT_SSL_TRUSTSTORE_TYPE.key())
+ .hasMessageContaining("NOPE");
+ }
+
+ @Test
+ void testDefaultProtocolsAreNarrowedToWhatTheJvmSupports() {
+ // the 8u252 case: the JDK has no TLS 1.3, so the default must fall back to TLS 1.2
+ // rather than fail validation on a list the operator never chose.
+ assertThat(
+ SslConfig.narrowToSupported(
+ Arrays.asList("TLSv1.2", "TLSv1.3"),
+ new LinkedHashSet<>(Collections.singletonList("TLSv1.2")),
+ ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS))
+ .containsExactly("TLSv1.2");
+ }
+
+ @Test
+ void testDefaultProtocolsNoneSupportedIsRejected() {
+ assertThatThrownBy(
+ () ->
+ SslConfig.narrowToSupported(
+ Collections.singletonList("TLSv1.3"),
+ new LinkedHashSet<>(Collections.singletonList("TLSv1.2")),
+ ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS.key());
+ }
+
+ @Test
+ void testUnsetProtocolsAreAllSupportedOnThisJvm() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+
+ // no explicit list: whatever comes out must be usable by this JVM.
+ assertThat(SslConfig.fromServerConfig(conf).get().enabledProtocols()).isNotEmpty();
+ }
+
+ @Test
+ void testServerConfigRejectsUnsupportedProtocol() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS.key(), "TLSv1.2,TLSv1.4");
+
+ // caught here rather than per connection, where the engine reports "Unsupported protocol"
+ // without naming the option at fault.
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS.key())
+ .hasMessageContaining("TLSv1.4");
+ }
+
+ @Test
+ void testServerConfigRejectsEmptyProtocolList() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS.key(), "");
+
+ // an engine built with no enabled protocol starts fine and fails every handshake.
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS.key());
+ }
+
+ @Test
+ void testServerConfigRejectsUnsupportedCipherSuite() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+ conf.setString(
+ ConfigOptions.SERVER_SSL_CIPHER_SUITES.key(),
+ "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_BOGUS_CIPHER");
+
+ assertThatThrownBy(() -> SslConfig.fromServerConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.SERVER_SSL_CIPHER_SUITES.key())
+ .hasMessageContaining("TLS_BOGUS_CIPHER");
+ }
+
+ @Test
+ void testClientConfigRejectsUnsupportedProtocol() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, null);
+ conf.setString(ConfigOptions.CLIENT_SSL_ENABLED_PROTOCOLS.key(), "TLSv13");
+
+ assertThatThrownBy(() -> SslConfig.fromClientConfig(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.CLIENT_SSL_ENABLED_PROTOCOLS.key())
+ .hasMessageContaining("TLSv13");
+ }
+
+ @Test
+ void testServerConfigEmptyWithoutEnabledListeners() {
+ // key material alone does not switch TLS on: no listener is enabled.
+ Configuration conf = new Configuration();
+ conf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key(), keyStore.toString());
+ conf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD.key(), TestSslUtils.PASSWORD);
+
+ assertThat(SslConfig.fromServerConfig(conf)).isNotPresent();
+ assertThat(SslContextFactory.createServerSslContext(conf)).isNotPresent();
+ }
+
+ @Test
+ void testClientConfigEmptyWhenSslDisabled() {
+ Configuration conf = new Configuration();
+
+ assertThat(SslConfig.fromClientConfig(conf)).isNotPresent();
+ assertThat(SslContextFactory.createClientSslContext(conf)).isNotPresent();
+ }
+
+ @Test
+ void testServerConfigExposesEnabledListeners() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setServerSslConfig(conf, keyStore, null);
+
+ assertThat(SslConfig.fromServerConfig(conf).get().enabledListeners())
+ .containsExactly(TestSslUtils.TLS_LISTENER);
+ assertThat(SslConfig.fromClientConfig(clientConf()).get().enabledListeners()).isEmpty();
+ }
+
+ private Configuration clientConf() {
+ Configuration conf = new Configuration();
+ TestSslUtils.setClientSslConfig(conf, trustStore, null);
+ return conf;
+ }
+}
diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/TestSslUtils.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/TestSslUtils.java
new file mode 100644
index 0000000000..01cc6c2be3
--- /dev/null
+++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/TestSslUtils.java
@@ -0,0 +1,138 @@
+/*
+ * 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.fluss.rpc.netty.ssl;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.shaded.netty4.io.netty.handler.ssl.util.SelfSignedCertificate;
+
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyStore;
+import java.security.cert.Certificate;
+
+/**
+ * Test helpers for generating self-signed certificates and JKS keystores/truststores on the fly, so
+ * TLS tests do not need committed key material. Backed by Netty's {@link SelfSignedCertificate},
+ * which generates the certificate with BouncyCastle when it is on the classpath — hence the
+ * test-scoped {@code bcpkix-jdk18on} dependency. Its fallback generator reaches into {@code
+ * sun.security.x509}, which needs {@code --add-opens} from JDK 9 on and stops working altogether
+ * from JDK 18 on.
+ */
+public class TestSslUtils {
+
+ public static final String PASSWORD = "test-password";
+
+ /** The listener name {@link #setServerSslConfig} enables TLS for. */
+ public static final String TLS_LISTENER = "CLIENT";
+
+ private TestSslUtils() {}
+
+ /** Generate a self-signed certificate whose subject CN is {@code fqdn}. */
+ public static SelfSignedCertificate generateCertificate(String fqdn) throws Exception {
+ return new SelfSignedCertificate(fqdn);
+ }
+
+ /** Create a JKS keystore file holding the private key and certificate of {@code cert}. */
+ public static Path createKeyStore(Path dir, String fileName, SelfSignedCertificate cert)
+ throws Exception {
+ return createKeyStore(dir, fileName, "JKS", cert);
+ }
+
+ /**
+ * Create a keystore file of the given {@code storeType} (as accepted by {@code
+ * security.ssl.keystore.type}) holding the private key and certificate of {@code cert}.
+ */
+ public static Path createKeyStore(
+ Path dir, String fileName, String storeType, SelfSignedCertificate cert)
+ throws Exception {
+ KeyStore keyStore = KeyStore.getInstance(storeType);
+ keyStore.load(null, null);
+ keyStore.setKeyEntry(
+ "key", cert.key(), PASSWORD.toCharArray(), new Certificate[] {cert.cert()});
+ return store(dir, fileName, keyStore);
+ }
+
+ /** Create a JKS truststore file trusting the certificate of {@code cert}. */
+ public static Path createTrustStore(Path dir, String fileName, SelfSignedCertificate cert)
+ throws Exception {
+ return createTrustStore(dir, fileName, "JKS", cert);
+ }
+
+ /**
+ * Create a truststore file of the given {@code storeType} (as accepted by {@code
+ * security.ssl.truststore.type}) trusting the certificate of {@code cert}.
+ */
+ public static Path createTrustStore(
+ Path dir, String fileName, String storeType, SelfSignedCertificate cert)
+ throws Exception {
+ KeyStore trustStore = KeyStore.getInstance(storeType);
+ trustStore.load(null, null);
+ trustStore.setCertificateEntry("cert", cert.cert());
+ return store(dir, fileName, trustStore);
+ }
+
+ private static Path store(Path dir, String fileName, KeyStore keyStore) throws Exception {
+ Path path = dir.resolve(fileName);
+ try (OutputStream out = Files.newOutputStream(path)) {
+ keyStore.store(out, PASSWORD.toCharArray());
+ }
+ return path;
+ }
+
+ /**
+ * Populate {@code conf} with the server-side {@code security.ssl.*} transport options: TLS
+ * enabled for the {@link #TLS_LISTENER} listener, a keystore (and, when {@code trustStore !=
+ * null}, a truststore — needed for mTLS listeners). Mark the listener as requiring a client
+ * certificate with {@link #setMutualTlsProtocolMap}.
+ */
+ public static void setServerSslConfig(Configuration conf, Path keyStore, Path trustStore) {
+ conf.setString(ConfigOptions.SERVER_SSL_ENABLED_LISTENERS.key(), TLS_LISTENER);
+ conf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key(), keyStore.toString());
+ conf.setString(ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD.key(), PASSWORD);
+ if (trustStore != null) {
+ conf.setString(ConfigOptions.SERVER_SSL_TRUSTSTORE_PATH.key(), trustStore.toString());
+ conf.setString(ConfigOptions.SERVER_SSL_TRUSTSTORE_PASSWORD.key(), PASSWORD);
+ }
+ }
+
+ /**
+ * Make {@link #TLS_LISTENER} an {@code mTLS} listener via {@code security.protocol.map}, i.e.
+ * one that requires a client certificate.
+ */
+ public static void setMutualTlsProtocolMap(Configuration conf) {
+ conf.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), TLS_LISTENER + ":mTLS");
+ }
+
+ /**
+ * Populate {@code conf} with the client-side {@code client.security.ssl.*} options for a TLS
+ * truststore (and, when {@code keyStore != null}, a keystore for mutual TLS).
+ */
+ public static void setClientSslConfig(Configuration conf, Path trustStore, Path keyStore) {
+ conf.setBoolean(ConfigOptions.CLIENT_SSL_ENABLED.key(), true);
+ if (trustStore != null) {
+ conf.setString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_PATH.key(), trustStore.toString());
+ conf.setString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_PASSWORD.key(), PASSWORD);
+ }
+ if (keyStore != null) {
+ conf.setString(ConfigOptions.CLIENT_SSL_KEYSTORE_PATH.key(), keyStore.toString());
+ conf.setString(ConfigOptions.CLIENT_SSL_KEYSTORE_PASSWORD.key(), PASSWORD);
+ }
+ }
+}