ADH-8535: [Backport] Hive2: ZooKeeper SSL/TLS support

common/src/java/org/apache/hadoop/hive/common/SSLZookeeperFactory.java

@@ -0,0 +1,89 @@
+/**
+ * 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.hive.common;
+
+import java.security.KeyStore;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.curator.utils.ZookeeperFactory;
+import org.apache.zookeeper.ClientCnxnSocketNetty;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.ClientX509Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory to create ZooKeeper clients with TLS enabled.
+ */
+public class SSLZookeeperFactory implements ZookeeperFactory {
+  private static final Logger LOG = LoggerFactory.getLogger(SSLZookeeperFactory.class);
+
+  private final boolean sslEnabled;
+  private final String keyStoreLocation;
+  private final String keyStorePassword;
+  private final String keyStoreType;
+  private final String trustStoreLocation;
+  private final String trustStorePassword;
+  private final String trustStoreType;
+
+  public SSLZookeeperFactory(boolean sslEnabled, String keyStoreLocation, String keyStorePassword,
+      String keyStoreType, String trustStoreLocation, String trustStorePassword,
+      String trustStoreType) {
+    this.sslEnabled = sslEnabled;
+    this.keyStoreLocation = StringUtils.defaultString(keyStoreLocation, "");
+    this.keyStorePassword = StringUtils.defaultString(keyStorePassword, "");
+    this.keyStoreType = StringUtils.isBlank(keyStoreType) ? KeyStore.getDefaultType() : keyStoreType;
+    this.trustStoreLocation = StringUtils.defaultString(trustStoreLocation, "");
+    this.trustStorePassword = StringUtils.defaultString(trustStorePassword, "");
+    this.trustStoreType = StringUtils.isBlank(trustStoreType) ? KeyStore.getDefaultType() : trustStoreType;
+    if (sslEnabled) {
+      if (StringUtils.isBlank(keyStoreLocation)) {
+        LOG.warn("Missing ZooKeeper keystore location");
+      }
+      if (StringUtils.isBlank(trustStoreLocation)) {
+        LOG.warn("Missing ZooKeeper truststore location");
+      }
+    }
+  }
+
+  @Override
+  public ZooKeeper newZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+      boolean canBeReadOnly) throws Exception {
+    if (!sslEnabled) {
+      return new ZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly);
+    }
+
+    ZKClientConfig clientConfig = new ZKClientConfig();
+    clientConfig.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
+    clientConfig.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET,
+        ClientCnxnSocketNetty.class.getName());
+
+    ClientX509Util x509Util = new ClientX509Util();
+    clientConfig.setProperty(x509Util.getSslKeystoreLocationProperty(), keyStoreLocation);
+    clientConfig.setProperty(x509Util.getSslKeystorePasswdProperty(), keyStorePassword);
+    clientConfig.setProperty(x509Util.getSslKeystoreTypeProperty(), keyStoreType);
+    clientConfig.setProperty(x509Util.getSslTruststoreLocationProperty(), trustStoreLocation);
+    clientConfig.setProperty(x509Util.getSslTruststorePasswdProperty(), trustStorePassword);
+    clientConfig.setProperty(x509Util.getSslTruststoreTypeProperty(), trustStoreType);
+
+    return new ZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly, clientConfig);
+  }
+}

common/src/java/org/apache/hadoop/hive/common/ZooKeeperHiveHelper.java

