[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/2518


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r83553892
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientSslTest.java
 ---
@@ -0,0 +1,284 @@
+/*
+ * 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.blob;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.security.MessageDigest;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+import org.codehaus.jackson.map.DeserializerFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * This class contains unit tests for the {@link BlobClient} with ssl 
enabled.
+ */
+public class BlobClientSslTest {
+
+   /** The buffer size used during the tests in bytes. */
+   private static final int TEST_BUFFER_SIZE = 17 * 1000;
+
+   /** The instance of the BLOB server used during the tests. */
+   private static BlobServer BLOB_SERVER;
+
+   /** The blob service client configuration */
+   private static Configuration clientConfig;
+   /**
+* Starts the BLOB server.
+*/
+   @BeforeClass
+   public static void startServer() {
+   try {
+   Configuration config = new Configuration();
+   config.setBoolean(ConfigConstants.SECURITY_SSL_ENABLED, 
true);
+   config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE, 
"src/test/resources/local127.keystore");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE_PASSWORD, "password");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEY_PASSWORD, "password");
--- End diff --

Added few more testcases for failure scenarios


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82413936
  
--- Diff: 
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/files/StaticFileServerHandler.java
 ---
@@ -304,8 +314,15 @@ private void respondAsLeader(ChannelHandlerContext 
ctx, HttpRequest request, Str
ctx.write(response);
 
// write the content.
-   ctx.write(new DefaultFileRegion(raf.getChannel(), 0, 
fileLength), ctx.newProgressivePromise());
-   ChannelFuture lastContentFuture = 
ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+   ChannelFuture lastContentFuture;
+   if (ctx.pipeline().get(SslHandler.class) == null) {
+   ctx.write(new DefaultFileRegion(raf.getChannel(), 0, 
fileLength), ctx.newProgressivePromise());
+   lastContentFuture = 
ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+   } else {
+   lastContentFuture = ctx.writeAndFlush(new 
HttpChunkedInput(new ChunkedFile(raf, 0, fileLength, 8192)),
+   ctx.newProgressivePromise());
+   // HttpChunkedInput will write the end marker 
(LastHttpContent) for us.
--- End diff --

This is based on the Netty example code - 
https://netty.io/4.0/xref/io/netty/example/http/file/HttpStaticFileServerHandler.html

The initial flink code somehow did not include these checks. Cannot do zero 
copy file transfer with SSL enabled and the existing stream handling code hangs 
without any data transfer. Hence had to introduce this.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82412954
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientSslTest.java
 ---
@@ -0,0 +1,284 @@
+/*
+ * 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.blob;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.security.MessageDigest;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+import org.codehaus.jackson.map.DeserializerFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * This class contains unit tests for the {@link BlobClient} with ssl 
enabled.
+ */
+public class BlobClientSslTest {
+
+   /** The buffer size used during the tests in bytes. */
+   private static final int TEST_BUFFER_SIZE = 17 * 1000;
+
+   /** The instance of the BLOB server used during the tests. */
+   private static BlobServer BLOB_SERVER;
+
+   /** The blob service client configuration */
+   private static Configuration clientConfig;
+   /**
+* Starts the BLOB server.
+*/
+   @BeforeClass
+   public static void startServer() {
+   try {
+   Configuration config = new Configuration();
+   config.setBoolean(ConfigConstants.SECURITY_SSL_ENABLED, 
true);
+   config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE, 
"src/test/resources/local127.keystore");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE_PASSWORD, "password");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEY_PASSWORD, "password");
--- End diff --

Sure, I'll add another test case for that


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82412891
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -544,6 +552,11 @@
public static final String JOB_MANAGER_WEB_PORT_KEY = 
"jobmanager.web.port";
 
/**
+* Config parameter to override SSL support for the JobManager Web UI
+*/
+   public static final String JOB_MANAGER_WEB_SSL_ENABLED = 
"jobmanager.web.ssl.enabled";
--- End diff --

please see previous comment related to this


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82412822
  
--- Diff: docs/setup/security-ssl.md ---
@@ -0,0 +1,139 @@
+---
+title: "SSL Setup"
+nav-parent_id: setup
+nav-pos: 8
+---
+
+
+This page provides instructions on how to enable SSL for the network 
communication between different flink components.
+
+## SSL Configuration
+
+SSL can be enabled for all network communication between flink components. 
SSL keystores and truststore has to be deployed on each flink node and 
configured (conf/flink-conf.yaml) using keys in the security.ssl.* namespace 
(Please see the [configuration page](config.html) for details). SSL can be 
selectively enabled/disabled for different transports using the following 
flags. These flags are only applicable when security.ssl.enabled is set to true.
+
+* **taskmanager.data.ssl.enabled**: SSL flag for data communication 
between task managers
+* **blob.service.ssl.enabled**: SSL flag for blob service client/server 
communication
+* **akka.ssl.enabled**: SSL flag for the akka based control connection 
between the flink client, jobmanager and taskmanager 
+* **jobmanager.web.ssl.enabled**: Flag to enable https access to the 
jobmanager's web frontend
+
+## Deploying Keystores and Truststores
+
+You need to have a Java Keystore generated and copied to each node in the 
flink cluster. The common name or subject alternative names in the certificate 
should match the node's hostname and IP address. Keystores and truststores can 
be generated using the keytool utility 
(https://docs.oracle.com/javase/8/docs/technotes/tools/unix/keytool.html). All 
flink components should have read access to the keystore and truststore files.
+
+### Example: Creating self signed CA and keystores for a 2 node cluster
+
+Execute the following keytool commands to create a truststore with a self 
signed CA
+~~~
+keytool -genkeypair -alias ca -keystore ca.keystore -dname "CN=Sample CA" 
-storepass password -keypass password -keyalg RSA -ext bc=ca:true
+keytool -keystore ca.keystore -storepass password -alias ca -exportcert > 
ca.cer
+keytool -importcert -keystore ca.truststore -alias ca -storepass password 
-noprompt -file ca.cer
+~~~
+
+Now create keystores for each node with certificates signed by the above 
CA. Let node1.company.org and node2.company.org be the hostnames with IPs 
192.168.1.1 and 192.168.1.2 respectively
+
+**Node 1**
+~~~
+keytool -genkeypair -alias node1 -keystore node1.keystore -dname 
"CN=node1.company.org" -ext SAN=dns:node1.company.org,ip:192.168.1.1 -storepass 
password -keypass password -keyalg RSA
+keytool -certreq -keystore node1.keystore -storepass password -alias node1 
-file node1.csr
+keytool -gencert -keystore ca.keystore -storepass password -alias ca -ext 
SAN=dns:node1.company.org,ip:192.168.1.1 -infile node1.csr -outfile node1.cer
+keytool -importcert -keystore node1.keystore -storepass password -file 
ca.cer -alias ca -noprompt
+keytool -importcert -keystore node1.keystore -storepass password -file 
node1.cer -alias node1 -noprompt
+~~~
+
+**Node 2**
+~~~
+keytool -genkeypair -alias node2 -keystore node2.keystore -dname 
"CN=node2.company.org" -ext SAN=dns:node2.company.org,ip:192.168.1.2 -storepass 
password -keypass password -keyalg RSA
+keytool -certreq -keystore node2.keystore -storepass password -alias node2 
-file node2.csr
+keytool -gencert -keystore ca.keystore -storepass password -alias ca -ext 
SAN=dns:node2.company.org,ip:192.168.1.2 -infile node2.csr -outfile node2.cer
+keytool -importcert -keystore node2.keystore -storepass password -file 
ca.cer -alias ca -noprompt
+keytool -importcert -keystore node2.keystore -storepass password -file 
node2.cer -alias node2 -noprompt
+~~~
+
+## Standalone Deployment
+Configure each node in the standalone cluster to pick up the keystore and 
truststore files present in the local file system.
+
+### Example: 2 node cluster
+* Generate 2 keystores, one for each node, and copy them to the filesystem 
on the respective node. Also copy the pulic key of the CA (which was used to 
sign the certificates in the keystore) as a Java truststore on both the nodes
+* Configure conf/flink-conf.yaml to pick up these files
+
+ Node 1
+~~~
+security.ssl.enabled: true
+security.ssl.keystore: /usr/local/node1.keystore
+security.ssl.keystore-password: abc123
+security.ssl.key-password: abc123
+security.ssl.truststore: /usr/local/ca.truststore
+security.ssl.truststore-password: abc123
+~~~
+
+ Node 2
+~~~
+security.ssl.enabled: true
+security.ssl.keystore: /usr/local/node2.keystore
+security.ssl.keystore-password: abc123
+security.ssl.key-password: abc123
   

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82412518
  
--- Diff: docs/setup/config.md ---
@@ -222,6 +238,7 @@ The following parameters configure Flink's JobManager 
and TaskManagers.
 - `jobmanager.web.backpressure.refresh-interval`: Time after which 
available stats are deprecated and need to be refreshed (DEFAULT: `6`, 1 
min).
 - `jobmanager.web.backpressure.num-samples`: Number of stack trace samples 
to take to determine back pressure (DEFAULT: `100`).
 - `jobmanager.web.backpressure.delay-between-samples`: Delay between stack 
trace samples to determine back pressure (DEFAULT: `50`, 50 ms).
+- `jobmanager.web.ssl.enabled`: Enable https access to the web frontend. 
This is applicable only when the global ssl flag security.ssl.enabled is set to 
true (DEFAULT: `true`).
--- End diff --

please see previous comment related to this


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82412499
  
--- Diff: docs/setup/config.md ---
@@ -211,6 +214,19 @@ The following parameters configure Flink's JobManager 
and TaskManagers.
 - `akka.throughput`: Number of messages that are processed in a batch 
before returning the thread to the pool. Low values denote a fair scheduling 
whereas high values can increase the performance at the cost of unfairness 
(DEFAULT: **15**).
 - `akka.log.lifecycle.events`: Turns on the Akka's remote logging of 
events. Set this value to 'true' in case of debugging (DEFAULT: **false**).
 - `akka.startup-timeout`: Timeout after which the startup of a remote 
component is considered being failed (DEFAULT: **akka.ask.timeout**).
+- `akka.ssl.enabled`: Turns on SSL for Akka's remote communication. This 
is applicable only when the global ssl flag security.ssl.enabled is set to true 
(DEFAULT: **true**).
--- End diff --

please see previous comment related to this


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[skrishnappa]

Github user skrishnappa commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82412401
  
--- Diff: docs/setup/config.md ---
@@ -140,6 +140,8 @@ will be used under the directory specified by 
jobmanager.web.tmpdir.
 
 - `blob.server.port`: Port definition for the blob server (serving user 
jar's) on the Taskmanagers. By default the port is set to 0, which means that 
the operating system is picking an ephemeral port. Flink also accepts a list of 
ports ("50100,50101"), ranges ("50100-50200") or a combination of both. It is 
recommended to set a range of ports to avoid collisions when multiple 
JobManagers are running on the same machine.
 
+- `blob.service.ssl.enabled`: Flag to enable ssl for the blob 
client/server communication. This is applicable only when the global ssl flag 
security.ssl.enabled is set to true (DEFAULT: true).
--- End diff --

Setting security.ssl.enabled to true will enable ssl for all communication. 
The other flags are required only for selectively disabling ssl (they are set 
to true by default). The reasons I chose to provide the extra flags are the 
following
* web frontend - this is useful if the admin chooses to encrypt only 
external traffic
* taskmanager data trasnfer - enabling ssl here might have significant 
performance impact and the admin might choose to encrypt only management and 
control traffic

I am slightly biased towards keeping these extra config (based on past 
experiences) and wasn't sure what the community would prefer. I am fine 
removing these and doing SSL - all or nothing, please let me know.



---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82214217
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -628,6 +641,11 @@
public static final String AKKA_TCP_TIMEOUT = "akka.tcp.timeout";
 
/**
+* Override SSL support for the Akka transport
+*/
+   public static final String AKKA_SSL_ENABLED = "akka.ssl.enabled";
--- End diff --

Should be removed unless there is a very good reason to introduce this 
parameter.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82214165
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -157,6 +157,9 @@
 */
public static final String BLOB_SERVER_PORT = "blob.server.port";
 
+   /** Flag to override ssl support for the blob service transport */
+   public static final String BLOB_SERVICE_SSL_ENABLED = 
"blob.service.ssl.enabled";
+
--- End diff --

Should be removed unless there is a very good reason to introduce this 
parameter.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82216311
  
--- Diff: 
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobManagerConfigHandler.java
 ---
@@ -45,7 +45,18 @@ public String handleRequest(Map 
pathParams, Map
for (String key : config.keySet()) {
gen.writeStartObject();
gen.writeStringField("key", key);
-   gen.writeStringField("value", config.getString(key, 
null));
+
+   // Mask key values which contain sensitive information
+   if(key.toLowerCase().contains("password")) {
+   String value = config.getString(key, null);
+   if(value != null) {
+   value = "**";
--- End diff --

Good point!


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82215957
  
--- Diff: 
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/files/StaticFileServerHandler.java
 ---
@@ -304,8 +314,15 @@ private void respondAsLeader(ChannelHandlerContext 
ctx, HttpRequest request, Str
ctx.write(response);
 
// write the content.
-   ctx.write(new DefaultFileRegion(raf.getChannel(), 0, 
fileLength), ctx.newProgressivePromise());
-   ChannelFuture lastContentFuture = 
ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+   ChannelFuture lastContentFuture;
+   if (ctx.pipeline().get(SslHandler.class) == null) {
+   ctx.write(new DefaultFileRegion(raf.getChannel(), 0, 
fileLength), ctx.newProgressivePromise());
+   lastContentFuture = 
ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+   } else {
+   lastContentFuture = ctx.writeAndFlush(new 
HttpChunkedInput(new ChunkedFile(raf, 0, fileLength, 8192)),
+   ctx.newProgressivePromise());
+   // HttpChunkedInput will write the end marker 
(LastHttpContent) for us.
--- End diff --

Why is this change necessary?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82214189
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -544,6 +552,11 @@
public static final String JOB_MANAGER_WEB_PORT_KEY = 
"jobmanager.web.port";
 
/**
+* Config parameter to override SSL support for the JobManager Web UI
+*/
+   public static final String JOB_MANAGER_WEB_SSL_ENABLED = 
"jobmanager.web.ssl.enabled";
--- End diff --

Should be removed unless there is a very good reason to introduce this 
parameter.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r8650
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientSslTest.java
 ---
@@ -0,0 +1,284 @@
+/*
+ * 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.blob;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.security.MessageDigest;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+import org.codehaus.jackson.map.DeserializerFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * This class contains unit tests for the {@link BlobClient} with ssl 
enabled.
+ */
+public class BlobClientSslTest {
+
+   /** The buffer size used during the tests in bytes. */
+   private static final int TEST_BUFFER_SIZE = 17 * 1000;
+
+   /** The instance of the BLOB server used during the tests. */
+   private static BlobServer BLOB_SERVER;
+
+   /** The blob service client configuration */
+   private static Configuration clientConfig;
+   /**
+* Starts the BLOB server.
+*/
+   @BeforeClass
+   public static void startServer() {
+   try {
+   Configuration config = new Configuration();
+   config.setBoolean(ConfigConstants.SECURITY_SSL_ENABLED, 
true);
+   config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE, 
"src/test/resources/local127.keystore");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE_PASSWORD, "password");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEY_PASSWORD, "password");
--- End diff --

Where do you explictly enable SSL for the BlobClient/BlobServer?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82214177
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -178,6 +181,11 @@
public static final String TASK_MANAGER_DATA_PORT_KEY = 
"taskmanager.data.port";
 
/**
+* Config parameter to override SSL support for taskmanager's data 
transport
+*/
+   public static final String TASK_MANAGER_DATA_SSL_ENABLED = 
"taskmanager.data.ssl.enabled";
--- End diff --

Should be removed unless there is a very good reason to introduce this 
parameter.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82213335
  
--- Diff: docs/setup/security-ssl.md ---
@@ -0,0 +1,139 @@
+---
+title: "SSL Setup"
+nav-parent_id: setup
+nav-pos: 8
+---
+
+
+This page provides instructions on how to enable SSL for the network 
communication between different flink components.
+
+## SSL Configuration
+
+SSL can be enabled for all network communication between flink components. 
SSL keystores and truststore has to be deployed on each flink node and 
configured (conf/flink-conf.yaml) using keys in the security.ssl.* namespace 
(Please see the [configuration page](config.html) for details). SSL can be 
selectively enabled/disabled for different transports using the following 
flags. These flags are only applicable when security.ssl.enabled is set to true.
+
+* **taskmanager.data.ssl.enabled**: SSL flag for data communication 
between task managers
+* **blob.service.ssl.enabled**: SSL flag for blob service client/server 
communication
+* **akka.ssl.enabled**: SSL flag for the akka based control connection 
between the flink client, jobmanager and taskmanager 
+* **jobmanager.web.ssl.enabled**: Flag to enable https access to the 
jobmanager's web frontend
+
+## Deploying Keystores and Truststores
+
+You need to have a Java Keystore generated and copied to each node in the 
flink cluster. The common name or subject alternative names in the certificate 
should match the node's hostname and IP address. Keystores and truststores can 
be generated using the keytool utility 
(https://docs.oracle.com/javase/8/docs/technotes/tools/unix/keytool.html). All 
flink components should have read access to the keystore and truststore files.
+
+### Example: Creating self signed CA and keystores for a 2 node cluster
+
+Execute the following keytool commands to create a truststore with a self 
signed CA
+~~~
+keytool -genkeypair -alias ca -keystore ca.keystore -dname "CN=Sample CA" 
-storepass password -keypass password -keyalg RSA -ext bc=ca:true
+keytool -keystore ca.keystore -storepass password -alias ca -exportcert > 
ca.cer
+keytool -importcert -keystore ca.truststore -alias ca -storepass password 
-noprompt -file ca.cer
+~~~
+
+Now create keystores for each node with certificates signed by the above 
CA. Let node1.company.org and node2.company.org be the hostnames with IPs 
192.168.1.1 and 192.168.1.2 respectively
+
+**Node 1**
+~~~
+keytool -genkeypair -alias node1 -keystore node1.keystore -dname 
"CN=node1.company.org" -ext SAN=dns:node1.company.org,ip:192.168.1.1 -storepass 
password -keypass password -keyalg RSA
+keytool -certreq -keystore node1.keystore -storepass password -alias node1 
-file node1.csr
+keytool -gencert -keystore ca.keystore -storepass password -alias ca -ext 
SAN=dns:node1.company.org,ip:192.168.1.1 -infile node1.csr -outfile node1.cer
+keytool -importcert -keystore node1.keystore -storepass password -file 
ca.cer -alias ca -noprompt
+keytool -importcert -keystore node1.keystore -storepass password -file 
node1.cer -alias node1 -noprompt
+~~~
+
+**Node 2**
+~~~
+keytool -genkeypair -alias node2 -keystore node2.keystore -dname 
"CN=node2.company.org" -ext SAN=dns:node2.company.org,ip:192.168.1.2 -storepass 
password -keypass password -keyalg RSA
+keytool -certreq -keystore node2.keystore -storepass password -alias node2 
-file node2.csr
+keytool -gencert -keystore ca.keystore -storepass password -alias ca -ext 
SAN=dns:node2.company.org,ip:192.168.1.2 -infile node2.csr -outfile node2.cer
+keytool -importcert -keystore node2.keystore -storepass password -file 
ca.cer -alias ca -noprompt
+keytool -importcert -keystore node2.keystore -storepass password -file 
node2.cer -alias node2 -noprompt
+~~~
+
+## Standalone Deployment
+Configure each node in the standalone cluster to pick up the keystore and 
truststore files present in the local file system.
+
+### Example: 2 node cluster
+* Generate 2 keystores, one for each node, and copy them to the filesystem 
on the respective node. Also copy the pulic key of the CA (which was used to 
sign the certificates in the keystore) as a Java truststore on both the nodes
+* Configure conf/flink-conf.yaml to pick up these files
+
+ Node 1
+~~~
+security.ssl.enabled: true
+security.ssl.keystore: /usr/local/node1.keystore
+security.ssl.keystore-password: abc123
+security.ssl.key-password: abc123
+security.ssl.truststore: /usr/local/ca.truststore
+security.ssl.truststore-password: abc123
+~~~
+
+ Node 2
+~~~
+security.ssl.enabled: true
+security.ssl.keystore: /usr/local/node2.keystore
+security.ssl.keystore-password: abc123
+security.ssl.key-password: abc123

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82211062
  
--- Diff: docs/setup/config.md ---
@@ -181,6 +183,7 @@ The following parameters configure Flink's JobManager 
and TaskManagers.
 - `taskmanager.hostname`: The hostname of the network interface that the 
TaskManager binds to. By default, the TaskManager searches for network 
interfaces that can connect to the JobManager and other TaskManagers. This 
option can be used to define a hostname if that strategy fails for some reason. 
Because different TaskManagers need different values for this option, it 
usually is specified in an additional non-shared TaskManager-specific config 
file.
 - `taskmanager.rpc.port`: The task manager's IPC port (DEFAULT: **0**, 
which lets the OS choose a free port).
 - `taskmanager.data.port`: The task manager's port used for data exchange 
operations (DEFAULT: **0**, which lets the OS choose a free port).
+- `taskmanager.data.ssl.enabled`: Enable SSL support for the taskmanager 
data transport. This is applicable only when the global ssl flag 
security.ssl.enabled is set to true (DEFAULT: **true**)
--- End diff --

Do we really need a switch for this? If we have `security.ssl.enabled` set 
to `true`, then this should always be enabled.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82211083
  
--- Diff: docs/setup/config.md ---
@@ -211,6 +214,19 @@ The following parameters configure Flink's JobManager 
and TaskManagers.
 - `akka.throughput`: Number of messages that are processed in a batch 
before returning the thread to the pool. Low values denote a fair scheduling 
whereas high values can increase the performance at the cost of unfairness 
(DEFAULT: **15**).
 - `akka.log.lifecycle.events`: Turns on the Akka's remote logging of 
events. Set this value to 'true' in case of debugging (DEFAULT: **false**).
 - `akka.startup-timeout`: Timeout after which the startup of a remote 
component is considered being failed (DEFAULT: **akka.ask.timeout**).
+- `akka.ssl.enabled`: Turns on SSL for Akka's remote communication. This 
is applicable only when the global ssl flag security.ssl.enabled is set to true 
(DEFAULT: **true**).
--- End diff --

Do we really need a switch for this? If we have `security.ssl.enabled` set 
to `true`, then this should always be enabled.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82230310
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientSslTest.java
 ---
@@ -0,0 +1,284 @@
+/*
+ * 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.blob;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.security.MessageDigest;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+import org.codehaus.jackson.map.DeserializerFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * This class contains unit tests for the {@link BlobClient} with ssl 
enabled.
+ */
+public class BlobClientSslTest {
+
+   /** The buffer size used during the tests in bytes. */
+   private static final int TEST_BUFFER_SIZE = 17 * 1000;
+
+   /** The instance of the BLOB server used during the tests. */
+   private static BlobServer BLOB_SERVER;
+
+   /** The blob service client configuration */
+   private static Configuration clientConfig;
+   /**
+* Starts the BLOB server.
+*/
+   @BeforeClass
+   public static void startServer() {
+   try {
+   Configuration config = new Configuration();
+   config.setBoolean(ConfigConstants.SECURITY_SSL_ENABLED, 
true);
+   config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE, 
"src/test/resources/local127.keystore");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE_PASSWORD, "password");
+   
config.setString(ConfigConstants.SECURITY_SSL_KEY_PASSWORD, "password");
--- End diff --

Ah, it is enabled by default when the global SSL flag is set. Can we fail 
the test if SSL is not enabled?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82211018
  
--- Diff: docs/setup/config.md ---
@@ -140,6 +140,8 @@ will be used under the directory specified by 
jobmanager.web.tmpdir.
 
 - `blob.server.port`: Port definition for the blob server (serving user 
jar's) on the Taskmanagers. By default the port is set to 0, which means that 
the operating system is picking an ephemeral port. Flink also accepts a list of 
ports ("50100,50101"), ranges ("50100-50200") or a combination of both. It is 
recommended to set a range of ports to avoid collisions when multiple 
JobManagers are running on the same machine.
 
+- `blob.service.ssl.enabled`: Flag to enable ssl for the blob 
client/server communication. This is applicable only when the global ssl flag 
security.ssl.enabled is set to true (DEFAULT: true).
--- End diff --

Do we really need a switch for this? If we have `security.ssl.enabled` set 
to `true`, then this should always be enabled.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2518: [FLINK-3931] Implement Transport Encryption (SSL/T...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2518#discussion_r82211248
  
--- Diff: docs/setup/config.md ---
@@ -222,6 +238,7 @@ The following parameters configure Flink's JobManager 
and TaskManagers.
 - `jobmanager.web.backpressure.refresh-interval`: Time after which 
available stats are deprecated and need to be refreshed (DEFAULT: `6`, 1 
min).
 - `jobmanager.web.backpressure.num-samples`: Number of stack trace samples 
to take to determine back pressure (DEFAULT: `100`).
 - `jobmanager.web.backpressure.delay-between-samples`: Delay between stack 
trace samples to determine back pressure (DEFAULT: `50`, 50 ms).
+- `jobmanager.web.ssl.enabled`: Enable https access to the web frontend. 
This is applicable only when the global ssl flag security.ssl.enabled is set to 
true (DEFAULT: `true`).
--- End diff --

Do we really need a switch for this? If we have `security.ssl.enabled` set 
to `true`, then this should always be enabled.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---