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 233902cb39..6f49a0aad5 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 @@ -609,6 +609,107 @@ 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> SERVER_SSL_ENABLED_LISTENERS = + key("security.ssl.enabled.listeners") + .stringType() + .asList() + .noDefaultValue() + .withDescription( + "A comma-separated list of listener names for which TLS transport " + + "encryption is enabled, e.g. `CLIENT,INTERNAL`. A keystore " + + "(`security.ssl.keystore.path`) must be configured when any " + + "listener is listed here. Listeners not listed accept plaintext " + + "connections. TLS is orthogonal to the authentication protocol " + + "configured via `security.protocol.map`."); + + public static final ConfigOption> SERVER_SSL_ENABLED_PROTOCOLS = + key("security.ssl.enabled.protocols") + .stringType() + .asList() + .defaultValues("TLSv1.2", "TLSv1.3") + .withDescription( + "The list of TLS protocols enabled for incoming connections. " + + "`TLSv1.2` is kept in the default list for compatibility with " + + "JDK 8 builds older than 8u261 (which do not support TLS 1.3)."); + + public static final ConfigOption> SERVER_SSL_CIPHER_SUITES = + key("security.ssl.cipher.suites") + .stringType() + .asList() + .noDefaultValue() + .withDescription( + "A comma-separated list of cipher suites enabled for TLS connections. " + + "If not set, the JDK/provider defaults for the negotiated " + + "protocol are used."); + + public static final ConfigOption SERVER_SSL_KEYSTORE_PATH = + key("security.ssl.keystore.path") + .stringType() + .noDefaultValue() + .withDescription( + "The location of the keystore file holding the server certificate and " + + "private key. Required when any listener is listed in " + + "`security.ssl.enabled.listeners`."); + + public static final ConfigOption SERVER_SSL_KEYSTORE_PASSWORD = + key("security.ssl.keystore.password") + .passwordType() + .noDefaultValue() + .withDescription("The password to access the server keystore."); + + public static final ConfigOption SERVER_SSL_KEYSTORE_TYPE = + key("security.ssl.keystore.type") + .stringType() + .defaultValue("JKS") + .withDescription( + "The format of the server keystore file. Supported values are `JKS` and " + + "`PKCS12`. The default is `JKS`."); + + public static final ConfigOption SERVER_SSL_KEY_PASSWORD = + key("security.ssl.key.password") + .passwordType() + .noDefaultValue() + .withDescription( + "The password of the private key in the server keystore. If not set, the " + + "keystore password (`security.ssl.keystore.password`) is used."); + + public static final ConfigOption SERVER_SSL_TRUSTSTORE_PATH = + key("security.ssl.truststore.path") + .stringType() + .noDefaultValue() + .withDescription( + "The location of the truststore file holding the certificates trusted " + + "by the server. Required for `mTLS` listeners (a listener whose " + + "`security.protocol.map` entry is `mTLS`), to validate client " + + "certificates."); + + public static final ConfigOption SERVER_SSL_TRUSTSTORE_PASSWORD = + key("security.ssl.truststore.password") + .passwordType() + .noDefaultValue() + .withDescription("The password to access the server truststore."); + + public static final ConfigOption SERVER_SSL_TRUSTSTORE_TYPE = + key("security.ssl.truststore.type") + .stringType() + .defaultValue("JKS") + .withDescription( + "The format of the server truststore file. Supported values are `JKS` " + + "and `PKCS12`. The default is `JKS`."); + + public static final ConfigOption SERVER_SSL_RELOAD_INTERVAL = + key("security.ssl.reload.interval") + .durationType() + .defaultValue(Duration.ofMinutes(5)) + .withDescription( + "How often the server checks the keystore/truststore files for changes and " + + "rebuilds the SSL context (certificate hot-reload). Set to 0 to " + + "disable periodic reloading."); + public static final ConfigOption TABLET_SERVER_ID = key("tablet-server.id") .intType() @@ -1433,6 +1534,112 @@ 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 CLIENT_SSL_ENABLED = + key("client.security.ssl.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether the client establishes TLS-encrypted connections to the server. " + + "Must match the security posture of the listener the client " + + "connects to (see `security.ssl.enabled.listeners` on the server)."); + + public static final ConfigOption> CLIENT_SSL_ENABLED_PROTOCOLS = + key("client.security.ssl.enabled.protocols") + .stringType() + .asList() + .defaultValues("TLSv1.2", "TLSv1.3") + .withDescription( + "The list of TLS protocols enabled for client connections. `TLSv1.2` is " + + "kept in the default list for compatibility with JDK 8 builds " + + "older than 8u261."); + + public static final ConfigOption> CLIENT_SSL_CIPHER_SUITES = + key("client.security.ssl.cipher.suites") + .stringType() + .asList() + .noDefaultValue() + .withDescription( + "A comma-separated list of cipher suites enabled for client TLS " + + "connections. If not set, JDK/provider defaults are used."); + + public static final ConfigOption CLIENT_SSL_TRUSTSTORE_PATH = + key("client.security.ssl.truststore.path") + .stringType() + .noDefaultValue() + .withDescription( + "The location of the truststore file holding the certificates the client " + + "uses to verify the server certificate. If unset, the JVM " + + "default trust store is used."); + + public static final ConfigOption CLIENT_SSL_TRUSTSTORE_PASSWORD = + key("client.security.ssl.truststore.password") + .passwordType() + .noDefaultValue() + .withDescription("The password to access the client truststore."); + + public static final ConfigOption CLIENT_SSL_TRUSTSTORE_TYPE = + key("client.security.ssl.truststore.type") + .stringType() + .defaultValue("JKS") + .withDescription( + "The format of the client truststore file. Supported values are `JKS` " + + "and `PKCS12`. The default is `JKS`."); + + public static final ConfigOption CLIENT_SSL_KEYSTORE_PATH = + key("client.security.ssl.keystore.path") + .stringType() + .noDefaultValue() + .withDescription( + "The location of the keystore file holding the client certificate and " + + "private key. Only needed for mutual TLS, where the client " + + "presents a certificate to the server."); + + public static final ConfigOption CLIENT_SSL_KEYSTORE_PASSWORD = + key("client.security.ssl.keystore.password") + .passwordType() + .noDefaultValue() + .withDescription("The password to access the client keystore."); + + public static final ConfigOption CLIENT_SSL_KEYSTORE_TYPE = + key("client.security.ssl.keystore.type") + .stringType() + .defaultValue("JKS") + .withDescription( + "The format of the client keystore file. Supported values are `JKS` and " + + "`PKCS12`. The default is `JKS`."); + + public static final ConfigOption CLIENT_SSL_KEY_PASSWORD = + key("client.security.ssl.key.password") + .passwordType() + .noDefaultValue() + .withDescription( + "The password of the private key in the client keystore. If not set, the " + + "keystore password (`client.security.ssl.keystore.password`) is " + + "used."); + + public static final ConfigOption CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM = + key("client.security.ssl.endpoint.identification.algorithm") + .stringType() + .defaultValue("https") + .withDescription( + "The endpoint identification algorithm used by the client to verify the " + + "server hostname against the server certificate. The default " + + "`https` enables hostname verification; an empty string disables " + + "it (not recommended in production)."); + + public static final ConfigOption CLIENT_SSL_RELOAD_INTERVAL = + key("client.security.ssl.reload.interval") + .durationType() + .defaultValue(Duration.ofMinutes(5)) + .withDescription( + "How often the client checks the keystore/truststore files for changes and " + + "rebuilds the SSL context (certificate hot-reload). Set to 0 to " + + "disable periodic reloading."); + public static final ConfigOption CLIENT_SCANNER_LOG_FETCH_MAX_BYTES = key("client.scanner.log.fetch.max-bytes") .memoryType() 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..296ac51e9e --- /dev/null +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslConfig.java @@ -0,0 +1,192 @@ +/* + * 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.config.Password; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** + * The parsed and validated TLS configuration for one side (server or client) of an RPC connection. + * + *

This is a thin, immutable holder over the {@code security.ssl.*} (server) and {@code + * client.security.ssl.*} (client) {@link ConfigOptions}. Validation that does not depend on the + * actual key material (e.g. "a keystore must be configured for a TLS server") happens in the {@link + * #fromServerConfig} / {@link #fromClientConfig} factory methods so misconfiguration fails fast + * with a clear message. + * + *

The per-listener client-certificate requirement is not held here: it is derived per + * listener from the listener's authentication protocol ({@code mTLS} ⇒ require) when the server + * pipeline is wired. The server context is always built with a trust manager when a truststore is + * configured, so any listener can request client certs. + */ +@Internal +public final class SslConfig { + + private final List enabledProtocols; + private final List 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; + + /** How often to poll the key material for changes (0 disables periodic reload). */ + private final Duration reloadInterval; + + private SslConfig( + List enabledProtocols, + List cipherSuites, + @Nullable String keystorePath, + @Nullable String keystorePassword, + String keystoreType, + @Nullable String keyPassword, + @Nullable String truststorePath, + @Nullable String truststorePassword, + String truststoreType, + String endpointIdentificationAlgorithm, + Duration reloadInterval) { + 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; + this.reloadInterval = reloadInterval; + } + + /** Build and validate the server-side TLS configuration. */ + public static SslConfig fromServerConfig(Configuration conf) { + String keystorePath = conf.getString(ConfigOptions.SERVER_SSL_KEYSTORE_PATH); + checkArgument( + keystorePath != null, + "'%s' must be configured when any listener enables TLS via '%s'.", + ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key(), + ConfigOptions.SERVER_SSL_ENABLED_LISTENERS.key()); + + return new SslConfig( + conf.get(ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS), + orEmpty(conf.get(ConfigOptions.SERVER_SSL_CIPHER_SUITES)), + keystorePath, + password(conf.get(ConfigOptions.SERVER_SSL_KEYSTORE_PASSWORD)), + conf.getString(ConfigOptions.SERVER_SSL_KEYSTORE_TYPE), + password(conf.get(ConfigOptions.SERVER_SSL_KEY_PASSWORD)), + conf.getString(ConfigOptions.SERVER_SSL_TRUSTSTORE_PATH), + password(conf.get(ConfigOptions.SERVER_SSL_TRUSTSTORE_PASSWORD)), + conf.getString(ConfigOptions.SERVER_SSL_TRUSTSTORE_TYPE), + "", + conf.get(ConfigOptions.SERVER_SSL_RELOAD_INTERVAL)); + } + + /** Build and validate the client-side TLS configuration. */ + public static SslConfig fromClientConfig(Configuration conf) { + return new SslConfig( + conf.get(ConfigOptions.CLIENT_SSL_ENABLED_PROTOCOLS), + orEmpty(conf.get(ConfigOptions.CLIENT_SSL_CIPHER_SUITES)), + conf.getString(ConfigOptions.CLIENT_SSL_KEYSTORE_PATH), + password(conf.get(ConfigOptions.CLIENT_SSL_KEYSTORE_PASSWORD)), + conf.getString(ConfigOptions.CLIENT_SSL_KEYSTORE_TYPE), + password(conf.get(ConfigOptions.CLIENT_SSL_KEY_PASSWORD)), + conf.getString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_PATH), + password(conf.get(ConfigOptions.CLIENT_SSL_TRUSTSTORE_PASSWORD)), + conf.getString(ConfigOptions.CLIENT_SSL_TRUSTSTORE_TYPE), + conf.getString(ConfigOptions.CLIENT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM), + conf.get(ConfigOptions.CLIENT_SSL_RELOAD_INTERVAL)); + } + + @Nullable + private static String password(@Nullable Password password) { + return password == null ? null : password.value(); + } + + private static List orEmpty(@Nullable List list) { + return list == null ? Collections.emptyList() : list; + } + + public String[] enabledProtocols() { + return enabledProtocols.toArray(new String[0]); + } + + public List 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; + } + + public Duration reloadInterval() { + return reloadInterval; + } +} 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..b1f90dd273 --- /dev/null +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslContextFactory.java @@ -0,0 +1,188 @@ +/* + * 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.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.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLEngine; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.TrustManagerFactory; + +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.security.KeyStore; + +/** + * 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. + * + *

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}; the per-listener client-certificate + * requirement is then 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. */ + public static SslContext createServerSslContext(Configuration conf) { + return createServerSslContext(SslConfig.fromServerConfig(conf)); + } + + /** Build a server {@link SslContext} from a parsed {@link SslConfig}. */ + public static SslContext createServerSslContext(SslConfig config) { + try { + KeyManagerFactory kmf = + keyManagerFactory( + 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( + config.truststorePath(), + config.truststoreType(), + config.truststorePassword())); + } + 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. */ + public static SslContext createClientSslContext(Configuration conf) { + return createClientSslContext(SslConfig.fromClientConfig(conf)); + } + + /** Build a client {@link SslContext} from a parsed {@link SslConfig}. */ + public static SslContext createClientSslContext(SslConfig config) { + try { + 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( + 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( + config.keystorePath(), + config.keystoreType(), + config.keystorePassword(), + config.keyPassword())); + } + 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 (an {@code mTLS} listener) the engine demands a client certificate + * during the handshake. + */ + 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( + String path, String type, String storePassword, String keyPassword) throws Exception { + KeyStore keyStore = loadKeyStore(path, type, storePassword); + KeyManagerFactory kmf = + KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(keyStore, keyPassword == null ? null : keyPassword.toCharArray()); + return kmf; + } + + private static TrustManagerFactory trustManagerFactory( + String path, String type, String storePassword) throws Exception { + KeyStore trustStore = loadKeyStore(path, type, storePassword); + TrustManagerFactory tmf = + TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(trustStore); + return tmf; + } + + private static KeyStore loadKeyStore(String path, String type, String password) + throws Exception { + KeyStore keyStore = KeyStore.getInstance(type); + try (InputStream in = Files.newInputStream(Paths.get(path))) { + keyStore.load(in, password == null ? null : password.toCharArray()); + } + return keyStore; + } +} 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..ad18ccb3bf --- /dev/null +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/SslContextFactoryTest.java @@ -0,0 +1,205 @@ +/* + * 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.buffer.ByteBufAllocator; +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.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; + +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); + 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); + 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); + 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); + 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); + assertThat(config.keyPassword()).isEqualTo(TestSslUtils.PASSWORD); + + conf.setString(ConfigOptions.SERVER_SSL_KEY_PASSWORD.key(), "key-only-password"); + config = SslConfig.fromServerConfig(conf); + assertThat(config.keyPassword()).isEqualTo("key-only-password"); + } + + @Test + void testClientSslHandlerEndpointIdentification() { + Configuration conf = new Configuration(); + TestSslUtils.setClientSslConfig(conf, trustStore, null); + SslContext sslContext = SslContextFactory.createClientSslContext(conf); + + 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); + + 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); + + SslHandler serverHandler = + SslContextFactory.createServerSslHandler( + SslContextFactory.createServerSslContext(serverConf), + ByteBufAllocator.DEFAULT, + false); + // endpoint identification disabled here: this raw handshake test only checks encryption. + SslHandler clientHandler = + SslContextFactory.createClientSslHandler( + SslContextFactory.createClientSslContext(clientConf), + ByteBufAllocator.DEFAULT, + "localhost", + 9123, + ""); + + EmbeddedChannel serverChannel = new EmbeddedChannel(serverHandler); + EmbeddedChannel clientChannel = new EmbeddedChannel(clientHandler); + try { + // Pump the handshake bytes back and forth until both sides complete. + for (int i = 0; + i < 20 + && !(clientHandler.handshakeFuture().isDone() + && serverHandler.handshakeFuture().isDone()); + i++) { + transferOutbound(clientChannel, serverChannel); + transferOutbound(serverChannel, clientChannel); + } + + assertThat(clientHandler.handshakeFuture().isSuccess()).isTrue(); + assertThat(serverHandler.handshakeFuture().isSuccess()).isTrue(); + // Proves real encryption was negotiated (not a NULL/plaintext cipher). + assertThat(clientHandler.engine().getSession().getProtocol()).startsWith("TLS"); + assertThat(clientHandler.engine().getSession().getCipherSuite()).doesNotContain("NULL"); + } finally { + clientChannel.finishAndReleaseAll(); + serverChannel.finishAndReleaseAll(); + } + } + + 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(); + assertThatThrownBy(() -> SslConfig.fromServerConfig(conf)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(ConfigOptions.SERVER_SSL_KEYSTORE_PATH.key()); + } +} 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..9b0d45500f --- /dev/null +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/ssl/TestSslUtils.java @@ -0,0 +1,102 @@ +/* + * 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}. + */ +public class TestSslUtils { + + public static final String PASSWORD = "test-password"; + + 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 { + KeyStore keyStore = KeyStore.getInstance("JKS"); + 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 { + KeyStore trustStore = KeyStore.getInstance("JKS"); + 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 for a TLS + * keystore (and, when {@code trustStore != null}, a truststore — needed for mTLS listeners). + * The per-listener client-certificate requirement is derived from {@code security.protocol.map} + * ({@code mTLS}), which the caller sets. + */ + public static void setServerSslConfig(Configuration conf, Path keyStore, Path trustStore) { + 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); + } + } + + /** + * 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); + } + } +}