[#3489] improvement(hive-catalog): Add user authentication e2e test for Hive catalog

api/src/main/java/com/datastrato/gravitino/rel/expressions/sorts/SortOrders.java

@@ -10,6 +10,8 @@
 /** Helper methods to create SortOrders to pass into Gravitino. */
 public class SortOrders {
 
+  /** NONE is used to indicate that there is no sort order. */
+  public static final SortOrder[] NONE = new SortOrder[0];
   /**
    * Create a sort order by the given expression with the ascending sort direction and nulls first
    * ordering.

catalogs/catalog-hive/build.gradle.kts

@@ -165,6 +165,7 @@ tasks.test {
 
     doFirst {
       environment("GRAVITINO_CI_HIVE_DOCKER_IMAGE", "datastrato/gravitino-ci-hive:0.1.12")
+      environment("GRAVITINO_CI_KERBEROS_HIVE_DOCKER_IMAGE", "datastrato/gravitino-ci-kerberos-hive:0.1.1")
     }
 
     val init = project.extra.get("initIntegrationTest") as (Test) -> Unit

catalogs/catalog-hive/src/main/java/com/datastrato/gravitino/catalog/hive/HiveProxyPlugin.java

@@ -58,7 +58,7 @@ public Object doAs(
             ops.getClientPool()
                 .run(
                     client -> {
-                      return client.getDelegationToken(realUser.getUserName(), principal.getName());
+                      return client.getDelegationToken(principal.getName(), realUser.getUserName());
                     });
 
         Token<DelegationTokenIdentifier> delegationToken = new Token<DelegationTokenIdentifier>();

catalogs/catalog-hive/src/test/java/com/datastrato/gravitino/catalog/hive/integration/test/HiveUserAuthenticationIT.java

@@ -0,0 +1,226 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.catalog.hive.integration.test;
+
+import static com.datastrato.gravitino.catalog.hive.HiveCatalogPropertiesMeta.IMPERSONATION_ENABLE;
+import static com.datastrato.gravitino.catalog.hive.HiveCatalogPropertiesMeta.KET_TAB_URI;
+import static com.datastrato.gravitino.catalog.hive.HiveCatalogPropertiesMeta.METASTORE_URIS;
+import static com.datastrato.gravitino.catalog.hive.HiveCatalogPropertiesMeta.PRINCIPAL;
+import static com.datastrato.gravitino.connector.BaseCatalog.CATALOG_BYPASS_PREFIX;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION;
+
+import com.datastrato.gravitino.Catalog;
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.client.GravitinoAdminClient;
+import com.datastrato.gravitino.client.GravitinoMetalake;
+import com.datastrato.gravitino.client.KerberosTokenProvider;
+import com.datastrato.gravitino.integration.test.container.ContainerSuite;
+import com.datastrato.gravitino.integration.test.container.HiveContainer;
+import com.datastrato.gravitino.integration.test.util.AbstractIT;
+import com.datastrato.gravitino.rel.Column;
+import com.datastrato.gravitino.rel.expressions.distributions.Distributions;
+import com.datastrato.gravitino.rel.expressions.sorts.SortOrders;
+import com.datastrato.gravitino.rel.expressions.transforms.Transforms;
+import com.datastrato.gravitino.rel.types.Types;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Tag("gravitino-docker-it")
+public class HiveUserAuthenticationIT extends AbstractIT {
+  private static final Logger LOG = LoggerFactory.getLogger(HiveUserAuthenticationIT.class);
+
+  private static final ContainerSuite containerSuite = ContainerSuite.getInstance();
+
+  private static final String SDK_KERBEROS_PRINCIPAL_KEY = "client.kerberos.principal";
+  private static final String SDK_KERBEROS_KEYTAB_KEY = "client.kerberos.keytab";
+
+  private static final String GRAVITINO_CLIENT_PRINCIPAL = "gravitino_client@HADOOPKRB";
+  private static final String GRAVITINO_CLIENT_KEYTAB = "/gravitino_client.keytab";
+
+  private static final String GRAVITINO_SERVER_PRINCIPAL = "HTTP/localhost@HADOOPKRB";
+  private static final String GRAVITINO_SERVER_KEYTAB = "/gravitino_server.keytab";
+
+  private static final String HIVE_METASTORE_CLIENT_PRINCIPAL = "cli@HADOOPKRB";
+  private static final String HIVE_METASTORE_CLIENT_KEYTAB = "/client.keytab";
+
+  private static String TMP_DIR;
+
+  private static String HIVE_METASTORE_URI;
+
+  private static GravitinoAdminClient adminClient;
+
+  private static HiveContainer kerberosHiveContainer;
+
+  private static final String METALAKE_NAME = "test_metalake";
+  private static final String CATALOG_NAME = "test_catalog";
+  private static final String SCHEMA_NAME = "test_schema";
+  private static final String TABLE_NAME = "test_table";
+
+  private static final String HIVE_COL_NAME1 = "col1";
+  private static final String HIVE_COL_NAME2 = "col2";
+  private static final String HIVE_COL_NAME3 = "col3";
+
+  @BeforeAll
+  public static void startIntegrationTest() throws Exception {
+    containerSuite.startKerberosHiveContainer();
+    kerberosHiveContainer = containerSuite.getKerberosHiveContainer();
+
+    File baseDir = new File(System.getProperty("java.io.tmpdir"));
+    File file = Files.createTempDirectory(baseDir.toPath(), "test").toFile();
+    file.deleteOnExit();
+    TMP_DIR = file.getAbsolutePath();
+
+    HIVE_METASTORE_URI =
+        String.format(
+            "thrift://%s:%d",
+            kerberosHiveContainer.getContainerIpAddress(), HiveContainer.HIVE_METASTORE_PORT);
+
+    // Prepare kerberos related-config;
+    prepareKerberosConfig();
+
+    // Config kerberos configuration for Gravitino server
+    addKerberosConfig();
+
+    // Start Gravitino server
+    AbstractIT.startIntegrationTest();
+  }
+
+  @AfterAll
+  public static void stopIntegrationTest() {
+    // Reset the UGI
+    UserGroupInformation.reset();
+
+    // Clean up the kerberos configuration
+    System.clearProperty("java.security.krb5.conf");
+    System.clearProperty("sun.security.krb5.debug");
+  }
+
+  private static void prepareKerberosConfig() throws IOException {
+    // Keytab of the Gravitino SDK client
+    kerberosHiveContainer
+        .getContainer()
+        .copyFileFromContainer("/gravitino_client.keytab", TMP_DIR + GRAVITINO_CLIENT_KEYTAB);
+
+    // Keytab of the Gravitino server
+    kerberosHiveContainer
+        .getContainer()
+        .copyFileFromContainer("/gravitino_server.keytab", TMP_DIR + GRAVITINO_SERVER_KEYTAB);
+
+    // Keytab of Gravitino server to connector to Hive
+    kerberosHiveContainer
+        .getContainer()
+        .copyFileFromContainer("/etc/admin.keytab", TMP_DIR + HIVE_METASTORE_CLIENT_KEYTAB);
+
+    String tmpKrb5Path = TMP_DIR + "krb5.conf_tmp";
+    String krb5Path = TMP_DIR + "krb5.conf";
+    kerberosHiveContainer.getContainer().copyFileFromContainer("/etc/krb5.conf", tmpKrb5Path);
+
+    // Modify the krb5.conf and change the kdc and admin_server to the container IP
+    String ip = containerSuite.getKerberosHiveContainer().getContainerIpAddress();
+    String content = FileUtils.readFileToString(new File(tmpKrb5Path), StandardCharsets.UTF_8);
+    content = content.replace("kdc = localhost:88", "kdc = " + ip + ":88");
+    content = content.replace("admin_server = localhost", "admin_server = " + ip + ":749");
+    FileUtils.write(new File(krb5Path), content, StandardCharsets.UTF_8);
+
+    LOG.info("Kerberos kdc config:\n{}", content);
+    System.setProperty("java.security.krb5.conf", krb5Path);
+    System.setProperty("sun.security.krb5.debug", "true");
+  }
+
+  private static void addKerberosConfig() {
+    AbstractIT.customConfigs.put("gravitino.authenticator", "kerberos");
+    AbstractIT.customConfigs.put(
+        "gravitino.authenticator.kerberos.principal", GRAVITINO_SERVER_PRINCIPAL);
+    AbstractIT.customConfigs.put(
+        "gravitino.authenticator.kerberos.keytab", TMP_DIR + GRAVITINO_SERVER_KEYTAB);
+    AbstractIT.customConfigs.put(SDK_KERBEROS_KEYTAB_KEY, TMP_DIR + GRAVITINO_CLIENT_KEYTAB);
+    AbstractIT.customConfigs.put(SDK_KERBEROS_PRINCIPAL_KEY, GRAVITINO_CLIENT_PRINCIPAL);
+  }
+
+  @Test
+  public void testUserAuthentication() {
+    KerberosTokenProvider provider =
+        KerberosTokenProvider.builder()
+            .withClientPrincipal(GRAVITINO_CLIENT_PRINCIPAL)
+            .withKeyTabFile(new File(TMP_DIR + GRAVITINO_CLIENT_KEYTAB))
+            .build();
+    adminClient = GravitinoAdminClient.builder(serverUri).withKerberosAuth(provider).build();
+
+    GravitinoMetalake[] metalakes = adminClient.listMetalakes();
+    Assertions.assertEquals(0, metalakes.length);
+
+    GravitinoMetalake gravitinoMetalake =
+        adminClient.createMetalake(METALAKE_NAME, null, ImmutableMap.of());
+
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put(METASTORE_URIS, HIVE_METASTORE_URI);
+    properties.put(IMPERSONATION_ENABLE, "true");
+    properties.put(KET_TAB_URI, TMP_DIR + HIVE_METASTORE_CLIENT_KEYTAB);
+    properties.put(PRINCIPAL, HIVE_METASTORE_CLIENT_PRINCIPAL);
+
+    properties.put(CATALOG_BYPASS_PREFIX + HADOOP_SECURITY_AUTHENTICATION, "kerberos");
+    properties.put(
+        CATALOG_BYPASS_PREFIX + "hive.metastore.kerberos.principal",
+        "hive/_HOST@HADOOPKRB"
+            .replace("_HOST", containerSuite.getKerberosHiveContainer().getHostName()));
+    properties.put(CATALOG_BYPASS_PREFIX + "hive.metastore.sasl.enabled", "true");
+
+    Catalog catalog =
+        gravitinoMetalake.createCatalog(
+            CATALOG_NAME, Catalog.Type.RELATIONAL, "hive", "comment", properties);
+
+    Exception exception =
+        Assertions.assertThrows(
+            Exception.class,
+            () -> catalog.asSchemas().createSchema(SCHEMA_NAME, "comment", ImmutableMap.of()));
+    String exceptionMessage = Throwables.getStackTraceAsString(exception);
+    // Make sure real user is 'gravitino_client'
+    Assertions.assertTrue(
+        exceptionMessage.contains("Permission denied: user=gravitino_client, access=WRITE"));
+
+    // Now try to give the user the permission to create schema again
+    kerberosHiveContainer.executeInContainer(
+        "hadoop", "fs", "-chmod", "-R", "777", "/user/hive/warehouse");
+    Assertions.assertDoesNotThrow(
+        () -> catalog.asSchemas().createSchema(SCHEMA_NAME, "comment", ImmutableMap.of()));
+
+    // Create table
+    NameIdentifier tableNameIdentifier =
+        NameIdentifier.of(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME, TABLE_NAME);
+    catalog
+        .asTableCatalog()
+        .createTable(
+            tableNameIdentifier,
+            createColumns(),
+            "",
+            ImmutableMap.of(),
+            Transforms.EMPTY_TRANSFORM,
+            Distributions.NONE,
+            SortOrders.NONE);
+  }
+
+  private static Column[] createColumns() {
+    Column col1 = Column.of(HIVE_COL_NAME1, Types.ByteType.get(), "col_1_comment");
+    Column col2 = Column.of(HIVE_COL_NAME2, Types.DateType.get(), "col_2_comment");
+    Column col3 = Column.of(HIVE_COL_NAME3, Types.StringType.get(), "col_3_comment");
+    return new Column[] {col1, col2, col3};
+  }
+}

dev/docker/kerberos-hive/core-site.xml

@@ -24,6 +24,41 @@
     <value>*</value>
   </property>
 
+  <property>
+    <name>hadoop.proxyuser.hive.users</name>
+    <value>*</value>
+  </property>
+
+  <property>
+    <name>hadoop.proxyuser.cli.hosts</name>
+    <value>*</value>
+  </property>
+
+  <property>
+    <name>hadoop.proxyuser.cli.groups</name>
+    <value>*</value>
+  </property>
+
+  <property>
+    <name>hadoop.proxyuser.cli.users</name>
+    <value>*</value>
+  </property>
+
+  <property>
+    <name>hadoop.proxyuser.hadoop.hosts</name>
+    <value>*</value>
+  </property>
+
+  <property>
+    <name>hadoop.proxyuser.hadoop.groups</name>
+    <value>*</value>
+  </property>
+
+  <property>
+    <name>hadoop.proxyuser.hadoop.users</name>
+    <value>*</value>
+  </property>
+
   <property>
     <name>hadoop.proxyuser.root.groups</name>
     <value>*</value>
@@ -34,6 +69,11 @@
     <value>*</value>
   </property>
 
+  <property>
+    <name>hadoop.proxyuser.root.users</name>
+    <value>*</value>
+  </property>
+
   <property>
     <name>hadoop.security.auth_to_local</name>
     <value>

dev/docker/kerberos-hive/start.sh

@@ -42,6 +42,13 @@ kadmin.local -q "ktadd -norandkey -k ${KRB5_KTNAME} hive/${HOSTNAME}@${FQDN}"
 kadmin.local -q "xst -k /hive.keytab -norandkey hive/${HOSTNAME}@${FQDN}"
 kadmin.local -q "xst -k /cli.keytab -norandkey cli@${FQDN}"
 
+# For Gravitino web server
+echo -e "${PASS}\n${PASS}" | kadmin.local -q "addprinc gravitino_client@${FQDN}"
+kadmin.local -q "ktadd -norandkey -k /gravitino_client.keytab gravitino_client@${FQDN}"
+
+echo -e "${PASS}\n${PASS}" | kadmin.local -q "addprinc HTTP/localhost@${FQDN}"
+kadmin.local -q "ktadd -norandkey -k /gravitino_server.keytab HTTP/localhost@${FQDN}"
+
 echo -e "${PASS}\n" | kinit hive/${HOSTNAME}
 
 # Update the configuration file

integration-test-common/src/test/java/com/datastrato/gravitino/integration/test/MiniGravitino.java

@@ -12,7 +12,9 @@
 import com.datastrato.gravitino.auth.AuthenticatorType;
 import com.datastrato.gravitino.auxiliary.AuxiliaryServiceManager;
 import com.datastrato.gravitino.client.ErrorHandlers;
+import com.datastrato.gravitino.client.GravitinoAdminClient;
 import com.datastrato.gravitino.client.HTTPClient;
+import com.datastrato.gravitino.client.KerberosTokenProvider;
 import com.datastrato.gravitino.client.RESTClient;
 import com.datastrato.gravitino.dto.responses.VersionResponse;
 import com.datastrato.gravitino.exceptions.RESTException;
@@ -31,6 +33,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
@@ -52,6 +55,7 @@ public class MiniGravitino {
   private final File mockConfDir;
   private final ServerConfig serverConfig = new ServerConfig();
   private final ExecutorService executor = Executors.newSingleThreadExecutor();
+  private Properties properties;
 
   private String host;
 
@@ -84,7 +88,7 @@ public void start() throws Exception {
         Paths.get(ITUtils.joinPath(gravitinoRootDir, "conf", "gravitino-env.sh.template")),
         Paths.get(ITUtils.joinPath(mockConfDir.getAbsolutePath(), "gravitino-env.sh")));
 
-    Properties properties =
+    properties =
         serverConfig.loadPropertiesFromFile(
             new File(ITUtils.joinPath(mockConfDir.getAbsolutePath(), "gravitino.conf")));
 
@@ -245,8 +249,32 @@ private boolean checkIfServerIsRunning() {
     String URI = String.format("http://%s:%d", host, port);
     LOG.info("checkIfServerIsRunning() URI: {}", URI);
 
+    // Use kerberos, we need to use sdk to check
+    if (Objects.equals(serverConfig.get(Configs.AUTHENTICATOR), "kerberos")) {
+      try {
+        Thread.sleep(5000);
+        KerberosTokenProvider provider =
+            KerberosTokenProvider.builder()
+                .withClientPrincipal((String) properties.get("client.kerberos.principal"))
+                .withKeyTabFile(new File((String) properties.get("client.kerberos.keytab")))
+                .build();
+        GravitinoAdminClient adminClient =
+            GravitinoAdminClient.builder(URI).withKerberosAuth(provider).build();
+
+        adminClient.listMetalakes();
+        return true;
+      } catch (Exception e) {
+        LOG.warn(
+            "Kerberos checkIfServerIsRunning() fails, GravitinoServer is not running {}",
+            e.getMessage());
+        return true;
+      }
+    }
+
+    // Not auth, we can use the rest client to check
     VersionResponse response = null;
     try {
+      Thread.sleep(5000);
       response =
           restClient.get(
               "api/version",
@@ -256,7 +284,10 @@ private boolean checkIfServerIsRunning() {
     } catch (RESTException e) {
       LOG.warn("checkIfServerIsRunning() fails, GravitinoServer is not running {}", e.getMessage());
       return false;
+    } catch (Exception e1) {
+      return false;
     }
+
     if (response != null && response.getCode() == 0) {
       return true;
     } else {