HADOOP-15980. Securing Hadoop RPC using SSL

The work done in this pull request implements SSL support for Hadoop RPC using the following steps,

1. Add SSL support to the RPC server
   a. Creates a RPC server implementation (Server.java) to use Netty
   b. Juxtaposes the Netty Implementation alongside the Native Java NIO APIs.
   c. Add SSL Handlers to the Netty Pipeline
   d. Parameterizes the RPC unit tests to run with and without SSL.
   e. Split the new classes into the constituent source files to reduce the clutter in Server.java.

2. Add Netty support to the RPC client
   a. Creates a RPC Client implementation (Client.java) to use Netty
   b. Juxtaposes the Netty Implementation alongside the Native Java NIO APIs.
   c. Add SSL Handlers to the Netty Pipeline
   d. Parameterizes the RPC unit tests to run with and without Netty.
   e. Split the new classes into the constituent source files to reduce the clutter in Client.java.

3. Add configuration to turn Netty on and off.

Author: Narayanan Venkateswaran

hadoop-client-modules/hadoop-client-minicluster/pom.xml

@@ -120,6 +120,10 @@
           <groupId>io.netty</groupId>
           <artifactId>netty</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>io.netty</groupId>
+          <artifactId>netty-all</artifactId>
+        </exclusion>
         <exclusion>
           <groupId>javax.servlet</groupId>
           <artifactId>javax.servlet-api</artifactId>

hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml

@@ -355,7 +355,7 @@
        to process initial sasl response token included in the INITIATE
      -->
      <Match>
-       <Class name="org.apache.hadoop.ipc.Server$Connection" />
+       <Class name="org.apache.hadoop.ipc.netty.server.Connection" />
        <Method name="processSaslMessage" />
        <Bug pattern="SF_SWITCH_FALLTHROUGH" />
      </Match>

hadoop-common-project/hadoop-common/pom.xml

@@ -408,6 +408,12 @@
       <artifactId>lz4-java</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-all</artifactId>
+      <version>4.1.27.Final</version>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
 
   <build>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java

@@ -59,6 +59,35 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
       "ipc.client.rpc-timeout.ms";
   /** Default value for IPC_CLIENT_RPC_TIMEOUT_KEY. */
   public static final int IPC_CLIENT_RPC_TIMEOUT_DEFAULT = 120000;
+  /** Enable SSL. */
+  public static final String IPC_SSL_KEY =
+      "ipc.ssl.enable";
+  /** Default value for IPC_SSL_KEY */
+  public static final boolean IPC_SSL_DEFAULT = false;
+  /**
+   * Enable the use of SSL self-signed certificates.
+   *
+   * NOTE:
+   *
+   * THIS IS INSECURE AND IS PURELY PRESENT FOR TESTING. THIS SHALL BE DISABLED
+   * IN FUTURE REVISIONS.
+   *
+   * **/
+  public static final String IPC_SSL_SELF_SIGNED_CERTIFICATE_TEST =
+      "test.ipc.ssl.self-signed-cert";
+  /**
+   * Default value for using SSL self-signed certificates
+   *
+   *
+   * NOTE:
+   *
+   * THIS IS INSECURE AND IS PURELY PRESENT FOR TESTING. THIS SHALL BE DISABLED
+   * IN FUTURE REVISIONS. FOR NOW THE DEFAULT IS false AND WILL NEED TO BE
+   * EXPLICITLY ENABLED FOR USE.
+   *
+   **/
+  public static final boolean IPC_SSL_SELF_SIGNED_CERTIFICATE_TEST_DEFAULT =
+      false;
   /** Responses larger than this will be logged */
   public static final String  IPC_SERVER_RPC_MAX_RESPONSE_SIZE_KEY =
     "ipc.server.max.response.size";

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallQueueManager.java

@@ -209,7 +209,7 @@ void addResponseTime(String name, Schedulable e, ProcessingDetails details) {
   }
 
   // This should be only called once per call and cached in the call object
