Use Kerberos for Hive Authentication (#317)

Author: ishmum123

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/ConnectorArguments.java

@@ -126,6 +126,7 @@ public class ConnectorArguments extends DefaultArguments {
   public static final String OPT_HIVE_METASTORE_DUMP_PARTITION_METADATA =
       "hive-metastore-dump-partition-metadata";
   public static final String OPT_HIVE_METASTORE_DUMP_PARTITION_METADATA_DEFAULT = "true";
+  public static final String OPT_HIVE_KERBEROS_URL = "hive-kerberos-url";
 
   public static final String OPT_REQUIRED_IF_NOT_URL = "if --url is not specified";
   public static final String OPT_THREAD_POOL_SIZE = "thread-pool-size";
@@ -349,6 +350,18 @@ public class ConnectorArguments extends DefaultArguments {
           .withOptionalArg()
           .withValuesConvertedBy(BooleanValueConverter.INSTANCE)
           .defaultsTo(Boolean.parseBoolean(OPT_HIVE_METASTORE_DUMP_PARTITION_METADATA_DEFAULT));
+  private final OptionSpec<String> optionHiveKerberosUrl =
+      parser
+          .accepts(
+              OPT_HIVE_KERBEROS_URL,
+              "Kerberos URL to use to authenticate Hive Thrift API. Please note that we don't"
+                  + " accept Kerberos `REALM` in the URL. Please ensure that the tool runs in an"
+                  + " environment where the default `REALM` is known and used. It's recommended to"
+                  + " generate a Kerberos ticket with the same user before running the dumper. The"
+                  + " tool will prompt for credentials if a ticket is not provided.")
+          .withOptionalArg()
+          .ofType(String.class)
+          .describedAs("principal/host");
 
   // Threading / Pooling
   private final OptionSpec<Integer> optionThreadPoolSize =
@@ -818,6 +831,11 @@ public boolean isHiveMetastorePartitionMetadataDumpingEnabled() {
     return BooleanUtils.isTrue(getOptions().valueOf(optionHivePartitionMetadataCollection));
   }
 
+  @CheckForNull
+  public String getHiveKerberosUrl() {
+    return getOptions().valueOf(optionHiveKerberosUrl);
+  }
+
   public boolean saveResponseFile() {
     return getOptions().has(optionSaveResponse);
   }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/hive/AbstractHiveConnector.java

@@ -39,6 +39,7 @@
 import javax.annotation.Nonnull;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
+import javax.security.sasl.SaslException;
 import org.apache.thrift.transport.TTransportException;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.slf4j.Logger;
@@ -63,7 +64,8 @@
     order = 401,
     arg = ConnectorArguments.OPT_HIVE_METASTORE_DUMP_PARTITION_METADATA,
     description =
-        "Dump partition metadata; you may wish to disable this for production metastores with a significant number of partitions due to Thrift client performance implications.",
+        "Dump partition metadata; you may wish to disable this for production metastores with a"
+            + " significant number of partitions due to Thrift client performance implications.",
     defaultValue = ConnectorArguments.OPT_HIVE_METASTORE_DUMP_PARTITION_METADATA_DEFAULT)
 @RespectsInput(
     order = 500,
@@ -122,7 +124,7 @@ public ThriftClientPool(
                     builtClients.add(client);
                   }
                   return client;
-                } catch (TTransportException e) {
+                } catch (Exception e) {
                   throw new RuntimeException(
                       "Unable to build Thrift client '"
                           + threadName
@@ -191,7 +193,8 @@ public ThriftClientHandle(
 
     /** Returns a thread-unsafe Thrift client unsuitable for use in multi-threaded contexts. */
     @Nonnull
-    public HiveMetastoreThriftClient newClient(@Nonnull String name) throws TTransportException {
+    public HiveMetastoreThriftClient newClient(@Nonnull String name)
+        throws TTransportException, SaslException {
       LOG.debug("Creating a new Thrift client named '{}'.", name);
       return new HiveMetastoreThriftClient.Builder(thriftClientBuilder).withName(name).build();
     }
@@ -200,7 +203,8 @@ public HiveMetastoreThriftClient newClient(@Nonnull String name) throws TTranspo
     @Nonnull
     public ThriftClientPool newMultiThreadedThriftClientPool(@Nonnull String name) {
       LOG.debug(
-          "Creating a new multi-threaded pooled Thrift client named '{}' backed by a thread pool of size {}.",
+          "Creating a new multi-threaded pooled Thrift client named '{}' backed by a thread pool of"
+              + " size {}.",
           name,
           threadPoolSize);
       return new ThriftClientPool(name, thriftClientBuilder, threadPoolSize);
@@ -255,7 +259,8 @@ public Handle open(ConnectorArguments arguments) throws Exception {
                 arguments.getPort(
                     Integer.parseInt(ConnectorArguments.OPT_HIVE_METASTORE_PORT_DEFAULT)))
             .withUnavailableClientVersionBehavior(
-                HiveMetastoreThriftClient.Builder.UnavailableClientVersionBehavior.FALLBACK);
+                HiveMetastoreThriftClient.Builder.UnavailableClientVersionBehavior.FALLBACK)
+            .withKerberosUrl(arguments.getHiveKerberosUrl());
     return new ThriftClientHandle(thriftClientBuilder, arguments.getThreadPoolSize());
   }
 }

dumper/lib-ext-hive-metastore/src/main/java/com/google/edwmigration/dumper/ext/hive/metastore/HiveMetastoreThriftClient.java

@@ -19,14 +19,19 @@
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import javax.annotation.Nonnegative;
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.NotThreadSafe;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
 import org.apache.thrift.TConfiguration;
 import org.apache.thrift.protocol.TBinaryProtocol;
 import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TSaslClientTransport;
 import org.apache.thrift.transport.TSocket;
 import org.apache.thrift.transport.TTransport;
 import org.apache.thrift.transport.TTransportException;
@@ -70,6 +75,7 @@ public static enum UnavailableClientVersionBehavior {
     @Nonnull private String name = "unnamed-thrift-client";
     @Nonnull private String host = "localhost";
     @Nonnegative private int port;
+    @Nullable private String kerberosUrl;
 
     @Nonnull
     private UnavailableClientVersionBehavior unavailableClientBehavior =
@@ -89,6 +95,7 @@ public Builder(@Nonnull Builder builder) {
       this.port = builder.port;
       this.unavailableClientBehavior = builder.unavailableClientBehavior;
       this.debug = builder.debug;
+      this.kerberosUrl = builder.kerberosUrl;
     }
 
     @Nonnull
@@ -109,6 +116,12 @@ public Builder withPort(@Nonnegative int port) {
       return this;
     }
 
+    @Nonnull
+    public Builder withKerberosUrl(@Nullable String kerberosUrl) {
+      this.kerberosUrl = kerberosUrl;
+      return this;
+    }
+
     @Nonnull
     public Builder withUnavailableClientVersionBehavior(
         @Nonnull UnavailableClientVersionBehavior behavior) {
@@ -123,15 +136,7 @@ public Builder withDebug(boolean value) {
     }
 
     @Nonnull
-    public HiveMetastoreThriftClient build() throws TTransportException {
-
-      // We used to support Kerberos authentication, but that was when we used the Hive metastore
-      // client
-      // wrapper around Thrift. Now that we are connecting via Thrift directly, we would need to
-      // wrap
-      // the TTransport here with a TSaslClientTransport parameterized accordingly. This hasn't been
-      // done yet
-      // in the interest of expediency.
+    private TTransport createTransport() throws SaslException, TTransportException {
       TTransport transport =
           new TSocket(
               new TConfiguration(
@@ -140,14 +145,41 @@ public HiveMetastoreThriftClient build() throws TTransportException {
                   TConfiguration.DEFAULT_RECURSION_DEPTH),
               host,
               port);
+
+      if (kerberosUrl == null) {
+        return transport;
+      }
+
+      String[] urlParts = kerberosUrl.split("/");
+
+      if (urlParts.length != 2) {
+        throw new IllegalArgumentException(
+            "Please provide an URL in the format of `principal/cluster`");
+      }
+
+      Map<String, String> saslProperties = new HashMap<>();
+      saslProperties.put(Sasl.SERVER_AUTH, "true");
+      saslProperties.put(Sasl.QOP, "auth-conf");
+
+      // See:
+      // https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/single-signon.html
+      return new TSaslClientTransport(
+          "GSSAPI", null, urlParts[0], urlParts[1], saslProperties, null, transport);
+    }
+
+    @Nonnull
+    public HiveMetastoreThriftClient build() throws TTransportException, SaslException {
+      TTransport transport = createTransport();
+
       TProtocol protocol = new TBinaryProtocol(transport);
       transport.open();
 
       final HiveMetastoreThriftClient client;
       if (supportedVersionMappings.containsKey(requestedVersionString)) {
         if (debug)
           LOG.debug(
-              "The request for Hive metastore Thrift client version '{}' is satisfiable; building it now.",
+              "The request for Hive metastore Thrift client version '{}' is satisfiable; building"
+                  + " it now.",
               requestedVersionString);
         client = supportedVersionMappings.get(requestedVersionString).provide(name, protocol);
       } else {
@@ -161,9 +193,10 @@ public HiveMetastoreThriftClient build() throws TTransportException {
         if (unavailableClientBehavior == UnavailableClientVersionBehavior.FALLBACK) {
           LOG.warn(
               messagePrefix
-                  + " The caller requested fallback behavior, so a client compiled against a superset Thrift specification will be used instead. "
-                  + "If you encounter an error when using the fallback client, please contact CompilerWorks support and provide "
-                  + "the originally requested version number.");
+                  + " The caller requested fallback behavior, so a client compiled against a"
+                  + " superset Thrift specification will be used instead. If you encounter an error"
+                  + " when using the fallback client, please contact CompilerWorks support and"
+                  + " provide the originally requested version number.");
           client = new HiveMetastoreThriftClient_Superset(name, protocol);
         } else {
           throw new UnsupportedOperationException(