HADOOP-18956. Zookeeper SSL/TLS support in ZKDelegationTokenSecretManager and ZKSignerSecretProvider (#6263)

Author: fapifta

hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/util/ZKSignerSecretProvider.java

@@ -16,25 +16,13 @@
 import org.apache.hadoop.classification.VisibleForTesting;
 import java.nio.ByteBuffer;
 import java.security.SecureRandom;
-import java.util.Collections;
-import java.util.List;
 import java.util.Properties;
 import java.util.Random;
-import javax.security.auth.login.Configuration;
 import javax.servlet.ServletContext;
-import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.framework.api.ACLProvider;
-import org.apache.curator.framework.imps.DefaultACLProvider;
-import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooDefs.Perms;
-import org.apache.zookeeper.client.ZKClientConfig;
-import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -92,6 +80,16 @@ public class ZKSignerSecretProvider extends RolloverSignerSecretProvider {
   public static final String ZOOKEEPER_KERBEROS_PRINCIPAL =
           CONFIG_PREFIX + "kerberos.principal";
 
+  public static final String ZOOKEEPER_SSL_ENABLED = CONFIG_PREFIX + "ssl.enabled";
+  public static final String ZOOKEEPER_SSL_KEYSTORE_LOCATION =
+      CONFIG_PREFIX + "ssl.keystore.location";
+  public static final String ZOOKEEPER_SSL_KEYSTORE_PASSWORD =
+      CONFIG_PREFIX + "ssl.keystore.password";
+  public static final String ZOOKEEPER_SSL_TRUSTSTORE_LOCATION =
+      CONFIG_PREFIX + "ssl.truststore.location";
+  public static final String ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD =
+      CONFIG_PREFIX + "ssl.truststore.password";
+
   /**
    * Constant for the property that specifies whether or not the Curator client
    * should disconnect from ZooKeeper on shutdown.  The default is "true".  Only
@@ -350,80 +348,33 @@ protected byte[] generateRandomSecret() {
    * This method creates the Curator client and connects to ZooKeeper.
    * @param config configuration properties
    * @return A Curator client
-   * @throws Exception thrown if an error occurred
    */
-  protected CuratorFramework createCuratorClient(Properties config)
-          throws Exception {
-    String connectionString = config.getProperty(
-            ZOOKEEPER_CONNECTION_STRING, "localhost:2181");
-
-    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
-    ACLProvider aclProvider;
+  protected CuratorFramework createCuratorClient(Properties config) {
+    String connectionString = config.getProperty(ZOOKEEPER_CONNECTION_STRING, "localhost:2181");
     String authType = config.getProperty(ZOOKEEPER_AUTH_TYPE, "none");
-    if (authType.equals("sasl")) {
-      LOG.info("Connecting to ZooKeeper with SASL/Kerberos"
-              + "and using 'sasl' ACLs");
-      String principal = setJaasConfiguration(config);
-      System.setProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY,
-              JAAS_LOGIN_ENTRY_NAME);
-      System.setProperty("zookeeper.authProvider.1",
-              "org.apache.zookeeper.server.auth.SASLAuthenticationProvider");
-      aclProvider = new SASLOwnerACLProvider(principal);
-    } else {  // "none"
-      LOG.info("Connecting to ZooKeeper without authentication");
-      aclProvider = new DefaultACLProvider();     // open to everyone
-    }
-    CuratorFramework cf = CuratorFrameworkFactory.builder()
-            .connectString(connectionString)
-            .retryPolicy(retryPolicy)
-            .aclProvider(aclProvider)
-            .build();
-    cf.start();
-    return cf;
-  }
-
-  private String setJaasConfiguration(Properties config) throws Exception {
-    String keytabFile = config.getProperty(ZOOKEEPER_KERBEROS_KEYTAB).trim();
-    if (keytabFile == null || keytabFile.length() == 0) {
-      throw new IllegalArgumentException(ZOOKEEPER_KERBEROS_KEYTAB
-              + " must be specified");
-    }
-    String principal = config.getProperty(ZOOKEEPER_KERBEROS_PRINCIPAL)
-            .trim();
-    if (principal == null || principal.length() == 0) {
-      throw new IllegalArgumentException(ZOOKEEPER_KERBEROS_PRINCIPAL
-              + " must be specified");
-    }
+    String keytab = config.getProperty(ZOOKEEPER_KERBEROS_KEYTAB, "").trim();
+    String principal = config.getProperty(ZOOKEEPER_KERBEROS_PRINCIPAL, "").trim();
 
-    // This is equivalent to writing a jaas.conf file and setting the system
-    // property, "java.security.auth.login.config", to point to it
-    JaasConfiguration jConf =
-            new JaasConfiguration(JAAS_LOGIN_ENTRY_NAME, principal, keytabFile);
-    Configuration.setConfiguration(jConf);
-    return principal.split("[/@]")[0];
-  }
+    boolean sslEnabled = Boolean.parseBoolean(config.getProperty(ZOOKEEPER_SSL_ENABLED, "false"));
+    String keystoreLocation = config.getProperty(ZOOKEEPER_SSL_KEYSTORE_LOCATION, "");
+    String keystorePassword = config.getProperty(ZOOKEEPER_SSL_KEYSTORE_PASSWORD, "");
+    String truststoreLocation = config.getProperty(ZOOKEEPER_SSL_TRUSTSTORE_LOCATION, "");
+    String truststorePassword = config.getProperty(ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD, "");
 
-  /**
-   * Simple implementation of an {@link ACLProvider} that simply returns an ACL
-   * that gives all permissions only to a single principal.
-   */
-  private static class SASLOwnerACLProvider implements ACLProvider {
-
-    private final List<ACL> saslACL;
-
-    private SASLOwnerACLProvider(String principal) {
-      this.saslACL = Collections.singletonList(
-              new ACL(Perms.ALL, new Id("sasl", principal)));
-    }
-
-    @Override
-    public List<ACL> getDefaultAcl() {
-      return saslACL;
-    }
-
-    @Override
-    public List<ACL> getAclForPath(String path) {
-      return saslACL;
-    }
+    CuratorFramework zkClient =
+        ZookeeperClient.configure()
+            .withConnectionString(connectionString)
+            .withAuthType(authType)
+            .withKeytab(keytab)
+            .withPrincipal(principal)
+            .withJaasLoginEntryName(JAAS_LOGIN_ENTRY_NAME)
+            .enableSSL(sslEnabled)
+            .withKeystore(keystoreLocation)
+            .withKeystorePassword(keystorePassword)
+            .withTruststore(truststoreLocation)
+            .withTruststorePassword(truststorePassword)
+            .create();
+    zkClient.start();
+    return zkClient;
   }
 }