-  int getPriorityLevel(Schedulable e) {
+  public int getPriorityLevel(Schedulable e) {
     return scheduler.getPriorityLevel(e);
   }
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java

@@ -31,10 +31,10 @@
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableUtils;
 import org.apache.hadoop.io.retry.RetryPolicies;
 import org.apache.hadoop.io.retry.RetryPolicy;
 import org.apache.hadoop.io.retry.RetryPolicy.RetryAction;
+import org.apache.hadoop.ipc.netty.client.IpcStreams;
 import org.apache.hadoop.ipc.RPC.RpcKind;
 import org.apache.hadoop.ipc.Server.AuthProtocol;
 import org.apache.hadoop.ipc.protobuf.IpcConnectionContextProtos.IpcConnectionContextProto;
@@ -62,13 +62,34 @@
 import javax.net.SocketFactory;
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslException;
-import java.io.*;
-import java.net.*;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InterruptedIOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.security.PrivilegedExceptionAction;
-import java.util.*;
+import java.util.Arrays;
+import java.util.Hashtable;
+import java.util.Iterator;
 import java.util.Map.Entry;
-import java.util.concurrent.*;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -77,6 +98,7 @@
 
 import static org.apache.hadoop.ipc.RpcConstants.CONNECTION_CONTEXT_CALL_ID;
 import static org.apache.hadoop.ipc.RpcConstants.PING_CALL_ID;
+import static org.apache.hadoop.ipc.netty.client.IpcStreams.*;
 
 /** A client for an IPC service.  IPC calls take a single {@link Writable} as a
  * parameter, and return a {@link Writable} as their value.  A service runs on
@@ -150,13 +172,7 @@ public static void setCallIdAndRetryCount(int cid, int rc,
   private final int maxAsyncCalls;
   private final AtomicInteger asyncCallCounter = new AtomicInteger(0);
 
-  /**
-   * Executor on which IPC calls' parameters are sent.
-   * Deferring the sending of parameters to a separate
-   * thread isolates them from thread interruptions in the
-   * calling code.
-   */
-  private final ExecutorService sendParamsExecutor;
+
   private final static ClientExecutorServiceFactory clientExcecutorFactory =
       new ClientExecutorServiceFactory();
 
@@ -437,6 +453,7 @@ private class Connection extends Thread {
     private final boolean tcpNoDelay; // if T then disable Nagle's Algorithm
     private final boolean tcpLowLatency; // if T then use low-delay QoS
     private final boolean doPing; //do we need to send ping message
+    private final boolean useNettySSL; // do we need SSL on in the implementation
     private final int pingInterval; // how often sends ping to the server
     private final int soTimeout; // used by ipc ping and rpc timeout
     private byte[] pingRequest; // ping message
@@ -460,6 +477,9 @@ private class Connection extends Thread {
       this.maxResponseLength = remoteId.conf.getInt(
           CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH,
           CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH_DEFAULT);
+      this.useNettySSL = remoteId.conf.getBoolean(
+          CommonConfigurationKeys.IPC_SSL_KEY,
+          CommonConfigurationKeys.IPC_SSL_DEFAULT);
       this.rpcTimeout = remoteId.getRpcTimeout();
       this.maxIdleTime = remoteId.getMaxIdleTime();
       this.connectionRetryPolicy = remoteId.connectionRetryPolicy;
@@ -837,7 +857,8 @@ private synchronized void setupIOstreams(
         Random rand = null;
         while (true) {
           setupConnection(ticket);
-          ipcStreams = new IpcStreams(socket, maxResponseLength);
+          ipcStreams = newInstance(socket, maxResponseLength,
+              remoteId.conf);
           writeConnectionHeader(ipcStreams);
           if (authProtocol == AuthProtocol.SASL) {
             try {
@@ -944,14 +965,17 @@ private void closeConnection() {
         return;
       }
       // close the current connection
-      try {
-        socket.close();
-      } catch (IOException e) {
-        LOG.warn("Not able to close a socket", e);
-      }
+      IOUtils.cleanupWithLogger(LOG, ipcStreams, socket);
       // set socket to null so that the next call to setupIOstreams
       // can start the process of connect all over again.
       socket = null;
+      // TODO: This change causes TestApplicationClientProtocolOnHA to throw
+      //       a NullPointerException on synchronized(ipcStreams.out). This
+      //       happens because closeConnection is called before the thread that
+      //       handles client.submit finishes. This change was introduced as
+      //       part of the SSL changes. Revisit later for further investigation
+      //       and possibly better handling.
+      // ipcStreams = null;
     }
 
     /* Handle connection failures due to timeout on connect
@@ -1165,11 +1189,8 @@ public void sendRpcRequest(final Call call)
         return;
       }
 
-      // Serialize the call to be sent. This is done from the actual
-      // caller thread, rather than the sendParamsExecutor thread,
-      
-      // so that if the serialization throws an error, it is reported
-      // properly. This also parallelizes the serialization.
+      // Serialize the call to be sent so that if the serialization throws an
+      // error, it is reported properly.
       //
       // Format of a call on the wire:
       // 0) Length of rest below (1 + 2)
@@ -1186,7 +1207,7 @@ public void sendRpcRequest(final Call call)
       RpcWritable.wrap(call.rpcRequest).writeTo(buf);
 
       synchronized (sendRpcRequestLock) {
-        Future<?> senderFuture = sendParamsExecutor.submit(new Runnable() {
+        Future<?> senderFuture = ipcStreams.submit(new Runnable() {
           @Override
           public void run() {
             try {
@@ -1377,7 +1398,11 @@ public Client(Class<? extends Writable> valueClass, Configuration conf,
             CommonConfigurationKeys.IPC_CLIENT_BIND_WILDCARD_ADDR_DEFAULT);
 
     this.clientId = ClientId.getClientId();
-    this.sendParamsExecutor = clientExcecutorFactory.refAndGetInstance();
+    // TODO: This call can be moved to the place where getClientExecutor is
+    //       invoked. However this move will change the general behaviour of
+    //       the client, which initializes the factory in the constructor. This
+    //       should be done with additional testing.
+    clientExcecutorFactory.refAndGetInstance();
     this.maxAsyncCalls = conf.getInt(
         CommonConfigurationKeys.IPC_CLIENT_ASYNC_CALLS_MAX_KEY,
         CommonConfigurationKeys.IPC_CLIENT_ASYNC_CALLS_MAX_DEFAULT);
@@ -1900,79 +1925,4 @@ public static int nextCallId() {
   public void close() throws Exception {
     stop();
   }
-
-  /** Manages the input and output streams for an IPC connection.
-   *  Only exposed for use by SaslRpcClient.
-   */
-  @InterfaceAudience.Private
-  public static class IpcStreams implements Closeable, Flushable {
-    private DataInputStream in;
-    public DataOutputStream out;
-    private int maxResponseLength;
-    private boolean firstResponse = true;
-
-    IpcStreams(Socket socket, int maxResponseLength) throws IOException {
-      this.maxResponseLength = maxResponseLength;
-      setInputStream(
-          new BufferedInputStream(NetUtils.getInputStream(socket)));
-      setOutputStream(
-          new BufferedOutputStream(NetUtils.getOutputStream(socket)));
-    }
-
-    void setSaslClient(SaslRpcClient client) throws IOException {
-      // Wrap the input stream in a BufferedInputStream to fill the buffer
-      // before reading its length (HADOOP-14062).
-      setInputStream(new BufferedInputStream(client.getInputStream(in)));
-      setOutputStream(client.getOutputStream(out));
-    }
-
-    private void setInputStream(InputStream is) {
-      this.in = (is instanceof DataInputStream)
-          ? (DataInputStream)is : new DataInputStream(is);
-    }
-
-    private void setOutputStream(OutputStream os) {
-      this.out = (os instanceof DataOutputStream)
-          ? (DataOutputStream)os : new DataOutputStream(os);
-    }
-
-    public ByteBuffer readResponse() throws IOException {
-      int length = in.readInt();
-      if (firstResponse) {
-        firstResponse = false;
-        // pre-rpcv9 exception, almost certainly a version mismatch.
-        if (length == -1) {
-          in.readInt(); // ignore fatal/error status, it's fatal for us.
-          throw new RemoteException(WritableUtils.readString(in),
-                                    WritableUtils.readString(in));
-        }
-      }
-      if (length <= 0) {
-        throw new RpcException(String.format("RPC response has " +
-            "invalid length of %d", length));
-      }
-      if (maxResponseLength > 0 && length > maxResponseLength) {
-        throw new RpcException(String.format("RPC response has a " +
-            "length of %d exceeds maximum data length", length));
-      }
-      ByteBuffer bb = ByteBuffer.allocate(length);
-      in.readFully(bb.array());
-      return bb;
-    }
-
-    public void sendRequest(byte[] buf) throws IOException {
-      out.write(buf);
-    }
-
-    @Override
-    public void flush() throws IOException {
-      out.flush();
-    }
-
-    @Override
-    public void close() {
-      IOUtils.closeStream(out);
-      IOUtils.closeStream(in);
-    }
-  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine2.java

@@ -644,7 +644,7 @@ private RpcWritable call(RPC.Server server,
   // htrace in the ipc layer creates the span name based on toString()
   // which uses the rpc header.  in the normal case we want to defer decoding
   // the rpc header until needed by the rpc engine.
-  static class RpcProtobufRequest extends RpcWritable.Buffer {
+  public static class RpcProtobufRequest extends RpcWritable.Buffer {
     private volatile RequestHeaderProto requestHeader;
     private Message payload;
 
@@ -656,7 +656,7 @@ static class RpcProtobufRequest extends RpcWritable.Buffer {
       this.payload = payload;
     }
 
-    RequestHeaderProto getRequestHeader() throws IOException {
+    public RequestHeaderProto getRequestHeader() throws IOException {
       if (getByteBuffer() != null && requestHeader == null) {
         requestHeader = getValue(RequestHeaderProto.getDefaultInstance());
       }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcException.java

@@ -30,7 +30,7 @@ public class RpcException extends IOException {
    * 
    * @param messages detailed message.
    */
-  RpcException(final String message) {
+  public RpcException(final String message) {
     super(message);
   }
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcWritable.java

@@ -37,7 +37,7 @@
 @InterfaceAudience.Private
 public abstract class RpcWritable implements Writable {
 
-  static RpcWritable wrap(Object o) {
+  public static RpcWritable wrap(Object o) {
     if (o instanceof RpcWritable) {
       return (RpcWritable)o;
     } else if (o instanceof Message) {