add after connected

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java

@@ -30,6 +30,9 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import org.apache.hadoop.hbase.io.crypto.tls.X509Util;
 import org.apache.hadoop.hbase.ipc.BufferCallBeforeInitHandler.BufferCallEvent;
 import org.apache.hadoop.hbase.ipc.HBaseRpcController.CancellationCallback;
 import org.apache.hadoop.hbase.security.NettyHBaseRpcConnectionHeaderHandler;
@@ -55,6 +58,7 @@
 import org.apache.hbase.thirdparty.io.netty.channel.ChannelPipeline;
 import org.apache.hbase.thirdparty.io.netty.channel.EventLoop;
 import org.apache.hbase.thirdparty.io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import org.apache.hbase.thirdparty.io.netty.handler.ssl.SslHandler;
 import org.apache.hbase.thirdparty.io.netty.handler.timeout.IdleStateHandler;
 import org.apache.hbase.thirdparty.io.netty.handler.timeout.ReadTimeoutHandler;
 import org.apache.hbase.thirdparty.io.netty.util.ReferenceCountUtil;
@@ -277,14 +281,7 @@ private void connect() throws UnknownHostException {
       .handler(new BufferCallBeforeInitHandler()).localAddress(rpcClient.localAddr)
       .remoteAddress(remoteAddr).connect().addListener(new ChannelFutureListener() {
 
-        @Override
-        public void operationComplete(ChannelFuture future) throws Exception {
-          Channel ch = future.channel();
-          if (!future.isSuccess()) {
-            failInit(ch, toIOE(future.cause()));
-            rpcClient.failedServers.addToFailedServers(remoteId.getAddress(), future.cause());
-            return;
-          }
+        private void succeed(Channel ch) throws IOException {
           ch.writeAndFlush(connectionHeaderPreamble.retainedDuplicate());
           if (useSasl) {
             saslNegotiate(ch);
@@ -294,6 +291,42 @@ public void operationComplete(ChannelFuture future) throws Exception {
             established(ch);
           }
         }
+
+        private void fail(Channel ch, Throwable error) {
+          failInit(ch, toIOE(error));
+          rpcClient.failedServers.addToFailedServers(remoteId.getAddress(), error);
+        }
+
+        @Override
+        public void operationComplete(ChannelFuture future) throws Exception {
+          Channel ch = future.channel();
+          if (!future.isSuccess()) {
+            fail(ch, future.cause());
+            return;
+          }
+          if (conf.getBoolean(X509Util.HBASE_CLIENT_NETTY_TLS_ENABLED, false)) {
+            X509Util x509Util = new X509Util(conf);
+            SSLContext sslContext = x509Util.createSSLContextAndOptions().getSSLContext();
+            SSLEngine sslEngine = sslContext.createSSLEngine(remoteId.address.getHostName(),
+              remoteId.address.getPort());
+            sslEngine.setUseClientMode(true);
+            SslHandler sslHandler = new SslHandler(sslEngine);
+            sslHandler.setHandshakeTimeoutMillis(
+              conf.getInt(X509Util.HBASE_CLIENT_NETTY_TLS_HANDSHAKETIMEOUT,
+                X509Util.DEFAULT_HANDSHAKE_DETECTION_TIMEOUT_MILLIS));
+            sslHandler.handshakeFuture().addListener(f -> {
+              if (f.isSuccess()) {
+                succeed(ch);
+              } else {
+                fail(ch, f.cause());
+              }
+            });
+            ch.pipeline().addFirst(sslHandler);
+          } else {
+            succeed(ch);
+          }
+
+        }
       }).channel();
   }
 

hbase-common/pom.xml

@@ -152,6 +152,16 @@
       <artifactId>kerb-simplekdc</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcprov-jdk15on</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcpkix-jdk15on</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

hbase-common/src/main/java/org/apache/hadoop/hbase/exceptions/X509Exception.java

@@ -0,0 +1,84 @@
+/*
+ * 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.hadoop.hbase.exceptions;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * This file has been copied from the Apache ZooKeeper project.
+ * @see <a href=
+ *      "https://github.com/apache/zookeeper/blob/c74658d398cdc1d207aa296cb6e20de00faec03e/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Exception.java">Base
+ *      revision</a>
+ */
+@SuppressWarnings("serial")
+@InterfaceAudience.Private
+public class X509Exception extends Exception {
+
+  public X509Exception(String message) {
+    super(message);
+  }
+
+  public X509Exception(Throwable cause) {
+    super(cause);
+  }
+
+  public X509Exception(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public static class KeyManagerException extends X509Exception {
+
+    public KeyManagerException(String message) {
+      super(message);
+    }
+
+    public KeyManagerException(Throwable cause) {
+      super(cause);
+    }
+
+  }
+
+  public static class TrustManagerException extends X509Exception {
+
+    public TrustManagerException(String message) {
+      super(message);
+    }
+
+    public TrustManagerException(Throwable cause) {
+      super(cause);
+    }
+
+  }
+
+  public static class SSLContextException extends X509Exception {
+
+    public SSLContextException(String message) {
+      super(message);
+    }
+
+    public SSLContextException(Throwable cause) {
+      super(cause);
+    }
+
+    public SSLContextException(String message, Throwable cause) {
+      super(message, cause);
+    }
+
+  }
+
+}

hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/tls/KeyStoreFileType.java

@@ -0,0 +1,122 @@
+/*
+ * 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.hadoop.hbase.io.crypto.tls;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * This enum represents the file type of a KeyStore or TrustStore. Currently, JKS (Java keystore),
+ * PEM, PKCS12, and BCFKS types are supported.
+ * <p/>
+ * This file has been copied from the Apache ZooKeeper project.
+ * @see <a href=
+ *      "https://github.com/apache/zookeeper/blob/c74658d398cdc1d207aa296cb6e20de00faec03e/zookeeper-server/src/main/java/org/apache/zookeeper/common/KeyStoreFileType.java">Base
+ *      revision</a>
+ */
+@InterfaceAudience.Private
+public enum KeyStoreFileType {
+  JKS(".jks"),
+  PEM(".pem"),
+  PKCS12(".p12"),
+  BCFKS(".bcfks");
+
+  private final String defaultFileExtension;
+
+  KeyStoreFileType(String defaultFileExtension) {
+    this.defaultFileExtension = defaultFileExtension;
+  }
+
+  /**
+   * The property string that specifies that a key store or trust store should use this store file
+   * type.
+   */
+  public String getPropertyValue() {
+    return this.name();
+  }
+
+  /**
+   * The file extension that is associated with this file type.
+   */
+  public String getDefaultFileExtension() {
+    return defaultFileExtension;
+  }
+
+  /**
+   * Converts a property value to a StoreFileType enum. If the property value is <code>null</code>
+   * or an empty string, returns <code>null</code>.
+   * @param propertyValue the property value.
+   * @return the KeyStoreFileType, or <code>null</code> if <code>propertyValue</code> is
+   *         <code>null</code> or empty.
+   * @throws IllegalArgumentException if <code>propertyValue</code> is not one of "JKS", "PEM",
+   *                                  "BCFKS", "PKCS12", or empty/null.
+   */
+  public static KeyStoreFileType fromPropertyValue(String propertyValue) {
+    if (propertyValue == null || propertyValue.length() == 0) {
+      return null;
+    }
+    return KeyStoreFileType.valueOf(propertyValue.toUpperCase());
+  }
+
+  /**
+   * Detects the type of KeyStore / TrustStore file from the file extension. If the file name ends
+   * with ".jks", returns <code>StoreFileType.JKS</code>. If the file name ends with ".pem", returns
+   * <code>StoreFileType.PEM</code>. If the file name ends with ".p12", returns
+   * <code>StoreFileType.PKCS12</code>. If the file name ends with ".bckfs", returns
+   * <code>StoreFileType.BCKFS</code>. Otherwise, throws an IllegalArgumentException.
+   * @param filename the filename of the key store or trust store file.
+   * @return a KeyStoreFileType.
+   * @throws IllegalArgumentException if the filename does not end with ".jks", ".pem", "p12" or
+   *                                  "bcfks".
+   */
+  public static KeyStoreFileType fromFilename(String filename) {
+    int i = filename.lastIndexOf('.');
+    if (i >= 0) {
+      String extension = filename.substring(i);
+      for (KeyStoreFileType storeFileType : KeyStoreFileType.values()) {
+        if (storeFileType.getDefaultFileExtension().equals(extension)) {
+          return storeFileType;
+        }
+      }
+    }
+    throw new IllegalArgumentException(
+      "Unable to auto-detect store file type from file name: " + filename);
+  }
+
+  /**
+   * If <code>propertyValue</code> is not null or empty, returns the result of
+   * <code>KeyStoreFileType.fromPropertyValue(propertyValue)</code>. Else, returns the result of
+   * <code>KeyStoreFileType.fromFileName(filename)</code>.
+   * @param propertyValue property value describing the KeyStoreFileType, or null/empty to
+   *                      auto-detect the type from the file name.
+   * @param filename      file name of the key store file. The file extension is used to auto-detect
+   *                      the KeyStoreFileType when <code>propertyValue</code> is null or empty.
+   * @return a KeyStoreFileType.
+   * @throws IllegalArgumentException if <code>propertyValue</code> is not one of "JKS", "PEM",
+   *                                  "PKCS12", "BCFKS", or empty/null.
+   * @throws IllegalArgumentException if <code>propertyValue</code>is empty or null and the type
+   *                                  could not be determined from the file name.
+   */
+  public static KeyStoreFileType fromPropertyValueOrFileName(String propertyValue,
+    String filename) {
+    KeyStoreFileType result = KeyStoreFileType.fromPropertyValue(propertyValue);
+    if (result == null) {
+      result = KeyStoreFileType.fromFilename(filename);
+    }
+    return result;
+  }
+}

hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/tls/SSLContextAndOptions.java

@@ -0,0 +1,88 @@
+/*
+ * 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.hadoop.hbase.io.crypto.tls;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.hadoop.hbase.io.crypto.tls.X509Util.CONFIG_PREFIX;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import javax.net.ssl.SSLContext;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hbase.thirdparty.io.netty.handler.ssl.ClientAuth;
+import org.apache.hbase.thirdparty.io.netty.handler.ssl.IdentityCipherSuiteFilter;
+import org.apache.hbase.thirdparty.io.netty.handler.ssl.JdkSslContext;
+import org.apache.hbase.thirdparty.io.netty.handler.ssl.SslContext;
+
+/**
+ * This file has been copied from the Apache ZooKeeper project.
+ * @see <a href=
+ *      "https://github.com/apache/zookeeper/blob/c74658d398cdc1d207aa296cb6e20de00faec03e/zookeeper-server/src/main/java/org/apache/zookeeper/common/SSLContextAndOptions.java">Base
+ *      revision</a>
+ */
+@InterfaceAudience.Private
+public class SSLContextAndOptions {
+  private static final String TLS_ENABLED_PROTOCOLS = CONFIG_PREFIX + "enabledProtocols";
+  private static final String TLS_CIPHER_SUITES = CONFIG_PREFIX + "ciphersuites";
+
+  private final String[] enabledProtocols;
+  private final List<String> cipherSuitesAsList;
+  private final SSLContext sslContext;
+
+  /**
+   * Note: constructor is intentionally package-private, only the X509Util class should be creating
+   * instances of this class.
+   * @param config     The HBase configuration
+   * @param sslContext The SSLContext.
+   */
+  SSLContextAndOptions(final Configuration config, final SSLContext sslContext) {
+    this.sslContext = requireNonNull(sslContext);
+    this.enabledProtocols = getEnabledProtocols(requireNonNull(config), sslContext);
+    String[] ciphers = getCipherSuites(config);
+    this.cipherSuitesAsList = Collections.unmodifiableList(Arrays.asList(ciphers));
+  }
+
+  public SSLContext getSSLContext() {
+    return sslContext;
+  }
+
+  public SslContext createNettyJdkSslContext(SSLContext sslContext, boolean isClientSocket) {
+    return new JdkSslContext(sslContext, isClientSocket, cipherSuitesAsList,
+      IdentityCipherSuiteFilter.INSTANCE, null, ClientAuth.NONE, enabledProtocols, false);
+  }
+
+  private String[] getEnabledProtocols(final Configuration config, final SSLContext sslContext) {
+    String enabledProtocolsInput = config.get(TLS_ENABLED_PROTOCOLS);
+    if (enabledProtocolsInput == null) {
+      return new String[] { sslContext.getProtocol() };
+    }
+    return enabledProtocolsInput.split(",");
+  }
+
+  private String[] getCipherSuites(final Configuration config) {
+    String cipherSuitesInput = config.get(TLS_CIPHER_SUITES);
+    if (cipherSuitesInput == null) {
+      return X509Util.getDefaultCipherSuites();
+    } else {
+      return cipherSuitesInput.split(",");
+    }
+  }
+}