This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b49684cb4a8 [FLINK-40727][security] Support comma-separated
security.ssl.protocol and fix internal SSL hostname verification regression
b49684cb4a8 is described below
commit b49684cb4a80106872b07cffd03261285c3bb271
Author: Gabor Somogyi <[email protected]>
AuthorDate: Wed Sep 23 17:25:42 2026 +0200
[FLINK-40727][security] Support comma-separated security.ssl.protocol and
fix internal SSL hostname verification regression
---
.../generated/security_configuration.html | 2 +-
.../shortcodes/generated/security_ssl_section.html | 2 +-
.../flink/configuration/SecurityOptions.java | 15 +-
flink-end-to-end-tests/test-scripts/common_ssl.sh | 5 +
.../test-scripts/test_file_sink.sh | 85 ++++--
.../runtime/rpc/pekko/CustomSSLEngineProvider.java | 61 ++++-
.../apache/flink/runtime/rpc/pekko/PekkoUtils.java | 2 +-
.../rpc/pekko/CustomSSLEngineProviderTest.java | 181 +++++++++++++
.../flink-rpc-akka/src/test/resources/rpc.keystore | Bin 0 -> 2576 bytes
.../src/test/resources/rpc.truststore | Bin 0 -> 1110 bytes
.../org/apache/flink/runtime/net/SSLUtils.java | 7 +
.../org/apache/flink/runtime/net/SSLUtilsTest.java | 288 +++++++++++++++++++++
tools/azure-pipelines/e2e-template.yml | 17 +-
13 files changed, 628 insertions(+), 37 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/security_configuration.html
b/docs/layouts/shortcodes/generated/security_configuration.html
index d4fc20d1981..757e8ff979d 100644
--- a/docs/layouts/shortcodes/generated/security_configuration.html
+++ b/docs/layouts/shortcodes/generated/security_configuration.html
@@ -186,7 +186,7 @@
<td><h5>security.ssl.protocol</h5></td>
<td style="word-wrap: break-word;">"TLSv1.2"</td>
<td>String</td>
- <td>The SSL protocol version to be supported for the ssl
transport. Note that it doesn’t support comma separated list.</td>
+ <td>The comma separated list of SSL protocol versions to be
supported for the ssl transport, e.g. <code
class="highlighter-rouge">TLSv1.2,TLSv1.3</code>. The highest protocol version
supported by both communication endpoints and for which a matching cipher suite
is available (see <code
class="highlighter-rouge">security.ssl.algorithms</code>) is negotiated; the
connection fails only if none of the listed protocols has a matching cipher
suite.</td>
</tr>
<tr>
<td><h5>security.ssl.provider</h5></td>
diff --git a/docs/layouts/shortcodes/generated/security_ssl_section.html
b/docs/layouts/shortcodes/generated/security_ssl_section.html
index ab9f337500f..7df25d3fd63 100644
--- a/docs/layouts/shortcodes/generated/security_ssl_section.html
+++ b/docs/layouts/shortcodes/generated/security_ssl_section.html
@@ -72,7 +72,7 @@
<td><h5>security.ssl.protocol</h5></td>
<td style="word-wrap: break-word;">"TLSv1.2"</td>
<td>String</td>
- <td>The SSL protocol version to be supported for the ssl
transport. Note that it doesn’t support comma separated list.</td>
+ <td>The comma separated list of SSL protocol versions to be
supported for the ssl transport, e.g. <code
class="highlighter-rouge">TLSv1.2,TLSv1.3</code>. The highest protocol version
supported by both communication endpoints and for which a matching cipher suite
is available (see <code
class="highlighter-rouge">security.ssl.algorithms</code>) is negotiated; the
connection fails only if none of the listed protocols has a matching cipher
suite.</td>
</tr>
<tr>
<td><h5>security.ssl.rest.authentication-enabled</h5></td>
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
index 3cfdca5c596..441b07f569e 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
@@ -503,15 +503,24 @@ public class SecurityOptions {
// ------------------------ ssl parameters --------------------------------
- /** SSL protocol version to be supported. */
+ /** SSL protocol version(s) to be supported. */
@Documentation.Section(Documentation.Sections.SECURITY_SSL)
public static final ConfigOption<String> SSL_PROTOCOL =
key("security.ssl.protocol")
.stringType()
.defaultValue("TLSv1.2")
.withDescription(
- "The SSL protocol version to be supported for the
ssl transport. Note that it doesn’t"
- + " support comma separated list.");
+ Description.builder()
+ .text(
+ "The comma separated list of SSL
protocol versions to be supported for"
+ + " the ssl transport,
e.g. %s. The highest protocol version"
+ + " supported by both
communication endpoints and for which a"
+ + " matching cipher suite
is available (see %s) is negotiated;"
+ + " the connection fails
only if none of the listed protocols"
+ + " has a matching cipher
suite.",
+ code("TLSv1.2,TLSv1.3"),
+ code("security.ssl.algorithms"))
+ .build());
/**
* The standard SSL algorithms to be supported.
diff --git a/flink-end-to-end-tests/test-scripts/common_ssl.sh
b/flink-end-to-end-tests/test-scripts/common_ssl.sh
index 87219c68352..b04aea6d6fa 100644
--- a/flink-end-to-end-tests/test-scripts/common_ssl.sh
+++ b/flink-end-to-end-tests/test-scripts/common_ssl.sh
@@ -79,6 +79,11 @@ function _set_conf_ssl_helper {
if [ "${provider}" = "OPENSSL" -a "${provider_lib}" = "dynamic" ]; then
cp $FLINK_DIR/opt/flink-shaded-netty-tcnative-dynamic-*.jar
$FLINK_DIR/lib/
+ # if the CI prepared a newer OpenSSL than the system one (see
e2e-template.yml), prefer it;
+ # local/developer runs without that variable set keep using the system
OpenSSL as before
+ if [ -n "${FLINK_E2E_OPENSSL32_LIB:-}" ]; then
+ export
LD_LIBRARY_PATH="${FLINK_E2E_OPENSSL32_LIB}${LD_LIBRARY_PATH:+:${LD_LIBRARY_PATH}}"
+ fi
elif [ "${provider}" = "OPENSSL" -a "${provider_lib}" = "static" ]; then
# Flink is not providing the statically-linked library because of
potential licensing issues
# -> we need to build it ourselves
diff --git a/flink-end-to-end-tests/test-scripts/test_file_sink.sh
b/flink-end-to-end-tests/test-scripts/test_file_sink.sh
index 5de4731ca4c..53735b21adf 100755
--- a/flink-end-to-end-tests/test-scripts/test_file_sink.sh
+++ b/flink-end-to-end-tests/test-scripts/test_file_sink.sh
@@ -21,15 +21,6 @@ OUT_TYPE="${1:-local}" # other type: s3
source "$(dirname "$0")"/common.sh
-# LOCAL_JOB_OUTPUT_PATH is a local folder that can be used as a download
folder for remote data
-# the helper functions will access this folder
-RANDOM_PREFIX="temp/test_file_sink-$(uuidgen)"
-LOCAL_JOB_OUTPUT_PATH="$TEST_DATA_DIR/${RANDOM_PREFIX}"
-mkdir -p "${LOCAL_JOB_OUTPUT_PATH}"
-
-# JOB_OUTPUT_PATH is the location where the job writes its data to
-JOB_OUTPUT_PATH="${LOCAL_JOB_OUTPUT_PATH}"
-
###################################
# Get all lines in part files and sort them numerically.
#
@@ -64,12 +55,6 @@ elif [ "${OUT_TYPE}" == "s3" ]; then
source "$(dirname "$0")"/common_s3_seaweedfs.sh
s3_setup hadoop
- # overwrites JOB_OUTPUT_PATH to point to S3
- S3_DATA_PREFIX="${RANDOM_PREFIX}"
- S3_CHECKPOINT_PREFIX="${RANDOM_PREFIX}-chk"
- JOB_OUTPUT_PATH="s3://$IT_CASE_S3_BUCKET/${S3_DATA_PREFIX}"
- set_config_key "execution.checkpointing.dir"
"s3://$IT_CASE_S3_BUCKET/${S3_CHECKPOINT_PREFIX}"
-
# overwrites implementation for local runs
function get_complete_result {
# copies the data from S3 to the local LOCAL_JOB_OUTPUT_PATH
@@ -79,33 +64,29 @@ elif [ "${OUT_TYPE}" == "s3" ]; then
find "${LOCAL_JOB_OUTPUT_PATH}" -type f \( -iname "part-*" \) -exec cat {}
+ | sort -g
}
- # make sure we delete the file at the end
+ # make sure we delete the data of the current OpenSSL linkage iteration;
also registered as a
+ # safety net for whichever iteration is active if the script exits
unexpectedly
function out_cleanup {
s3_delete_by_full_path_prefix "${S3_DATA_PREFIX}"
s3_delete_by_full_path_prefix "${S3_CHECKPOINT_PREFIX}"
}
-
on_exit out_cleanup
else
echo "[ERROR] Unknown out type: ${OUT_TYPE}"
exit 1
fi
-# randomly set up openSSL with dynamically/statically linked libraries
-OPENSSL_LINKAGE=$(if (( RANDOM % 2 )) ; then echo "dynamic"; else echo
"static"; fi)
-echo "Executing test with ${OPENSSL_LINKAGE} openSSL linkage (random selection
between 'dynamic' and 'static')"
-
-# Temporarily disable SSL in test_file_sink (see FLINK-39002 )
-#set_conf_ssl "mutual" "OPENSSL" "${OPENSSL_LINKAGE}"
-# set_conf_ssl moves netty libraries into FLINK_DIR which we want to rollback
at the end of the test run
-on_exit rollback_openssl_lib
-
set_config_key "metrics.fetcher.update-interval" "2000"
# this test relies on global failovers
set_config_key "jobmanager.execution.failover-strategy" "full"
TEST_PROGRAM_JAR="${END_TO_END_DIR}/flink-file-sink-test/target/FileSinkProgram.jar"
+# set_conf_ssl moves netty libraries into FLINK_DIR; rolled back explicitly
between iterations,
+# and registered here as a safety net for the currently active iteration if
the script exits
+# unexpectedly
+on_exit rollback_openssl_lib
+
###################################
# Waits until a number of values have been written within a timeout.
# If the timeout expires, exit with return code 1.
@@ -195,5 +176,53 @@ function run_file_sink_test {
check_result_hash "File Sink" "$TEST_DATA_DIR/complete_result"
"6727342fdd3aae2129e61fc8f433fb6f"
}
-# usual runtime is ~6 minutes
-run_test_with_timeout 900 run_file_sink_test
+###################################
+# Runs the file sink test once, with the given OpenSSL linkage.
+#
+# Globals:
+# TEST_DATA_DIR, OUT_TYPE, IT_CASE_S3_BUCKET
+# Arguments:
+# $1: OpenSSL linkage, 'dynamic' or 'static'
+# Returns:
+# None
+###################################
+function run_file_sink_test_with_linkage {
+ local openssl_linkage=$1
+
+ echo "Executing test with ${openssl_linkage} openSSL linkage"
+
+ # LOCAL_JOB_OUTPUT_PATH is a local folder that can be used as a download
folder for remote data
+ # the helper functions will access this folder
+ local random_prefix="temp/test_file_sink-${openssl_linkage}-$(uuidgen)"
+ LOCAL_JOB_OUTPUT_PATH="$TEST_DATA_DIR/${random_prefix}"
+ mkdir -p "${LOCAL_JOB_OUTPUT_PATH}"
+
+ # JOB_OUTPUT_PATH is the location where the job writes its data to
+ JOB_OUTPUT_PATH="${LOCAL_JOB_OUTPUT_PATH}"
+
+ if [ "${OUT_TYPE}" == "s3" ]; then
+ # overwrites JOB_OUTPUT_PATH to point to S3
+ S3_DATA_PREFIX="${random_prefix}"
+ S3_CHECKPOINT_PREFIX="${random_prefix}-chk"
+ JOB_OUTPUT_PATH="s3://$IT_CASE_S3_BUCKET/${S3_DATA_PREFIX}"
+ set_config_key "execution.checkpointing.dir"
"s3://$IT_CASE_S3_BUCKET/${S3_CHECKPOINT_PREFIX}"
+ fi
+
+ set_conf_ssl "mutual" "OPENSSL" "${openssl_linkage}"
+
+ run_test_with_timeout 900 run_file_sink_test
+
+ stop_cluster
+ # set_conf_ssl moves netty libraries into FLINK_DIR which we want to
rollback before the next
+ # linkage is set up
+ rollback_openssl_lib
+
+ if [ "${OUT_TYPE}" == "s3" ]; then
+ out_cleanup
+ fi
+}
+
+# usual runtime is ~6 minutes per linkage
+for openssl_linkage in dynamic static; do
+ run_file_sink_test_with_linkage "${openssl_linkage}"
+done
diff --git
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProvider.java
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProvider.java
index e2ea1801ea8..025db2b929c 100644
---
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProvider.java
+++
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProvider.java
@@ -24,6 +24,8 @@ import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.remote.RemoteTransportException;
import org.apache.pekko.remote.transport.netty.ConfigSSLEngineProvider;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
@@ -36,7 +38,16 @@ import java.security.KeyStore;
import java.util.List;
/**
- * Extension of the {@link ConfigSSLEngineProvider} to use a {@link
FingerprintTrustManagerFactory}.
+ * Extension of the {@link ConfigSSLEngineProvider} to use a {@link
FingerprintTrustManagerFactory}
+ * and to support a comma-separated protocol list in {@code security.protocol}.
+ *
+ * <p>{@link ConfigSSLEngineProvider}'s own engine creation reads {@code
security.protocol} as a
+ * single string and passes it unsplit to both {@code
SSLContext.getInstance(String)} and {@code
+ * SSLEngine#setEnabledProtocols(String[])}, so a comma-separated value breaks
it. Its {@code
+ * sslContext} and {@code createSSLEngine} members are {@code private} and
cannot be overridden, so
+ * {@link #createServerSSLEngine()} and {@link #createClientSSLEngine()} are
reimplemented here
+ * instead, splitting the protocol list the same way {@code SSLUtils} does for
the data-plane and
+ * REST SSL contexts.
*/
@SuppressWarnings("deprecation")
public class CustomSSLEngineProvider extends ConfigSSLEngineProvider {
@@ -45,6 +56,11 @@ public class CustomSSLEngineProvider extends
ConfigSSLEngineProvider {
private final List<String> sslCertFingerprints;
private final String sslKeyStoreType;
private final String sslTrustStoreType;
+ private final String[] sslProtocols;
+ private final String[] sslEnabledAlgorithms;
+ private final boolean sslRequireMutualAuthentication;
+
+ private SSLContext sslContext;
public CustomSSLEngineProvider(ActorSystem system) {
super(system);
@@ -55,6 +71,49 @@ public class CustomSSLEngineProvider extends
ConfigSSLEngineProvider {
sslCertFingerprints =
securityConfig.getStringList("cert-fingerprints");
sslKeyStoreType = securityConfig.getString("key-store-type");
sslTrustStoreType = securityConfig.getString("trust-store-type");
+ sslProtocols = securityConfig.getString("protocol").split(",");
+ sslEnabledAlgorithms =
+ securityConfig.getStringList("enabled-algorithms").toArray(new
String[0]);
+ sslRequireMutualAuthentication =
securityConfig.getBoolean("require-mutual-authentication");
+ }
+
+ @Override
+ public SSLEngine createServerSSLEngine() {
+ SSLEngine engine = newEngine();
+ engine.setUseClientMode(false);
+ if (sslRequireMutualAuthentication) {
+ engine.setNeedClientAuth(true);
+ }
+ return engine;
+ }
+
+ @Override
+ public SSLEngine createClientSSLEngine() {
+ SSLEngine engine = newEngine();
+ engine.setUseClientMode(true);
+ return engine;
+ }
+
+ private SSLEngine newEngine() {
+ SSLEngine engine = getSslContext().createSSLEngine();
+ engine.setEnabledCipherSuites(sslEnabledAlgorithms);
+ engine.setEnabledProtocols(sslProtocols);
+ return engine;
+ }
+
+ private synchronized SSLContext getSslContext() {
+ if (sslContext == null) {
+ try {
+ SSLContext context = SSLContext.getInstance("TLS");
+ context.init(keyManagers(), trustManagers(),
createSecureRandom());
+ sslContext = context;
+ } catch (GeneralSecurityException e) {
+ throw new RemoteTransportException(
+ "Server SSL connection could not be established
because SSL context could not be constructed",
+ e);
+ }
+ }
+ return sslContext;
}
@Override
diff --git
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoUtils.java
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoUtils.java
index 2825c834ffc..b12b03b4def 100644
---
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoUtils.java
+++
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoUtils.java
@@ -378,7 +378,7 @@ class PekkoUtils {
.add(" trust-store = \"" + sslTrustStore + "\"")
.add(" trust-store-password = \"" +
sslTrustStorePassword + "\"")
.add(" trust-store-type = \"" + sslTrustStoreType +
"\"")
- .add(" protocol = " + sslProtocol + "")
+ .add(" protocol = \"" + sslProtocol + "\"")
.add(" enabled-algorithms = " + sslAlgorithms + "")
.add(" random-number-generator = \"\"")
.add(" require-mutual-authentication = on")
diff --git
a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProviderTest.java
b/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProviderTest.java
new file mode 100644
index 00000000000..3282a2f83e6
--- /dev/null
+++
b/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProviderTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.flink.runtime.rpc.pekko;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.runtime.concurrent.pekko.ScalaFutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.bootstrap.Bootstrap;
+import org.apache.flink.shaded.netty4.io.netty.bootstrap.ServerBootstrap;
+import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer;
+import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
+import
org.apache.flink.shaded.netty4.io.netty.channel.MultiThreadIoEventLoopGroup;
+import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioIoHandler;
+import
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioServerSocketChannel;
+import
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslHandler;
+
+import org.apache.pekko.actor.ActorSystem;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLSession;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that {@link CustomSSLEngineProvider} correctly negotiates TLS when
{@link
+ * SecurityOptions#SSL_PROTOCOL} is configured with a comma-separated protocol
list.
+ *
+ * <p>This is a regression test for the fact that {@link
+ * org.apache.pekko.remote.transport.netty.ConfigSSLEngineProvider}, which
{@link
+ * CustomSSLEngineProvider} extends, only supports a single protocol name: it
feeds the configured
+ * string, unsplit, into both {@code SSLContext.getInstance(String)} and {@code
+ * SSLEngine#setEnabledProtocols(String[])}.
+ */
+class CustomSSLEngineProviderTest {
+
+ private static final String KEY_STORE_PATH =
+
checkNotNull(CustomSSLEngineProviderTest.class.getResource("/rpc.keystore")).getFile();
+ private static final String TRUST_STORE_PATH =
+
checkNotNull(CustomSSLEngineProviderTest.class.getResource("/rpc.truststore"))
+ .getFile();
+ private static final String STORE_PASSWORD = "password";
+
+ private static final String TLS_12_CIPHER =
+
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
+ private static final String TLS_13_CIPHER =
"TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384";
+
+ private ActorSystem actorSystem;
+
+ @AfterEach
+ void shutdown() throws Exception {
+ if (actorSystem != null) {
+ ScalaFutureUtils.toJava(actorSystem.terminate()).get(10,
TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ void negotiatesHighestCommonProtocolFromList() throws Exception {
+ SSLSession session = handshake("TLSv1.2,TLSv1.3", TLS_12_CIPHER + ","
+ TLS_13_CIPHER);
+
+ assertThat(session.getProtocol()).isEqualTo("TLSv1.3");
+ }
+
+ @Test
+ void fallsBackToLowerListedProtocolWhenHigherOneHasNoUsableCipher() throws
Exception {
+ SSLSession session = handshake("TLSv1.2,TLSv1.3", TLS_12_CIPHER);
+
+ assertThat(session.getProtocol()).isEqualTo("TLSv1.2");
+ }
+
+ /**
+ * Builds a {@link CustomSSLEngineProvider} from the given protocol list
and ciphers, performs a
+ * real, socket-based TLS handshake between a server and a client engine
it creates, and returns
+ * the client's negotiated session.
+ */
+ private SSLSession handshake(String protocolList, String ciphers) throws
Exception {
+ final Configuration configuration = new Configuration();
+ configuration.set(SecurityOptions.SSL_INTERNAL_ENABLED, true);
+ configuration.set(SecurityOptions.SSL_INTERNAL_KEYSTORE,
KEY_STORE_PATH);
+ configuration.set(SecurityOptions.SSL_INTERNAL_KEYSTORE_PASSWORD,
STORE_PASSWORD);
+ configuration.set(SecurityOptions.SSL_INTERNAL_KEY_PASSWORD,
STORE_PASSWORD);
+ configuration.set(SecurityOptions.SSL_INTERNAL_TRUSTSTORE,
TRUST_STORE_PATH);
+ configuration.set(SecurityOptions.SSL_INTERNAL_TRUSTSTORE_PASSWORD,
STORE_PASSWORD);
+ configuration.set(SecurityOptions.SSL_PROTOCOL, protocolList);
+ configuration.set(SecurityOptions.SSL_ALGORITHMS, ciphers);
+
+ actorSystem =
+ PekkoUtils.createActorSystem(
+ "CustomSSLEngineProviderTest",
+ PekkoUtils.getConfig(configuration, new
HostAndPort("localhost", 0)));
+
+ final CustomSSLEngineProvider provider = new
CustomSSLEngineProvider(actorSystem);
+
+ final EventLoopGroup group = new MultiThreadIoEventLoopGroup(2,
NioIoHandler.newFactory());
+ try {
+ final CompletableFuture<SSLSession> serverSession = new
CompletableFuture<>();
+ final CompletableFuture<SSLSession> clientSession = new
CompletableFuture<>();
+
+ final ServerBootstrap serverBootstrap =
+ new ServerBootstrap()
+ .group(group)
+ .channel(NioServerSocketChannel.class)
+ .childHandler(
+ new ChannelInitializer<Channel>() {
+ @Override
+ protected void initChannel(Channel ch)
{
+ SSLEngine engine =
provider.createServerSSLEngine();
+ SslHandler handler = new
SslHandler(engine);
+ ch.pipeline().addLast(handler);
+ completeWithSession(handler,
serverSession);
+ }
+ });
+ final Channel serverChannel =
serverBootstrap.bind(0).sync().channel();
+ final int port = ((InetSocketAddress)
serverChannel.localAddress()).getPort();
+
+ final Bootstrap clientBootstrap =
+ new Bootstrap()
+ .group(group)
+ .channel(NioSocketChannel.class)
+ .handler(
+ new ChannelInitializer<Channel>() {
+ @Override
+ protected void initChannel(Channel ch)
{
+ SSLEngine engine =
provider.createClientSSLEngine();
+ SslHandler handler = new
SslHandler(engine);
+ ch.pipeline().addLast(handler);
+ completeWithSession(handler,
clientSession);
+ }
+ });
+ clientBootstrap.connect("localhost", port).sync();
+
+ final SSLSession client = clientSession.get(10, TimeUnit.SECONDS);
+ final SSLSession server = serverSession.get(10, TimeUnit.SECONDS);
+ assertThat(client.getProtocol()).isEqualTo(server.getProtocol());
+
assertThat(client.getCipherSuite()).isEqualTo(server.getCipherSuite());
+
+ serverChannel.close().sync();
+ return client;
+ } finally {
+ group.shutdownGracefully(0, 1, TimeUnit.SECONDS).sync();
+ }
+ }
+
+ private static void completeWithSession(
+ SslHandler handler, CompletableFuture<SSLSession> future) {
+ handler.handshakeFuture()
+ .addListener(
+ f -> {
+ if (f.isSuccess()) {
+ future.complete(handler.engine().getSession());
+ } else {
+ future.completeExceptionally(f.cause());
+ }
+ });
+ }
+}
diff --git a/flink-rpc/flink-rpc-akka/src/test/resources/rpc.keystore
b/flink-rpc/flink-rpc-akka/src/test/resources/rpc.keystore
new file mode 100644
index 00000000000..b6cb8ce1522
Binary files /dev/null and
b/flink-rpc/flink-rpc-akka/src/test/resources/rpc.keystore differ
diff --git a/flink-rpc/flink-rpc-akka/src/test/resources/rpc.truststore
b/flink-rpc/flink-rpc-akka/src/test/resources/rpc.truststore
new file mode 100644
index 00000000000..5cd7f52f459
Binary files /dev/null and
b/flink-rpc/flink-rpc-akka/src/test/resources/rpc.truststore differ
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLUtils.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLUtils.java
index c89dd5eef7d..106f5c9af3c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLUtils.java
@@ -377,6 +377,13 @@ public class SSLUtils {
.clientAuth(clientAuth)
.sessionCacheSize(sessionCacheSize)
.sessionTimeout(sessionTimeoutMs / 1000)
+ // Internal SSL uses one shared, mutually-trusted certificate
across every node
+ // (see docs/deployment/security/security-ssl.md), so it must
not be required to
+ // also match each peer's hostname. Explicitly disabled
because Netty enables
+ // endpoint identification by default from 4.2 onward (both
JDK and OpenSSL
+ // providers), which would otherwise break every connection
whose resolved
+ // hostname isn't covered by the shared certificate.
+ .endpointIdentificationAlgorithm(null)
.build();
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/net/SSLUtilsTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/net/SSLUtilsTest.java
index 43e7de101ee..0a5200c988f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/net/SSLUtilsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/net/SSLUtilsTest.java
@@ -23,7 +23,16 @@ import
org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.SecurityOptions;
import org.apache.flink.runtime.io.network.netty.SSLHandlerFactory;
+import org.apache.flink.shaded.netty4.io.netty.bootstrap.Bootstrap;
+import org.apache.flink.shaded.netty4.io.netty.bootstrap.ServerBootstrap;
import org.apache.flink.shaded.netty4.io.netty.buffer.UnpooledByteBufAllocator;
+import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer;
+import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
+import
org.apache.flink.shaded.netty4.io.netty.channel.MultiThreadIoEventLoopGroup;
+import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioIoHandler;
+import
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioServerSocketChannel;
+import
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.flink.shaded.netty4.io.netty.handler.ssl.ClientAuth;
import org.apache.flink.shaded.netty4.io.netty.handler.ssl.JdkSslContext;
import org.apache.flink.shaded.netty4.io.netty.handler.ssl.OpenSsl;
@@ -33,10 +42,14 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
+import javax.net.ssl.SSLHandshakeException;
import javax.net.ssl.SSLServerSocket;
+import javax.net.ssl.SSLSession;
+import javax.net.ssl.SSLSocket;
import java.io.File;
import java.io.InputStream;
+import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.file.Files;
import java.security.KeyStore;
@@ -49,6 +62,8 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import static
org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslProvider.JDK;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -425,6 +440,279 @@ public class SSLUtilsTest {
assertThat(sslHandler.engine().getEnabledCipherSuites()).contains(sslAlgorithms);
}
+ // -------------------- hostname verification (shared certificate)
-------------------
+ //
+ // Internal SSL uses one shared, mutually-trusted certificate across every
node in the
+ // cluster (see docs/content/docs/deployment/security/security-ssl.md:
"Because internal
+ // connections are mutually authenticated with shared certificates, Flink
can skip hostname
+ // verification. This makes container-based setups easier."), e.g. a
Kubernetes deployment
+ // with hundreds of dynamically-scheduled, differently-named pods sharing
one cert. Simulate
+ // that: the client connects to a real "localhost" socket but is told to
verify a peer
+ // identity the shared certificate was never meant to cover; negotiation
must still succeed.
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void testInternalSSLIgnoresPeerHostMismatch(String sslProvider) throws
Exception {
+ Configuration config =
createInternalSslConfigWithKeyAndTrustStores(sslProvider);
+
+ SSLSession session =
+ negotiate(
+ SSLUtils.createInternalServerSSLEngineFactory(config),
+ SSLUtils.createInternalClientSSLEngineFactory(config),
+
"some-taskmanager-42.flink-headless.flink.svc.cluster.local");
+
+ assertThat(session).isNotNull();
+ }
+
+ // -------------------- multi-protocol negotiation -----------------------
+ //
+ // Regression coverage for SecurityOptions.SSL_PROTOCOL accepting a
comma-separated protocol
+ // list: every consumer below is expected to negotiate the highest
protocol both sides support,
+ // and to gracefully fall back rather than fail when the higher protocol
has no usable cipher.
+
+ private static final String TLS_13_CIPHERS =
"TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384";
+
+ private static String tls12Ciphers(String sslProvider) {
+ // openSSL does not support the same set of cipher algorithms as the
JDK provider, see
+ // testCreateSSLEngineFactory above.
+ return sslProvider.equalsIgnoreCase("OPENSSL")
+ ?
"TLS_RSA_WITH_AES_128_GCM_SHA256,TLS_RSA_WITH_AES_256_GCM_SHA384"
+ :
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void testInternalSSLNegotiatesTls13WhenBothSidesSupportIt(String
sslProvider) throws Exception {
+ Configuration config =
createInternalSslConfigWithKeyAndTrustStores(sslProvider);
+ config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.2,TLSv1.3");
+ config.set(
+ SecurityOptions.SSL_ALGORITHMS, tls12Ciphers(sslProvider) +
"," + TLS_13_CIPHERS);
+
+ SSLSession session =
+ negotiate(
+ SSLUtils.createInternalServerSSLEngineFactory(config),
+ SSLUtils.createInternalClientSSLEngineFactory(config));
+
+ assertThat(session.getProtocol()).isEqualTo("TLSv1.3");
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void testInternalSSLFallsBackToTls12WithoutTls13Cipher(String sslProvider)
throws Exception {
+ Configuration config =
createInternalSslConfigWithKeyAndTrustStores(sslProvider);
+ config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.2,TLSv1.3");
+ config.set(SecurityOptions.SSL_ALGORITHMS, tls12Ciphers(sslProvider));
+
+ SSLSession session =
+ negotiate(
+ SSLUtils.createInternalServerSSLEngineFactory(config),
+ SSLUtils.createInternalClientSSLEngineFactory(config));
+
+ assertThat(session.getProtocol()).isEqualTo("TLSv1.2");
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void
testInternalSSLNonContiguousProtocolListWithoutCipherForHighest(String
sslProvider)
+ throws Exception {
+ // TLSv1.3 has no usable cipher here (only TLSv1.2-style ones are
configured), so Netty
+ // drops it from the min/max range calculation entirely before any
OpenSSL-specific
+ // contiguous-range widening happens (ReferenceCountedOpenSslEngine's
+ // explicitDisableTLSv13 handling). That collapses the enabled set to
TLSv1.1 alone on
+ // both providers, which also has no usable cipher (GCM suites are
TLSv1.2+), so the
+ // handshake fails on both providers instead of widening to TLSv1.2.
+ Configuration config =
createInternalSslConfigWithKeyAndTrustStores(sslProvider);
+ config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.1,TLSv1.3");
+ config.set(SecurityOptions.SSL_ALGORITHMS, tls12Ciphers(sslProvider));
+
+ final SSLHandlerFactory server =
SSLUtils.createInternalServerSSLEngineFactory(config);
+ final SSLHandlerFactory client =
SSLUtils.createInternalClientSSLEngineFactory(config);
+
+ assertThatThrownBy(() -> negotiate(server, client))
+ .hasCauseInstanceOf(SSLHandshakeException.class);
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void testRestSSLNegotiatesTls13WhenBothSidesSupportIt(String sslProvider)
throws Exception {
+ Configuration config =
createRestSslConfigWithKeyAndTrustStores(sslProvider);
+ config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.2,TLSv1.3");
+ config.set(
+ SecurityOptions.SSL_ALGORITHMS, tls12Ciphers(sslProvider) +
"," + TLS_13_CIPHERS);
+
+ SSLSession session =
+ negotiate(
+ SSLUtils.createRestServerSSLEngineFactory(config),
+ SSLUtils.createRestClientSSLEngineFactory(config));
+
+ assertThat(session.getProtocol()).isEqualTo("TLSv1.3");
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void testRestSSLFallsBackToTls12WithoutTls13Cipher(String sslProvider)
throws Exception {
+ Configuration config =
createRestSslConfigWithKeyAndTrustStores(sslProvider);
+ config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.2,TLSv1.3");
+ config.set(SecurityOptions.SSL_ALGORITHMS, tls12Ciphers(sslProvider));
+
+ SSLSession session =
+ negotiate(
+ SSLUtils.createRestServerSSLEngineFactory(config),
+ SSLUtils.createRestClientSSLEngineFactory(config));
+
+ assertThat(session.getProtocol()).isEqualTo("TLSv1.2");
+ }
+
+ /**
+ * The Blob server/client socket path ({@link
SSLUtils#createSSLServerSocketFactory}) always
+ * uses the JDK provider regardless of {@link
SecurityOptions#SSL_PROVIDER}, so this is not
+ * parameterized.
+ */
+ @Test
+ void testBlobSocketSSLNegotiatesTls13WhenBothSidesSupportIt() throws
Exception {
+ Configuration config =
createInternalSslConfigWithKeyAndTrustStores("JDK");
+ config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.2,TLSv1.3");
+ config.set(SecurityOptions.SSL_ALGORITHMS, tls12Ciphers("JDK") + "," +
TLS_13_CIPHERS);
+
+ assertThat(negotiateViaSockets(config)).isEqualTo("TLSv1.3");
+ }
+
+ @Test
+ void testBlobSocketSSLFallsBackToTls12WithoutTls13Cipher() throws
Exception {
+ Configuration config =
createInternalSslConfigWithKeyAndTrustStores("JDK");
+ config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.2,TLSv1.3");
+ config.set(SecurityOptions.SSL_ALGORITHMS, tls12Ciphers("JDK"));
+
+ assertThat(negotiateViaSockets(config)).isEqualTo("TLSv1.2");
+ }
+
+ /**
+ * Performs a real, socket-based TLS handshake between the given server
and client {@link
+ * SSLHandlerFactory} and returns the client's negotiated session.
+ */
+ private static SSLSession negotiate(
+ SSLHandlerFactory serverFactory, SSLHandlerFactory clientFactory)
throws Exception {
+ return negotiate(serverFactory, clientFactory, "localhost");
+ }
+
+ /**
+ * Like {@link #negotiate(SSLHandlerFactory, SSLHandlerFactory)}, but the
client engine is told
+ * to verify the given {@code clientPeerIdentity} instead of the real TCP
destination
+ * ("localhost"), so a mismatch between the two can be simulated without
breaking the actual
+ * connection.
+ */
+ private static SSLSession negotiate(
+ SSLHandlerFactory serverFactory,
+ SSLHandlerFactory clientFactory,
+ String clientPeerIdentity)
+ throws Exception {
+ final EventLoopGroup group = new MultiThreadIoEventLoopGroup(2,
NioIoHandler.newFactory());
+ try {
+ final CompletableFuture<SSLSession> serverSession = new
CompletableFuture<>();
+ final CompletableFuture<SSLSession> clientSession = new
CompletableFuture<>();
+
+ final ServerBootstrap serverBootstrap =
+ new ServerBootstrap()
+ .group(group)
+ .channel(NioServerSocketChannel.class)
+ .childHandler(
+ new ChannelInitializer<Channel>() {
+ @Override
+ protected void initChannel(Channel ch)
{
+ SslHandler handler =
+
serverFactory.createNettySSLHandler(
+
UnpooledByteBufAllocator.DEFAULT);
+ ch.pipeline().addLast(handler);
+ completeWithSession(handler,
serverSession);
+ }
+ });
+ final Channel serverChannel =
serverBootstrap.bind(0).sync().channel();
+ final int port = ((InetSocketAddress)
serverChannel.localAddress()).getPort();
+
+ final Bootstrap clientBootstrap =
+ new Bootstrap()
+ .group(group)
+ .channel(NioSocketChannel.class)
+ .handler(
+ new ChannelInitializer<Channel>() {
+ @Override
+ protected void initChannel(Channel ch)
{
+ // Mirrors NettyClient/RestClient:
the client engine
+ // must be created with the peer
host/port, otherwise
+ // TLSv1.3 mutual-auth handshakes
fail with
+ // "certificate_unknown: Hostname
or IP address is
+ // undefined" (TLSv1.2 tolerates
the hostless engine).
+ SslHandler handler =
+
clientFactory.createNettySSLHandler(
+
UnpooledByteBufAllocator.DEFAULT,
+ clientPeerIdentity,
+ port);
+ ch.pipeline().addLast(handler);
+ completeWithSession(handler,
clientSession);
+ }
+ });
+ clientBootstrap.connect("localhost", port).sync();
+
+ final SSLSession client = clientSession.get(10, TimeUnit.SECONDS);
+ final SSLSession server = serverSession.get(10, TimeUnit.SECONDS);
+ assertThat(client.getProtocol()).isEqualTo(server.getProtocol());
+
+ serverChannel.close().sync();
+ return client;
+ } finally {
+ group.shutdownGracefully(0, 1, TimeUnit.SECONDS).sync();
+ }
+ }
+
+ private static void completeWithSession(
+ SslHandler handler, CompletableFuture<SSLSession> future) {
+ handler.handshakeFuture()
+ .addListener(
+ f -> {
+ if (f.isSuccess()) {
+ future.complete(handler.engine().getSession());
+ } else {
+ future.completeExceptionally(f.cause());
+ }
+ });
+ }
+
+ /**
+ * Performs a real handshake through the raw-socket path used by the Blob
server/client, and
+ * returns the negotiated protocol.
+ */
+ private static String negotiateViaSockets(Configuration config) throws
Exception {
+ try (ServerSocket serverSocket =
+
SSLUtils.createSSLServerSocketFactory(config).createServerSocket(0)) {
+ final int port = serverSocket.getLocalPort();
+
+ final CompletableFuture<String> serverProtocol = new
CompletableFuture<>();
+ final Thread serverThread =
+ new Thread(
+ () -> {
+ try (SSLSocket socket = (SSLSocket)
serverSocket.accept()) {
+ socket.startHandshake();
+
serverProtocol.complete(socket.getSession().getProtocol());
+ } catch (Exception e) {
+ serverProtocol.completeExceptionally(e);
+ }
+ });
+ serverThread.start();
+
+ final String clientProtocol;
+ try (SSLSocket client =
+ (SSLSocket)
+ SSLUtils.createSSLClientSocketFactory(config)
+ .createSocket("localhost", port)) {
+ client.startHandshake();
+ clientProtocol = client.getSession().getProtocol();
+ }
+
+ assertThat(serverProtocol.get(10,
TimeUnit.SECONDS)).isEqualTo(clientProtocol);
+ return clientProtocol;
+ }
+ }
+
@ParameterizedTest
@MethodSource("parameters")
void testInvalidFingerprintParsing(String sslProvider) throws Exception {
diff --git a/tools/azure-pipelines/e2e-template.yml
b/tools/azure-pipelines/e2e-template.yml
index 766ff15671e..aff0cfb1620 100644
--- a/tools/azure-pipelines/e2e-template.yml
+++ b/tools/azure-pipelines/e2e-template.yml
@@ -137,8 +137,21 @@ jobs:
echo "Installing required software for netty-tcnative"
sudo apt-get install -y bc libapr1
- wget -r --no-parent -nd --accept=libssl1.0.0_*ubuntu5.*_amd64.deb
http://security.ubuntu.com/ubuntu/pool/main/o/openssl1.0/
- sudo apt install ./libssl1.0.0_*.deb
+
+ # flink-shaded-netty-tcnative-dynamic links against OpenSSL symbols
newer than what this
+ # Ubuntu release ships (system libssl.so.3 lacks OPENSSL_3.2.0). Per
netty's own guidance
+ # for this exact case (http://netty.io/wiki/forked-tomcat-native.html:
"If your Linux
+ # distro's package manager doesn't provide a new enough OpenSSL, you
will need to compile
+ # it by your own and set the LD_LIBRARY_PATH"), fetch a newer prebuilt
libssl/libcrypto
+ # into a private directory instead of replacing the system package, so
the rest of this
+ # job keeps using the OS's own OpenSSL.
+ FLINK_E2E_OPENSSL32_LIB="$(pwd)/openssl32-lib"
+ mkdir -p "${FLINK_E2E_OPENSSL32_LIB}"
+ wget -q -O /tmp/libssl3t64.deb
http://security.ubuntu.com/ubuntu/pool/main/o/openssl/libssl3t64_3.5.5-1ubuntu3.5_amd64.deb
+ dpkg-deb -x /tmp/libssl3t64.deb /tmp/libssl3t64-extracted
+ cp /tmp/libssl3t64-extracted/usr/lib/x86_64-linux-gnu/libssl.so.3
/tmp/libssl3t64-extracted/usr/lib/x86_64-linux-gnu/libcrypto.so.3
"${FLINK_E2E_OPENSSL32_LIB}/"
+ rm -rf /tmp/libssl3t64.deb /tmp/libssl3t64-extracted
+ echo "##vso[task.setvariable
variable=FLINK_E2E_OPENSSL32_LIB]${FLINK_E2E_OPENSSL32_LIB}"
displayName: Prepare E2E run
condition: not(eq(variables['SKIP'], '1'))
# e2e only needs the built binary; MVN_LICENSE_CHECK_ARGS="" skips all QA
(convergence, license, structural) and the DEBUG shade logging, which run in
the license_check job.
