Skip to content

[FLINK-39969] [runtime] Apply configured SSL protocols and algorithms #28532

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
qiuyanjun888 wants to merge 1 commit into
base: master
Choose a base branch
from
+130 −13
Open

[FLINK-39969] [runtime] Apply configured SSL protocols and algorithms #28532

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 83 additions & 13 deletions flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,13 +42,16 @@
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLServerSocket;
import javax.net.ssl.SSLServerSocketFactory;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.file.Files;
import java.security.KeyStore;
import java.security.KeyStoreException;
Expand Down Expand Up @@ -96,7 +99,11 @@ public static SocketFactory createSSLClientSocketFactory(Configuration config)
throw new IllegalConfigurationException("SSL is not enabled");
}

return sslContext.getSocketFactory();
String[] protocols = getEnabledProtocols(config);
String[] cipherSuites = getEnabledCipherSuites(config);

SSLSocketFactory factory = sslContext.getSocketFactory();
return new ConfiguringSSLSocketFactory(factory, protocols, cipherSuites);
}

/** Creates a SSLEngineFactory to be used by internal communication server endpoints. */
Expand Down Expand Up @@ -437,18 +444,15 @@ public static SslContext createRestNettySSLContext(

if (clientMode || clientAuth != ClientAuth.NONE) {
Optional<TrustManagerFactory> tmf = getTrustManagerFactory(config, false);
tmf.map(
// Use specific ciphers and protocols if SSL is configured with self-signed
// certificates (user-supplied truststore)
tm ->
sslContextBuilder
.trustManager(tm)
.protocols(sslProtocols)
.ciphers(ciphers)
.clientAuth(clientAuth));
}

return sslContextBuilder.sslProvider(provider).build();
tmf.ifPresent(sslContextBuilder::trustManager);
}

return sslContextBuilder
.sslProvider(provider)
.protocols(sslProtocols)
.ciphers(ciphers)
.clientAuth(clientAuth)

@JozsefKuti JozsefKuti Jun 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move .clientAuth(clientAuth) a few lines before where clientAuth is enabled - since it is set even for service side as well at the moment.

.build();
}

// ------------------------------------------------------------------------
Expand Down Expand Up @@ -476,6 +480,72 @@ private static String getAndCheckOption(
// Wrappers for socket factories that additionally configure the sockets
// ------------------------------------------------------------------------

private static class ConfiguringSSLSocketFactory extends SSLSocketFactory {

private final SSLSocketFactory sslSocketFactory;
private final String[] protocols;
private final String[] cipherSuites;

ConfiguringSSLSocketFactory(
SSLSocketFactory sslSocketFactory, String[] protocols, String[] cipherSuites) {
this.sslSocketFactory = sslSocketFactory;
this.protocols = protocols;
this.cipherSuites = cipherSuites;
}

@Override
public String[] getDefaultCipherSuites() {
return sslSocketFactory.getDefaultCipherSuites();
}

@Override
public String[] getSupportedCipherSuites() {
return sslSocketFactory.getSupportedCipherSuites();
}

@Override
public Socket createSocket() throws IOException {
return configureSocket(sslSocketFactory.createSocket());
}

@Override
public Socket createSocket(Socket socket, String host, int port, boolean autoClose)
throws IOException {
return configureSocket(sslSocketFactory.createSocket(socket, host, port, autoClose));
}

@Override
public Socket createSocket(String host, int port) throws IOException {
return configureSocket(sslSocketFactory.createSocket(host, port));
}

@Override
public Socket createSocket(String host, int port, InetAddress localHost, int localPort)
throws IOException {
return configureSocket(sslSocketFactory.createSocket(host, port, localHost, localPort));
}

@Override
public Socket createSocket(InetAddress host, int port) throws IOException {
return configureSocket(sslSocketFactory.createSocket(host, port));
}

@Override
public Socket createSocket(
InetAddress address, int port, InetAddress localAddress, int localPort)
throws IOException {
return configureSocket(
sslSocketFactory.createSocket(address, port, localAddress, localPort));
}

private Socket configureSocket(Socket socket) {
SSLSocket sslSocket = (SSLSocket) socket;
sslSocket.setEnabledProtocols(protocols);
sslSocket.setEnabledCipherSuites(cipherSuites);
return sslSocket;
}
}

private static class ConfiguringSSLServerSocketFactory extends ServerSocketFactory {

private final SSLServerSocketFactory sslServerSocketFactory;
Expand Down
47 changes: 47 additions & 0 deletions flink-runtime/src/test/java/org/apache/flink/runtime/net/SSLUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,11 +33,14 @@
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;

import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLServerSocket;
import javax.net.ssl.SSLSocket;

import java.io.File;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.file.Files;
import java.security.KeyStore;
import java.security.KeyStoreException;
Expand Down Expand Up @@ -175,6 +178,27 @@ void testRESTSSLConfigCipherAlgorithms(String sslProvider) throws Exception {
assertThat(cipherSuites).containsExactlyInAnyOrder(testSSLAlgorithms.split(","));
}

@Test
void testRestServerAppliesConfiguredProtocolsAndCipherSuites() throws Exception {
final Configuration config = createRestSslConfigWithKeyStore("JDK");
config.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.2");
config.set(
SecurityOptions.SSL_ALGORITHMS,
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384");

final JdkSslContext nettySSLContext =
(JdkSslContext)
SSLUtils.createRestNettySSLContext(config, false, ClientAuth.NONE, JDK);
final SSLEngine sslEngine =
checkNotNull(nettySSLContext).newEngine(UnpooledByteBufAllocator.DEFAULT);

assertThat(sslEngine.getEnabledProtocols()).containsExactly("TLSv1.2");
assertThat(sslEngine.getEnabledCipherSuites())
.containsExactlyInAnyOrder(
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
"TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384");
}

// ------------------------ server --------------------------

/** Tests that REST Server SSL Engine is created given a valid SSL configuration. */
Expand Down Expand Up @@ -387,6 +411,29 @@ void testSetSSLVersionAndCipherSuitesForSSLServerSocket(String sslProvider) thro
}
}

@ParameterizedTest
@MethodSource("parameters")
void testSetSSLVersionAndCipherSuitesForSSLClientSocket(String sslProvider) throws Exception {
final Configuration clientConfig =
createInternalSslConfigWithKeyAndTrustStores(sslProvider);

clientConfig.set(SecurityOptions.SSL_PROTOCOL, "TLSv1.1");
clientConfig.set(
SecurityOptions.SSL_ALGORITHMS,
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384");

try (Socket socket = SSLUtils.createSSLClientSocketFactory(clientConfig).createSocket()) {
assertThat(socket).isInstanceOf(SSLSocket.class);
final SSLSocket sslSocket = (SSLSocket) socket;

assertThat(sslSocket.getEnabledProtocols()).containsExactly("TLSv1.1");
assertThat(sslSocket.getEnabledCipherSuites())
.containsExactlyInAnyOrder(
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
"TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384");
}
}

/** Tests that {@link SSLHandlerFactory} is created correctly. */
@ParameterizedTest
@MethodSource("parameters")
Expand Down