You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/05/09 22:24:40 UTC
[2/2] flink git commit: [FLINK-9261] [security] Fix SSL support for
REST API and Web UI.
[FLINK-9261] [security] Fix SSL support for REST API and Web UI.
- Remove wrong reuse of SSLEngine instances. SSLEngine must be re-created for
every SocketChannel initialization.
- Add ChunkedWriteHandler to REST server pipeline because StaticFileServerHandler
relies on it.
- Add integration tests to verify that SSL can be enabled.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3afa5eb3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3afa5eb3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3afa5eb3
Branch: refs/heads/master
Commit: 3afa5eb3c47158086ab29012a835f96682a85d34
Parents: 7c87c1a
Author: gyao <ga...@data-artisans.com>
Authored: Wed May 9 13:30:23 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu May 10 00:22:59 2018 +0200
----------------------------------------------------------------------
.../flink/runtime/net/SSLEngineFactory.java | 57 +++++++++
.../org/apache/flink/runtime/net/SSLUtils.java | 72 +++++++++--
.../apache/flink/runtime/rest/RestClient.java | 11 +-
.../runtime/rest/RestClientConfiguration.java | 29 ++---
.../flink/runtime/rest/RestServerEndpoint.java | 15 ++-
.../rest/RestServerEndpointConfiguration.java | 33 +++---
.../apache/flink/runtime/net/SSLUtilsTest.java | 28 +++++
.../runtime/rest/RestServerEndpointITCase.java | 118 +++++++++++++++----
8 files changed, 283 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLEngineFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLEngineFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLEngineFactory.java
new file mode 100644
index 0000000..7aca60c
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLEngineFactory.java
@@ -0,0 +1,57 @@
+/*
+ * 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.net;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Creates and configures {@link SSLEngine} instances.
+ */
+public class SSLEngineFactory {
+
+ private final SSLContext sslContext;
+
+ private final String[] enabledProtocols;
+
+ private final String[] enabledCipherSuites;
+
+ private final boolean clientMode;
+
+ public SSLEngineFactory(
+ final SSLContext sslContext,
+ final String[] enabledProtocols,
+ final String[] enabledCipherSuites,
+ final boolean clientMode) {
+ this.sslContext = requireNonNull(sslContext, "sslContext must not be null");
+ this.enabledProtocols = requireNonNull(enabledProtocols, "enabledProtocols must not be null");
+ this.enabledCipherSuites = requireNonNull(enabledCipherSuites, "cipherSuites must not be null");
+ this.clientMode = clientMode;
+ }
+
+ public SSLEngine createSSLEngine() {
+ final SSLEngine sslEngine = sslContext.createSSLEngine();
+ sslEngine.setEnabledProtocols(enabledProtocols);
+ sslEngine.setEnabledCipherSuites(enabledCipherSuites);
+ sslEngine.setUseClientMode(clientMode);
+ return sslEngine;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/main/java/org/apache/flink/runtime/net/SSLUtils.java
----------------------------------------------------------------------
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 663d221..b574d30 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
@@ -25,6 +25,7 @@ import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
@@ -38,6 +39,9 @@ import java.net.ServerSocket;
import java.security.KeyStore;
import java.util.Arrays;
+import static java.util.Objects.requireNonNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
/**
* Common utilities to manage SSL transport settings.
*/
@@ -82,15 +86,61 @@ public class SSLUtils {
}
/**
+ * Creates a {@link SSLEngineFactory} to be used by the Server.
+ *
+ * @param config The application configuration.
+ */
+ public static SSLEngineFactory createServerSSLEngineFactory(final Configuration config) throws Exception {
+ return createSSLEngineFactory(config, false);
+ }
+
+ /**
+ * Creates a {@link SSLEngineFactory} to be used by the Client.
+ * @param config The application configuration.
+ */
+ public static SSLEngineFactory createClientSSLEngineFactory(final Configuration config) throws Exception {
+ return createSSLEngineFactory(config, true);
+ }
+
+ private static SSLEngineFactory createSSLEngineFactory(
+ final Configuration config,
+ final boolean clientMode) throws Exception {
+
+ final SSLContext sslContext = clientMode ?
+ createSSLClientContext(config) :
+ createSSLServerContext(config);
+
+ checkState(sslContext != null, "%s it not enabled", SecurityOptions.SSL_ENABLED.key());
+
+ return new SSLEngineFactory(
+ sslContext,
+ getEnabledProtocols(config),
+ getEnabledCipherSuites(config),
+ clientMode);
+ }
+
+ /**
* Sets SSL version and cipher suites for SSLEngine.
- * @param engine
- * SSLEngine to be handled
- * @param config
- * The application configuration
+ *
+ * @param engine SSLEngine to be handled
+ * @param config The application configuration
+ * @deprecated Use {@link #createClientSSLEngineFactory(Configuration)} or
+ * {@link #createServerSSLEngineFactory(Configuration)}.
*/
+ @Deprecated
public static void setSSLVerAndCipherSuites(SSLEngine engine, Configuration config) {
- engine.setEnabledProtocols(config.getString(SecurityOptions.SSL_PROTOCOL).split(","));
- engine.setEnabledCipherSuites(config.getString(SecurityOptions.SSL_ALGORITHMS).split(","));
+ engine.setEnabledProtocols(getEnabledProtocols(config));
+ engine.setEnabledCipherSuites(getEnabledCipherSuites(config));
+ }
+
+ private static String[] getEnabledProtocols(final Configuration config) {
+ requireNonNull(config, "config must not be null");
+ return config.getString(SecurityOptions.SSL_PROTOCOL).split(",");
+ }
+
+ private static String[] getEnabledCipherSuites(final Configuration config) {
+ requireNonNull(config, "config must not be null");
+ return config.getString(SecurityOptions.SSL_ALGORITHMS).split(",");
}
/**
@@ -122,6 +172,7 @@ public class SSLUtils {
* @throws Exception
* Thrown if there is any misconfiguration
*/
+ @Nullable
public static SSLContext createSSLClientContext(Configuration sslConfig) throws Exception {
Preconditions.checkNotNull(sslConfig);
@@ -170,6 +221,7 @@ public class SSLUtils {
* @throws Exception
* Thrown if there is any misconfiguration
*/
+ @Nullable
public static SSLContext createSSLServerContext(Configuration sslConfig) throws Exception {
Preconditions.checkNotNull(sslConfig);
@@ -191,14 +243,8 @@ public class SSLUtils {
Preconditions.checkNotNull(certPassword, SecurityOptions.SSL_KEY_PASSWORD.key() + " was not configured.");
KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
- FileInputStream keyStoreFile = null;
- try {
- keyStoreFile = new FileInputStream(new File(keystoreFilePath));
+ try (FileInputStream keyStoreFile = new FileInputStream(new File(keystoreFilePath))) {
ks.load(keyStoreFile, keystorePassword.toCharArray());
- } finally {
- if (keyStoreFile != null) {
- keyStoreFile.close();
- }
}
// Set up key manager factory to use the server key store
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
index 8f7dfed..2e812d3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.rest;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.runtime.net.SSLEngineFactory;
import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rest.messages.MessageParameters;
@@ -66,8 +67,6 @@ import org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultThreadFact
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.net.ssl.SSLEngine;
-
import java.io.IOException;
import java.io.InputStream;
import java.io.StringWriter;
@@ -93,13 +92,13 @@ public class RestClient {
Preconditions.checkNotNull(configuration);
this.executor = Preconditions.checkNotNull(executor);
- SSLEngine sslEngine = configuration.getSslEngine();
+ final SSLEngineFactory sslEngineFactory = configuration.getSslEngineFactory();
ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
@Override
- protected void initChannel(SocketChannel socketChannel) throws Exception {
+ protected void initChannel(SocketChannel socketChannel) {
// SSL should be the first handler in the pipeline
- if (sslEngine != null) {
- socketChannel.pipeline().addLast("ssl", new SslHandler(sslEngine));
+ if (sslEngineFactory != null) {
+ socketChannel.pipeline().addLast("ssl", new SslHandler(sslEngineFactory.createSSLEngine()));
}
socketChannel.pipeline()
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClientConfiguration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClientConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClientConfiguration.java
index 0e98e8f..b1591be 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClientConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClientConfiguration.java
@@ -21,12 +21,12 @@ package org.apache.flink.runtime.rest;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.runtime.net.SSLEngineFactory;
import org.apache.flink.runtime.net.SSLUtils;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
-import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -37,18 +37,18 @@ import static org.apache.flink.util.Preconditions.checkArgument;
public final class RestClientConfiguration {
@Nullable
- private final SSLEngine sslEngine;
+ private final SSLEngineFactory sslEngineFactory;
private final long connectionTimeout;
private final int maxContentLength;
private RestClientConfiguration(
- @Nullable final SSLEngine sslEngine,
+ @Nullable final SSLEngineFactory sslEngineFactory,
final long connectionTimeout,
final int maxContentLength) {
checkArgument(maxContentLength > 0, "maxContentLength must be positive, was: %d", maxContentLength);
- this.sslEngine = sslEngine;
+ this.sslEngineFactory = sslEngineFactory;
this.connectionTimeout = connectionTimeout;
this.maxContentLength = maxContentLength;
}
@@ -58,9 +58,9 @@ public final class RestClientConfiguration {
*
* @return SSLEngine that the REST client endpoint should use, or null if SSL was disabled
*/
-
- public SSLEngine getSslEngine() {
- return sslEngine;
+ @Nullable
+ public SSLEngineFactory getSslEngineFactory() {
+ return sslEngineFactory;
}
/**
@@ -90,25 +90,22 @@ public final class RestClientConfiguration {
public static RestClientConfiguration fromConfiguration(Configuration config) throws ConfigurationException {
Preconditions.checkNotNull(config);
- SSLEngine sslEngine = null;
- boolean enableSSL = config.getBoolean(SecurityOptions.SSL_ENABLED);
+ final SSLEngineFactory sslEngineFactory;
+ final boolean enableSSL = config.getBoolean(SecurityOptions.SSL_ENABLED);
if (enableSSL) {
try {
- SSLContext sslContext = SSLUtils.createSSLServerContext(config);
- if (sslContext != null) {
- sslEngine = sslContext.createSSLEngine();
- SSLUtils.setSSLVerAndCipherSuites(sslEngine, config);
- sslEngine.setUseClientMode(false);
- }
+ sslEngineFactory = SSLUtils.createClientSSLEngineFactory(config);
} catch (Exception e) {
throw new ConfigurationException("Failed to initialize SSLContext for the web frontend", e);
}
+ } else {
+ sslEngineFactory = null;
}
final long connectionTimeout = config.getLong(RestOptions.CONNECTION_TIMEOUT);
int maxContentLength = config.getInteger(RestOptions.CLIENT_MAX_CONTENT_LENGTH);
- return new RestClientConfiguration(sslEngine, connectionTimeout, maxContentLength);
+ return new RestClientConfiguration(sslEngineFactory, connectionTimeout, maxContentLength);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
index 15fbbb2..01d1043 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.net.SSLEngineFactory;
import org.apache.flink.runtime.rest.handler.PipelineErrorHandler;
import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
import org.apache.flink.runtime.rest.handler.RouterHandler;
@@ -41,13 +42,13 @@ import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpServerCode
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.router.Handler;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.router.Router;
import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslHandler;
+import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedWriteHandler;
import org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
-import javax.net.ssl.SSLEngine;
import java.io.IOException;
import java.io.Serializable;
@@ -76,7 +77,8 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
private final String restAddress;
private final String restBindAddress;
private final int restBindPort;
- private final SSLEngine sslEngine;
+ @Nullable
+ private final SSLEngineFactory sslEngineFactory;
private final int maxContentLength;
protected final Path uploadDir;
@@ -96,7 +98,7 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
this.restAddress = configuration.getRestAddress();
this.restBindAddress = configuration.getRestBindAddress();
this.restBindPort = configuration.getRestBindPort();
- this.sslEngine = configuration.getSslEngine();
+ this.sslEngineFactory = configuration.getSslEngineFactory();
this.uploadDir = configuration.getUploadDir();
createUploadDir(uploadDir, log);
@@ -155,14 +157,15 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
Handler handler = new RouterHandler(router, responseHeaders);
// SSL should be the first handler in the pipeline
- if (sslEngine != null) {
- ch.pipeline().addLast("ssl", new SslHandler(sslEngine));
+ if (sslEngineFactory != null) {
+ ch.pipeline().addLast("ssl", new SslHandler(sslEngineFactory.createSSLEngine()));
}
ch.pipeline()
.addLast(new HttpServerCodec())
.addLast(new FileUploadHandler(uploadDir))
.addLast(new FlinkHttpObjectAggregator(maxContentLength, responseHeaders))
+ .addLast(new ChunkedWriteHandler())
.addLast(handler.name(), handler)
.addLast(new PipelineErrorHandler(log, responseHeaders));
}
@@ -198,7 +201,7 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
final String protocol;
- if (sslEngine != null) {
+ if (sslEngineFactory != null) {
protocol = "https://";
} else {
protocol = "http://";
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpointConfiguration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpointConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpointConfiguration.java
index 542a937..875230f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpointConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpointConfiguration.java
@@ -22,6 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.SecurityOptions;
import org.apache.flink.configuration.WebOptions;
+import org.apache.flink.runtime.net.SSLEngineFactory;
import org.apache.flink.runtime.net.SSLUtils;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.Preconditions;
@@ -29,7 +30,6 @@ import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders;
import javax.annotation.Nullable;
-import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import java.nio.file.Path;
@@ -52,7 +52,7 @@ public final class RestServerEndpointConfiguration {
private final int restBindPort;
@Nullable
- private final SSLEngine sslEngine;
+ private final SSLEngineFactory sslEngineFactory;
private final Path uploadDir;
@@ -64,9 +64,10 @@ public final class RestServerEndpointConfiguration {
final String restAddress,
@Nullable String restBindAddress,
int restBindPort,
- @Nullable SSLEngine sslEngine,
+ @Nullable SSLEngineFactory sslEngineFactory,
final Path uploadDir,
- final int maxContentLength, final Map<String, String> responseHeaders) {
+ final int maxContentLength,
+ final Map<String, String> responseHeaders) {
Preconditions.checkArgument(0 <= restBindPort && restBindPort < 65536, "The bing rest port " + restBindPort + " is out of range (0, 65536[");
Preconditions.checkArgument(maxContentLength > 0, "maxContentLength must be positive, was: %d", maxContentLength);
@@ -74,7 +75,7 @@ public final class RestServerEndpointConfiguration {
this.restAddress = requireNonNull(restAddress);
this.restBindAddress = restBindAddress;
this.restBindPort = restBindPort;
- this.sslEngine = sslEngine;
+ this.sslEngineFactory = sslEngineFactory;
this.uploadDir = requireNonNull(uploadDir);
this.maxContentLength = maxContentLength;
this.responseHeaders = Collections.unmodifiableMap(requireNonNull(responseHeaders));
@@ -110,8 +111,9 @@ public final class RestServerEndpointConfiguration {
*
* @return SSLEngine that the REST server endpoint should use, or null if SSL was disabled
*/
- public SSLEngine getSslEngine() {
- return sslEngine;
+ @Nullable
+ public SSLEngineFactory getSslEngineFactory() {
+ return sslEngineFactory;
}
/**
@@ -154,19 +156,16 @@ public final class RestServerEndpointConfiguration {
final String restBindAddress = config.getString(RestOptions.BIND_ADDRESS);
final int port = config.getInteger(RestOptions.PORT);
- SSLEngine sslEngine = null;
- final boolean enableSSL = config.getBoolean(SecurityOptions.SSL_ENABLED);
+ final SSLEngineFactory sslEngineFactory;
+ final boolean enableSSL = config.getBoolean(SecurityOptions.SSL_ENABLED) && config.getBoolean(WebOptions.SSL_ENABLED);
if (enableSSL) {
try {
- SSLContext sslContext = SSLUtils.createSSLServerContext(config);
- if (sslContext != null) {
- sslEngine = sslContext.createSSLEngine();
- SSLUtils.setSSLVerAndCipherSuites(sslEngine, config);
- sslEngine.setUseClientMode(false);
- }
+ sslEngineFactory = SSLUtils.createServerSSLEngineFactory(config);
} catch (Exception e) {
- throw new ConfigurationException("Failed to initialize SSLContext for REST server endpoint.", e);
+ throw new ConfigurationException("Failed to initialize SSLEngineFactory for REST server endpoint.", e);
}
+ } else {
+ sslEngineFactory = null;
}
final Path uploadDir = Paths.get(
@@ -183,7 +182,7 @@ public final class RestServerEndpointConfiguration {
restAddress,
restBindAddress,
port,
- sslEngine,
+ sslEngineFactory,
uploadDir,
maxContentLength,
responseHeaders);
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/test/java/org/apache/flink/runtime/net/SSLUtilsTest.java
----------------------------------------------------------------------
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 87d0ccc..1bf3173 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
@@ -29,6 +29,11 @@ import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLServerSocket;
import java.net.ServerSocket;
+import java.util.Arrays;
+
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertThat;
/**
* Tests for the {@link SSLUtils}.
@@ -227,4 +232,27 @@ public class SSLUtilsTest {
Assert.assertTrue(algorithms[1].equals("TLS_DHE_RSA_WITH_AES_128_CBC_SHA") || algorithms[1].equals("TLS_DHE_RSA_WITH_AES_128_CBC_SHA256"));
}
+ /**
+ * Tests that {@link SSLEngineFactory} is created correctly.
+ */
+ @Test
+ public void testCreateSSLEngineFactory() throws Exception {
+ Configuration serverConfig = new Configuration();
+ serverConfig.setBoolean(SecurityOptions.SSL_ENABLED, true);
+ serverConfig.setString(SecurityOptions.SSL_KEYSTORE, "src/test/resources/local127.keystore");
+ serverConfig.setString(SecurityOptions.SSL_KEYSTORE_PASSWORD, "password");
+ serverConfig.setString(SecurityOptions.SSL_KEY_PASSWORD, "password");
+ serverConfig.setString(SecurityOptions.SSL_PROTOCOL, "TLSv1");
+ serverConfig.setString(SecurityOptions.SSL_ALGORITHMS, "TLS_DHE_RSA_WITH_AES_128_CBC_SHA,TLS_DHE_RSA_WITH_AES_128_CBC_SHA256");
+
+ final SSLEngineFactory serverSSLEngineFactory = SSLUtils.createServerSSLEngineFactory(serverConfig);
+ final SSLEngine sslEngine = serverSSLEngineFactory.createSSLEngine();
+
+ assertThat(
+ Arrays.asList(sslEngine.getEnabledProtocols()),
+ contains("TLSv1"));
+ assertThat(
+ Arrays.asList(sslEngine.getEnabledCipherSuites()),
+ containsInAnyOrder("TLS_DHE_RSA_WITH_AES_128_CBC_SHA", "TLS_DHE_RSA_WITH_AES_128_CBC_SHA256"));
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3afa5eb3/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
index 09e36de..377de08 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
@@ -23,11 +23,15 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.SecurityOptions;
import org.apache.flink.configuration.WebOptions;
+import org.apache.flink.runtime.net.SSLUtils;
import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
+import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
+import org.apache.flink.runtime.rest.handler.legacy.files.WebContentHandlerSpecification;
import org.apache.flink.runtime.rest.messages.ConversionException;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
@@ -45,7 +49,6 @@ import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import org.apache.flink.testutils.category.New;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
@@ -54,15 +57,20 @@ import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.TooLongFrameException;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.commons.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import javax.annotation.Nonnull;
+import javax.net.ssl.SSLContext;
+import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
@@ -77,10 +85,10 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import static java.util.Objects.requireNonNull;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
@@ -95,6 +103,7 @@ import static org.mockito.Mockito.when;
* IT cases for {@link RestClient} and {@link RestServerEndpoint}.
*/
@Category(New.class)
+@RunWith(Parameterized.class)
public class RestServerEndpointITCase extends TestLogger {
private static final JobID PATH_JOB_ID = new JobID();
@@ -111,14 +120,56 @@ public class RestServerEndpointITCase extends TestLogger {
private TestUploadHandler testUploadHandler;
private InetSocketAddress serverAddress;
- @Before
- public void setup() throws Exception {
- Configuration config = new Configuration();
+ private final Configuration config;
+ private SSLContext defaultSSLContext;
+
+ public RestServerEndpointITCase(final Configuration config) {
+ this.config = requireNonNull(config);
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> data() {
+ final Configuration config = getBaseConfig();
+
+ final ClassLoader classLoader = ClassLoader.getSystemClassLoader();
+ final String truststorePath = new File(classLoader
+ .getResource("local127.truststore")
+ .getFile()).getAbsolutePath();
+ final String keystorePath = new File(classLoader
+ .getResource("local127.keystore")
+ .getFile()).getAbsolutePath();
+
+ final Configuration sslConfig = new Configuration(config);
+ sslConfig.setBoolean(SecurityOptions.SSL_ENABLED, true);
+ sslConfig.setString(SecurityOptions.SSL_TRUSTSTORE, truststorePath);
+ sslConfig.setString(SecurityOptions.SSL_TRUSTSTORE_PASSWORD, "password");
+ sslConfig.setString(SecurityOptions.SSL_KEYSTORE, keystorePath);
+ sslConfig.setString(SecurityOptions.SSL_KEYSTORE_PASSWORD, "password");
+ sslConfig.setString(SecurityOptions.SSL_KEY_PASSWORD, "password");
+
+ return Arrays.asList(new Object[][]{
+ {config}, {sslConfig}
+ });
+ }
+
+ private static Configuration getBaseConfig() {
+ final Configuration config = new Configuration();
config.setInteger(RestOptions.PORT, 0);
config.setString(RestOptions.ADDRESS, "localhost");
- config.setString(WebOptions.UPLOAD_DIR, temporaryFolder.newFolder().getCanonicalPath());
config.setInteger(RestOptions.SERVER_MAX_CONTENT_LENGTH, TEST_REST_MAX_CONTENT_LENGTH);
config.setInteger(RestOptions.CLIENT_MAX_CONTENT_LENGTH, TEST_REST_MAX_CONTENT_LENGTH);
+ return config;
+ }
+
+ @Before
+ public void setup() throws Exception {
+ config.setString(WebOptions.UPLOAD_DIR, temporaryFolder.newFolder().getCanonicalPath());
+
+ defaultSSLContext = SSLContext.getDefault();
+ final SSLContext sslClientContext = SSLUtils.createSSLClientContext(config);
+ if (sslClientContext != null) {
+ SSLContext.setDefault(sslClientContext);
+ }
RestServerEndpointConfiguration serverConfig = RestServerEndpointConfiguration.fromConfiguration(config);
RestClientConfiguration clientConfig = RestClientConfiguration.fromConfiguration(config);
@@ -126,8 +177,9 @@ public class RestServerEndpointITCase extends TestLogger {
final String restAddress = "http://localhost:1234";
RestfulGateway mockRestfulGateway = mock(RestfulGateway.class);
when(mockRestfulGateway.requestRestAddress(any(Time.class))).thenReturn(CompletableFuture.completedFuture(restAddress));
- GatewayRetriever<RestfulGateway> mockGatewayRetriever = mock(GatewayRetriever.class);
- when(mockGatewayRetriever.getNow()).thenReturn(Optional.of(mockRestfulGateway));
+
+ final GatewayRetriever<RestfulGateway> mockGatewayRetriever = () ->
+ CompletableFuture.completedFuture(mockRestfulGateway);
TestHandler testHandler = new TestHandler(
CompletableFuture.completedFuture(restAddress),
@@ -139,7 +191,18 @@ public class RestServerEndpointITCase extends TestLogger {
mockGatewayRetriever,
RpcUtils.INF_TIMEOUT);
- serverEndpoint = new TestRestServerEndpoint(serverConfig, testHandler, testUploadHandler);
+ final StaticFileServerHandler<RestfulGateway> staticFileServerHandler = new StaticFileServerHandler<>(
+ mockGatewayRetriever,
+ CompletableFuture.completedFuture(restAddress),
+ RpcUtils.INF_TIMEOUT,
+ temporaryFolder.getRoot());
+
+ final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = Arrays.asList(
+ Tuple2.of(new TestHeaders(), testHandler),
+ Tuple2.of(TestUploadHeaders.INSTANCE, testUploadHandler),
+ Tuple2.of(WebContentHandlerSpecification.getInstance(), staticFileServerHandler));
+
+ serverEndpoint = new TestRestServerEndpoint(serverConfig, handlers);
restClient = new TestRestClient(clientConfig);
serverEndpoint.start();
@@ -148,6 +211,8 @@ public class RestServerEndpointITCase extends TestLogger {
@After
public void teardown() throws Exception {
+ SSLContext.setDefault(defaultSSLContext);
+
if (restClient != null) {
restClient.shutdown(timeout);
restClient = null;
@@ -334,6 +399,22 @@ public class RestServerEndpointITCase extends TestLogger {
assertEquals(400, connection.getResponseCode());
}
+ /**
+ * Tests that files can be served with the {@link StaticFileServerHandler}.
+ */
+ @Test
+ public void testStaticFileServerHandler() throws Exception {
+ final File file = temporaryFolder.newFile();
+ Files.write(file.toPath(), Collections.singletonList("foobar"));
+
+ final URL url = new URL(serverEndpoint.getRestBaseUrl() + "/" + file.getName());
+ final HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+ connection.setRequestMethod("GET");
+ final String fileContents = IOUtils.toString(connection.getInputStream());
+
+ assertEquals("foobar", fileContents.trim());
+ }
+
private HttpURLConnection openHttpConnectionForUpload(final String boundary) throws IOException {
final HttpURLConnection connection =
(HttpURLConnection) new URL(serverEndpoint.getRestBaseUrl() + "/upload").openConnection();
@@ -356,29 +437,22 @@ public class RestServerEndpointITCase extends TestLogger {
private static class TestRestServerEndpoint extends RestServerEndpoint {
- private final TestHandler testHandler;
-
- private final TestUploadHandler testUploadHandler;
+ private final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers;
TestRestServerEndpoint(
- RestServerEndpointConfiguration configuration,
- TestHandler testHandler,
- TestUploadHandler testUploadHandler) throws IOException {
+ RestServerEndpointConfiguration configuration,
+ List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers) throws IOException {
super(configuration);
-
- this.testHandler = Preconditions.checkNotNull(testHandler);
- this.testUploadHandler = Preconditions.checkNotNull(testUploadHandler);
+ this.handlers = requireNonNull(handlers);
}
@Override
protected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(CompletableFuture<String> restAddressFuture) {
- return Arrays.asList(
- Tuple2.of(new TestHeaders(), testHandler),
- Tuple2.of(TestUploadHeaders.INSTANCE, testUploadHandler));
+ return handlers;
}
@Override
- protected void startInternal() throws Exception {}
+ protected void startInternal() {}
}
private static class TestHandler extends AbstractRestHandler<RestfulGateway, TestRequest, TestResponse, TestParameters> {