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> 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> 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> 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 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 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("Format of the server keystore: `JKS` or `PKCS12`."); + + public static final ConfigOption 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 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 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("Format of the server truststore: `JKS` or `PKCS12`."); + public static final ConfigOption 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 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> 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> CLIENT_SSL_CIPHER_SUITES = + key("client.security.ssl.cipher.suites") + .stringType() + .asList() + .defaultValues() + .withDescription("Cipher suites enabled for client connections."); + + public static final ConfigOption 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 CLIENT_SSL_TRUSTSTORE_PASSWORD = + key("client.security.ssl.truststore.password") + .stringType() + .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("Format of the client truststore: `JKS` or `PKCS12`."); + + public static final ConfigOption 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 CLIENT_SSL_KEYSTORE_PASSWORD = + key("client.security.ssl.keystore.password") + .stringType() + .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("Format of the client keystore: `JKS` or `PKCS12`."); + + public static final ConfigOption 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 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 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. + + 1.85 + + org.apache.fluss @@ -65,6 +69,35 @@ test-jar test + + + + org.bouncycastle + bcpkix-jdk18on + ${bouncycastle.version} + test + + + + org.bouncycastle + bcutil-jdk18on + ${bouncycastle.version} + test + + + + org.bouncycastle + bcprov-jdk18on + ${bouncycastle.version} + test + 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. + * + *

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

The per-listener client-certificate requirement is 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 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 enabledListeners; + + /** + * Server-only: the subset of {@link #enabledListeners} that requires a client certificate + * (empty for a client config). + */ + private final Set clientAuthListeners; + + 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; + + private SslConfig( + List enabledListeners, + Set clientAuthListeners, + 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) { + 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 fromServerConfig(Configuration conf) { + List 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 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 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 enabledProtocols = + effectiveProtocols(conf, ConfigOptions.SERVER_SSL_ENABLED_PROTOCOLS); + List 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 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 enabledProtocols = + effectiveProtocols(conf, ConfigOptions.CLIENT_SSL_ENABLED_PROTOCOLS); + List 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 effectiveProtocols( + Configuration conf, ConfigOption> option) { + Optional> configured = conf.getOptional(option); + if (configured.isPresent()) { + return orEmpty(configured.get()); + } + Set supported = + new LinkedHashSet<>( + Arrays.asList(supportedAlgorithmsProbe().getSupportedProtocols())); + return narrowToSupported(orEmpty(option.defaultValue()), supported, option); + } + + @VisibleForTesting + static List narrowToSupported( + List defaults, Set supported, ConfigOption> option) { + List 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. + * + *

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 enabledProtocols, + ConfigOption> protocolsOption, + List cipherSuites, + ConfigOption> cipherSuitesOption) { + if (enabledProtocols.isEmpty()) { + throw new IllegalConfigurationException( + "'%s' must list at least one TLS protocol.", protocolsOption.key()); + } + + SSLEngine probe = supportedAlgorithmsProbe(); + + List 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 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 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 unsupported(List configured, String[] supported) { + Set 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 orEmpty(@Nullable List 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 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 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 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..660f6a7d97 --- /dev/null +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/ssl/SslContextFactory.java @@ -0,0 +1,382 @@ +/* + * 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.TrustManager; +import javax.net.ssl.TrustManagerFactory; +import javax.net.ssl.X509TrustManager; + +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. + * + *

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 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 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); + try { + TrustManagerFactory tmf = + TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(trustStore); + checkHoldsCertificate(store, tmf, path, storePassword); + 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 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, TrustManagerFactory tmf, String path, @Nullable String storePassword) { + for (TrustManager tm : tmf.getTrustManagers()) { + if (tm instanceof X509TrustManager + && ((X509TrustManager) tm).getAcceptedIssuers().length > 0) { + return; + } + } + String hint = + storePassword == null + ? String.format( + " A PKCS12 store loads empty when '%s' is not set.", + store.passwordKey) + : ""; + throw new FlussRuntimeException( + String.format( + "The %s at '%s', configured by '%s', holds no trusted certificates.%s " + + "Every TLS handshake would fail with \"the trustAnchors " + + "parameter must be non-empty\".", + store.what, path, store.pathKey, hint)); + } + + /** + * 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 serverHandshake; + private final Future clientHandshake; + private final SslHandler clientHandler; + + private HandshakeResult( + Future serverHandshake, + Future 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); + } + } +}