@@ -62,6 +62,13 @@ public class ZooKeeperHiveHelper {
     private final int sessionTimeout;
     private final int baseSleepTime;
     private final int maxRetries;
+    private final boolean sslEnabled;
+    private final String keyStoreLocation;
+    private final String keyStorePassword;
+    private final String keyStoreType;
+    private final String trustStoreLocation;
+    private final String trustStorePassword;
+    private final String trustStoreType;
 
     private CuratorFramework zooKeeperClient;
     private PersistentEphemeralNode znode;
@@ -74,13 +81,23 @@ public ZooKeeperHiveHelper(String quorum, String clientPort, String rootNamespac
     public ZooKeeperHiveHelper(String quorum, String clientPort, String rootNamespace,
                                int connectionTimeout, int sessionTimeout, int baseSleepTime,
                                int maxRetries) {
+        this(quorum, clientPort, rootNamespace, connectionTimeout, sessionTimeout, baseSleepTime,
+                maxRetries, false, null, null, null, null, null, null);
+    }
+
+    public ZooKeeperHiveHelper(String quorum, String clientPort, String rootNamespace,
+                               int connectionTimeout, int sessionTimeout, int baseSleepTime,
+                               int maxRetries, boolean sslEnabled, String keyStoreLocation,
+                               String keyStorePassword, String keyStoreType,
+                               String trustStoreLocation, String trustStorePassword,
+                               String trustStoreType) {
         // Get the ensemble server addresses in the format host1:port1, host2:port2, ... . Append
         // the configured port to hostname if the hostname doesn't contain a port.
         String[] hosts = quorum.split(",");
         StringBuilder quorumServers = new StringBuilder();
         for (int i = 0; i < hosts.length; i++) {
             quorumServers.append(hosts[i].trim());
-            if (!hosts[i].contains(":")) {
+            if (!hosts[i].contains(":") && clientPort != null && !clientPort.trim().isEmpty()) {
                 quorumServers.append(":");
                 quorumServers.append(clientPort);
             }
@@ -96,6 +113,13 @@ public ZooKeeperHiveHelper(String quorum, String clientPort, String rootNamespac
         this.sessionTimeout = sessionTimeout;
         this.baseSleepTime = baseSleepTime;
         this.maxRetries = maxRetries;
+        this.sslEnabled = sslEnabled;
+        this.keyStoreLocation = keyStoreLocation;
+        this.keyStorePassword = keyStorePassword;
+        this.keyStoreType = keyStoreType;
+        this.trustStoreLocation = trustStoreLocation;
+        this.trustStorePassword = trustStorePassword;
+        this.trustStoreType = trustStoreType;
     }
 
     /**
@@ -156,20 +180,7 @@ public void addServerInstanceToZooKeeper(String znodePathPrefix, String znodeDat
 
     public CuratorFramework startZookeeperClient(ACLProvider zooKeeperAclProvider,
                                                  boolean addParentNode) throws Exception {
-        String zooKeeperEnsemble = getQuorumServers();
-        // Create a CuratorFramework instance to be used as the ZooKeeper client.
-        // Use the zooKeeperAclProvider, when specified, to create appropriate ACLs.
-        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
-                .connectString(zooKeeperEnsemble)
-                .sessionTimeoutMs(sessionTimeout)
-                .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries));
-        if (connectionTimeout > 0) {
-            builder = builder.connectionTimeoutMs(connectionTimeout);
-        }
-        if (zooKeeperAclProvider != null) {
-            builder = builder.aclProvider(zooKeeperAclProvider);
-        }
-        CuratorFramework zkClient = builder.build();
+        CuratorFramework zkClient = getNewZookeeperClient(zooKeeperAclProvider);
         zkClient.start();
 
         // Create the parent znodes recursively; ignore if the parent already exists.
@@ -190,6 +201,41 @@ public CuratorFramework startZookeeperClient(ACLProvider zooKeeperAclProvider,
         return zkClient;
     }
 
+    public CuratorFramework getNewZookeeperClient() {
+        return getNewZookeeperClient(null);
+    }
+
+    public CuratorFramework getNewZookeeperClient(ACLProvider zooKeeperAclProvider) {
+        return getNewZookeeperClient(zooKeeperAclProvider, null);
+    }
+
+    public CuratorFramework getNewZookeeperClient(ACLProvider zooKeeperAclProvider,
+                                                  String namespace) {
+        // Create a CuratorFramework instance to be used as the ZooKeeper client.
+        // Use the zooKeeperAclProvider, when specified, to create appropriate ACLs.
+        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+                .connectString(getQuorumServers())
+                .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries));
+        if (namespace != null) {
+            builder = builder.namespace(namespace);
+        }
+        if (sessionTimeout > 0) {
+            builder = builder.sessionTimeoutMs(sessionTimeout);
+        }
+        if (connectionTimeout > 0) {
+            builder = builder.connectionTimeoutMs(connectionTimeout);
+        }
+        if (sslEnabled) {
+            builder = builder.zookeeperFactory(new SSLZookeeperFactory(sslEnabled,
+                    keyStoreLocation, keyStorePassword, keyStoreType,
+                    trustStoreLocation, trustStorePassword, trustStoreType));
+        }
+        if (zooKeeperAclProvider != null) {
+            builder = builder.aclProvider(zooKeeperAclProvider);
+        }
+        return builder.build();
+    }
+
     public void removeServerInstanceFromZooKeeper() throws Exception {
         setDeregisteredWithZooKeeper(true);